← Files GH Review LoopARCHIVED FILE
skills/gh-review-loop/scripts/fetch_gemini_threads.py
166 KB · Oct 4, 2026 · 12:34 UTC
#!/usr/bin/env python3
"""Fetch thread-aware AI reviewer comments for a GitHub PR."""
from __future__ import annotations
import argparse
import dataclasses
import datetime as _dt
import hashlib
import json
import re
import subprocess
import sys
import time
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any
# Make the script's own directory importable so `from judge import ...` works
# when this script is invoked directly (e.g. `python3 .../fetch_gemini_threads.py`).
# Under `/plugin install` both files live in the same scripts/ directory.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import metrics # noqa: E402 — sibling module, pure/stdlib-only
import cluster_findings # noqa: E402 — sibling module, pure/stdlib-only
import reviewer_resolver # noqa: E402 — sibling module, pure/stdlib-only
import review_vendors # noqa: E402 — sibling module, pure/stdlib-only
from loop_color import color_loop, colors_enabled # noqa: E402 — sibling module
from loop_state import ( # noqa: E402 — sibling module, stdlib-only
any_active_run,
clear_sentinel,
find_active_run, # noqa: F401 — re-exported; hooks/tests import via this module
load_sticky_state,
migrate_legacy_state_dir,
resolve_current_repo,
save_sticky_state,
sentinel_path,
state_dir,
sticky_state_path, # noqa: F401 — re-exported; hooks/tests import via this module
summary_is_stale, # noqa: F401 — re-exported; hooks/tests import via this module
touch_sentinel,
)
from loop_state import safe_int as _safe_int # noqa: E402
# There is no default reviewer. An unconfigured PR resolves to "none configured"
# and the loop asks; see resolve_reviewer_selection(). These name the reviewer
# the loop may *offer*, which is a ping-first vendor so that accepting the offer
# starts a real cycle instead of a wait.
SUGGESTED_PROVIDER_NAME = review_vendors.SUGGESTED_VENDOR.display_name
SUGGESTED_AUTHOR = review_vendors.SUGGESTED_VENDOR.login
SUGGESTED_REVIEW_TRIGGER_MENTION = review_vendors.SUGGESTED_VENDOR.mention
# Back-compat aliases: --author's argparse default and the re-review trigger
# regex still need a concrete value to fall back on.
DEFAULT_PROVIDER_NAME = SUGGESTED_PROVIDER_NAME
DEFAULT_AUTHOR = SUGGESTED_AUTHOR
DEFAULT_REVIEW_TRIGGER_MENTION = SUGGESTED_REVIEW_TRIGGER_MENTION
DEFAULT_REREVIEW_LIMIT = 3
def _review_trigger_re(mention: Any) -> re.Pattern[str]:
if not isinstance(mention, str):
mention = DEFAULT_REVIEW_TRIGGER_MENTION
mention = mention.strip() or DEFAULT_REVIEW_TRIGGER_MENTION
if not mention.startswith("@"):
mention = f"@{mention}"
return re.compile(
rf"{re.escape(mention)}(?![\w-]).*\breview\b",
re.IGNORECASE | re.DOTALL,
)
REREVIEW_TRIGGER_RE = _review_trigger_re(DEFAULT_REVIEW_TRIGGER_MENTION)
# Gemini prefixes inline review comments with a priority image whose alt text
# is the severity. Example: 
SEVERITY_RE = re.compile(r"!\[(critical|high|medium|low)\]", re.IGNORECASE)
# Codex marks findings with a shields.io priority badge whose alt text is the
# priority. Example: 
CODEX_PRIORITY_RE = re.compile(r"!\[(P[0-3])(?:\s+Badge)?\]", re.IGNORECASE)
CODEX_PRIORITY_TO_SEVERITY = {
"p0": "critical",
"p1": "high",
"p2": "medium",
"p3": "low",
}
SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unknown": 4}
# Page sizes embedded in QUERY below. Used by the pagination guard to detect
# when we may be silently dropping data.
PAGE_LIMIT_REVIEW_THREADS = 100
PAGE_LIMIT_REVIEWS = 100
PAGE_LIMIT_PR_COMMENTS = 100
PAGE_LIMIT_THREAD_COMMENTS = 50
# Minimum reply length to count a non-bot comment as a substantive reply.
# Filters out token acks like "ok", "ack", "thanks", "👍".
ADDRESSED_BY_REPLY_MIN_CHARS = 30
@dataclass(frozen=True)
class PullRequest:
owner: str
repo: str
number: int
url: str | None = None
GH_NOT_FOUND_MESSAGE = (
"GitHub CLI (`gh`) not found on PATH. This skill needs `gh` installed and "
"authenticated (`gh auth login`); it does not fall back to raw API calls."
)
def run_gh(args: list[str], cwd: str | None = None) -> Any:
try:
proc = subprocess.run(
["gh", *args],
cwd=cwd,
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
except FileNotFoundError as exc:
# A missing binary must surface as the same clean `error: ...` line the
# rest of the script prints, not a traceback -- reviewers running the
# skill in a sandbox without gh should get one actionable sentence.
# subprocess raises the same exception for a missing cwd, so only
# attribute it to gh when the OS names gh as the missing file.
missing = str(exc.filename or "")
if missing == "gh" or missing.endswith("/gh"):
raise RuntimeError(GH_NOT_FOUND_MESSAGE) from exc
raise RuntimeError(f"gh {' '.join(args)} failed: {exc}") from exc
if proc.returncode != 0:
message = proc.stderr.strip() or proc.stdout.strip()
raise RuntimeError(f"gh {' '.join(args)} failed: {message}")
output = proc.stdout.strip()
if not output:
return None
try:
return json.loads(output)
except json.JSONDecodeError:
return output
def parse_pr_url(value: str) -> PullRequest | None:
match = re.match(r"https://github\.com/([^/]+)/([^/]+)/pull/(\d+)(?:[/?#].*)?$", value)
if not match:
return None
owner, repo, number = match.groups()
return PullRequest(owner=owner, repo=repo, number=int(number), url=value)
def resolve_current_pr() -> PullRequest:
run_gh(["auth", "status"])
view = run_gh(["pr", "view", "--json", "number,url"])
if not isinstance(view, dict) or "url" not in view or "number" not in view:
raise RuntimeError("Could not resolve the current branch PR with gh pr view.")
parsed = parse_pr_url(view["url"])
if not parsed:
raise RuntimeError(f"Could not parse PR URL: {view['url']}")
return PullRequest(parsed.owner, parsed.repo, int(view["number"]), view["url"])
def resolve_pr(value: str | None) -> PullRequest:
if not value:
return resolve_current_pr()
parsed = parse_pr_url(value)
if parsed:
return parsed
shorthand = re.match(r"([^/\s]+)/([^#\s]+)#(\d+)$", value)
if shorthand:
owner, repo, number = shorthand.groups()
return PullRequest(owner=owner, repo=repo, number=int(number))
raise RuntimeError("Use a PR URL like https://github.com/OWNER/REPO/pull/123 or OWNER/REPO#123.")
def select_stats_records(
records: list[dict[str, Any]], *, repo: str, window: int, all_repos: bool
) -> list[dict[str, Any]]:
if not all_repos:
repo_key = repo.casefold()
records = [
record
for record in records
if isinstance(record.get("repo"), str)
and record["repo"].casefold() == repo_key
]
return records[-window:] if window > 0 else records
QUERY = """
query($owner:String!, $repo:String!, $number:Int!) {
repository(owner:$owner, name:$repo) {
pullRequest(number:$number) {
number
url
title
headRefOid
baseRepository { nameWithOwner }
headRepository { nameWithOwner }
comments(last:100) {
nodes {
id
author { login __typename }
body
createdAt
url
}
}
reviews(last:100) {
nodes {
id
author { login __typename }
body
state
submittedAt
url
}
}
reviewThreads(first:100) {
nodes {
id
isResolved
isOutdated
path
line
startLine
originalLine
originalStartLine
diffSide
comments(first:50) {
nodes {
id
author { login __typename }
body
createdAt
updatedAt
url
path
line
startLine
originalLine
originalStartLine
diffHunk
}
}
}
}
}
}
}
"""
REVIEWER_DISCOVERY_QUERY = """
query($owner:String!, $repo:String!, $number:Int!, $threadsAfter:String) {
repository(owner:$owner, name:$repo) {
pullRequest(number:$number) {
number
url
reviewThreads(first:100, after:$threadsAfter) {
pageInfo {
hasNextPage
endCursor
}
nodes {
id
comments(first:100) {
pageInfo {
hasNextPage
endCursor
}
nodes {
author { login __typename }
}
}
}
}
}
}
}
"""
THREAD_COMMENTS_DISCOVERY_QUERY = """
query($threadId:ID!, $commentsAfter:String) {
node(id:$threadId) {
... on PullRequestReviewThread {
comments(first:100, after:$commentsAfter) {
pageInfo {
hasNextPage
endCursor
}
nodes {
author { login __typename }
}
}
}
}
}
"""
RESOLVE_THREAD_MUTATION = """
mutation($threadId:ID!) {
resolveReviewThread(input:{threadId:$threadId}) {
thread {
id
isResolved
}
}
}
"""
def fetch_threads(pr: PullRequest) -> dict[str, Any]:
result = run_gh(
[
"api",
"graphql",
"-f",
f"query={QUERY}",
"-F",
f"owner={pr.owner}",
"-F",
f"repo={pr.repo}",
"-F",
f"number={pr.number}",
]
)
if not isinstance(result, dict):
raise RuntimeError("Unexpected gh GraphQL response.")
try:
return result["data"]["repository"]["pullRequest"]
except KeyError as exc:
raise RuntimeError(f"Unexpected gh GraphQL shape: missing {exc}") from exc
def fetch_remaining_discovery_comments(
thread_id: str,
*,
after: str | None,
max_pages: int,
) -> dict[str, Any]:
nodes: list[dict[str, Any]] = []
warnings: list[str] = []
partial = False
cursor = after
for _ in range(max_pages):
gh_args = [
"api",
"graphql",
"-f",
f"query={THREAD_COMMENTS_DISCOVERY_QUERY}",
"-F",
f"threadId={thread_id}",
]
if cursor:
gh_args.extend(["-F", f"commentsAfter={cursor}"])
result = run_gh(gh_args)
if not isinstance(result, dict):
raise RuntimeError("Unexpected gh GraphQL response.")
try:
comments = result["data"]["node"]["comments"]
except KeyError as exc:
raise RuntimeError(f"Unexpected gh GraphQL shape: missing {exc}") from exc
page_nodes = comments.get("nodes") or []
nodes.extend(comment for comment in page_nodes if isinstance(comment, dict))
page_info = comments.get("pageInfo") or {}
if not page_info.get("hasNextPage"):
return {"nodes": nodes, "partial": partial, "warnings": warnings}
cursor = page_info.get("endCursor")
if not cursor:
partial = True
warnings.append(
f"thread {thread_id} comments pagination stopped without an end cursor"
)
break
else:
partial = True
warnings.append(f"thread {thread_id} comments hit discovery page cap")
return {"nodes": nodes, "partial": partial, "warnings": warnings}
def fetch_reviewer_discovery(pr: PullRequest, *, max_pages: int = 20) -> dict[str, Any]:
"""Fetch review-thread comment authors for reviewer discovery.
Normal loop fetches intentionally stay single-page for compatibility. This
discovery-only path paginates review threads so a high-activity PR does not
falsely appear to have no AI reviewer candidates.
"""
all_threads: list[dict[str, Any]] = []
warnings: list[str] = []
after: str | None = None
partial = False
for _ in range(max_pages):
gh_args = [
"api",
"graphql",
"-f",
f"query={REVIEWER_DISCOVERY_QUERY}",
"-F",
f"owner={pr.owner}",
"-F",
f"repo={pr.repo}",
"-F",
f"number={pr.number}",
]
if after:
gh_args.extend(["-F", f"threadsAfter={after}"])
result = run_gh(gh_args)
if not isinstance(result, dict):
raise RuntimeError("Unexpected gh GraphQL response.")
if "errors" in result:
raise RuntimeError(f"gh GraphQL errors: {result['errors']}")
try:
pull_request = result["data"]["repository"]["pullRequest"]
if not isinstance(pull_request, dict):
raise TypeError("pullRequest is not a dictionary")
review_threads = pull_request.get("reviewThreads") or {}
if not isinstance(review_threads, dict):
raise TypeError("reviewThreads is not a dictionary")
except (KeyError, TypeError) as exc:
raise RuntimeError(
f"Unexpected gh GraphQL shape: missing or invalid data ({exc})"
) from exc
nodes = review_threads.get("nodes") or []
for thread in nodes:
if not isinstance(thread, dict):
continue
comments = thread.get("comments") or {}
page_info = comments.get("pageInfo") or {}
comment_nodes = [
comment
for comment in (comments.get("nodes") or [])
if isinstance(comment, dict)
]
if page_info.get("hasNextPage"):
thread_id = thread.get("id")
if isinstance(thread_id, str) and thread_id:
remaining = fetch_remaining_discovery_comments(
thread_id,
after=page_info.get("endCursor"),
max_pages=max_pages,
)
comment_nodes.extend(remaining["nodes"])
partial = partial or bool(remaining["partial"])
warnings.extend(remaining["warnings"])
else:
partial = True
warnings.append("(unknown) thread comments pagination missing thread id")
merged_thread = dict(thread)
merged_thread["comments"] = {"nodes": comment_nodes}
all_threads.append(merged_thread)
page_info = review_threads.get("pageInfo") or {}
if not page_info.get("hasNextPage"):
merged = dict(pull_request)
merged["reviewThreads"] = {"nodes": all_threads}
return {"pull_request": merged, "partial": partial, "warnings": warnings}
after = page_info.get("endCursor")
if not after:
partial = True
warnings.append("reviewer discovery pagination stopped without an end cursor")
break
else:
partial = True
warnings.append("reviewer discovery hit page cap")
return {
"pull_request": {
"number": pr.number,
"url": pr.url,
"reviewThreads": {"nodes": all_threads},
},
"partial": partial,
"warnings": warnings,
}
def resolve_thread(thread_id: str, *, dry_run: bool = False, label: str = "thread") -> None:
if dry_run:
print(f"[dry-run] would resolve {label} {thread_id}", file=sys.stderr)
return
run_gh(
[
"api",
"graphql",
"-f",
f"query={RESOLVE_THREAD_MUTATION}",
"-F",
f"threadId={thread_id}",
]
)
def is_addressed_by_reply(thread: dict[str, Any], bot_author: str) -> bool:
"""An unresolved thread where a non-bot human posted a substantive reply.
The reply is treated as a deliberate deferral (defer/wontfix/explanation).
The loop should not retry fixing the same thread cycle after cycle.
"""
if thread.get("isResolved") or thread.get("isOutdated"):
return False
for comment in (thread.get("comments") or {}).get("nodes") or []:
login = (comment.get("author") or {}).get("login") or ""
if not login or login == bot_author or login.endswith("[bot]"):
continue
body = (comment.get("body") or "").strip()
if len(body) >= ADDRESSED_BY_REPLY_MIN_CHARS:
return True
return False
def filter_threads(
pull_request: dict[str, Any],
author: str,
include_resolved: bool,
include_outdated: bool,
include_addressed_by_reply: bool = False,
) -> list[dict[str, Any]]:
selected: list[dict[str, Any]] = []
for thread in pull_request.get("reviewThreads", {}).get("nodes", []):
if thread.get("isResolved") and not include_resolved:
continue
if thread.get("isOutdated") and not include_outdated:
continue
if not include_addressed_by_reply and is_addressed_by_reply(thread, author):
continue
comments = thread.get("comments", {}).get("nodes", [])
matching_comments = [
comment
for comment in comments
if (comment.get("author") or {}).get("login") == author
]
if not matching_comments:
continue
copied = dict(thread)
copied["comments"] = matching_comments
selected.append(copied)
return selected
def outdated_unresolved_threads(pull_request: dict[str, Any], author: str) -> list[dict[str, Any]]:
threads: list[dict[str, Any]] = []
for thread in pull_request.get("reviewThreads", {}).get("nodes", []):
if thread.get("isResolved") or not thread.get("isOutdated"):
continue
comments = thread.get("comments", {}).get("nodes", [])
if any((comment.get("author") or {}).get("login") == author for comment in comments):
threads.append(thread)
return threads
def addressed_by_reply_threads(pull_request: dict[str, Any], author: str) -> list[dict[str, Any]]:
threads: list[dict[str, Any]] = []
for thread in pull_request.get("reviewThreads", {}).get("nodes", []):
if not is_addressed_by_reply(thread, author):
continue
comments = thread.get("comments", {}).get("nodes", [])
if any((comment.get("author") or {}).get("login") == author for comment in comments):
threads.append(thread)
return threads
def resolve_outdated_threads(
pull_request: dict[str, Any], author: str, *, dry_run: bool = False
) -> int:
threads = outdated_unresolved_threads(pull_request, author)
for thread in threads:
resolve_thread(thread["id"], dry_run=dry_run, label="outdated")
return len(threads)
def resolve_addressed_by_reply(
pull_request: dict[str, Any], author: str, *, dry_run: bool = False
) -> int:
threads = addressed_by_reply_threads(pull_request, author)
for thread in threads:
resolve_thread(thread["id"], dry_run=dry_run, label="addressed-by-reply")
return len(threads)
def _iter_comments(thread: dict[str, Any]) -> list[dict[str, Any]]:
"""Return thread comments regardless of pre/post-filter shape."""
comments = thread.get("comments")
if isinstance(comments, list):
return comments
if isinstance(comments, dict):
return comments.get("nodes", []) or []
return []
def thread_severity(thread: dict[str, Any]) -> str:
"""Return reviewer-assigned severity (critical/high/medium/low) or 'unknown'.
Gemini marks severity with a priority image; Codex uses a P0-P3 badge.
Both are unambiguous, so both are read regardless of selected reviewer.
"""
for comment in _iter_comments(thread):
body = comment.get("body") or ""
match = SEVERITY_RE.search(body)
if match:
return match.group(1).lower()
codex_match = CODEX_PRIORITY_RE.search(body)
if codex_match:
return CODEX_PRIORITY_TO_SEVERITY[codex_match.group(1).lower()]
return "unknown"
def sort_by_severity(threads: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Sort threads by severity, highest first; stable for equal severities."""
return sorted(
threads,
key=lambda t: SEVERITY_ORDER.get(thread_severity(t), SEVERITY_ORDER["unknown"]),
)
def filter_by_min_severity(
threads: list[dict[str, Any]], min_severity: str | None, *, keep_unknown: bool = True
) -> list[dict[str, Any]]:
"""Drop threads with severity strictly lower than min_severity.
If `min_severity` is None, no minimum-severity threshold is enforced — the
function then acts only on the `keep_unknown` axis, which lets callers use
`--drop-unknown-severity` independently of `--min-severity`.
`unknown` threads (no Gemini priority marker) are kept by default — dropping
them on a strict filter would silently swallow every thread Gemini didn't
annotate, which is the wrong default for a fast-feedback loop. Pass
keep_unknown=False to drop them anyway.
"""
cap = SEVERITY_ORDER[min_severity] if min_severity else None
out: list[dict[str, Any]] = []
for t in threads:
sev = thread_severity(t)
if sev == "unknown":
if keep_unknown:
out.append(t)
continue
if cap is None or SEVERITY_ORDER[sev] <= cap:
out.append(t)
return out
def severity_counts(threads: list[dict[str, Any]]) -> dict[str, int]:
counts: dict[str, int] = {}
for thread in threads:
sev = thread_severity(thread)
counts[sev] = counts.get(sev, 0) + 1
return counts
def derive_record_fields(
*,
baseline_ids: set[str],
current_actionable_ids: set[str],
addressed_by_reply_ids: set[str],
outcome: str,
judge_ran: bool,
judge_results: dict[str, dict[str, Any]],
) -> dict[str, int]:
"""Compute the script-derived metric counts for a run record."""
findings_fetched = len(baseline_ids | current_actionable_ids)
observed_fixed_count = len(
baseline_ids - current_actionable_ids - addressed_by_reply_ids
)
remaining_actionable = len(current_actionable_ids)
addressed_by_reply = len(addressed_by_reply_ids)
if judge_ran:
needs_human = sum(
1
for tid, r in judge_results.items()
if tid in current_actionable_ids and r.get("verdict") == "needs_human"
)
elif outcome == "human":
needs_human = remaining_actionable
else:
needs_human = 0
return {
"findings_fetched": findings_fetched,
"observed_fixed_count": observed_fixed_count,
"remaining_actionable": remaining_actionable,
"addressed_by_reply": addressed_by_reply,
"needs_human": needs_human,
}
def _derive_outcome(
remaining_actionable: int,
verification: str,
cap_reached: bool,
*,
gemini_confirmed: bool = True,
likely_fixed_remaining: int = 0,
) -> str:
"""Best-effort outcome when the agent does not pass --outcome explicitly.
Produces the outcomes inferable from script-visible state plus the two
agent-supplied signals that disambiguate the terminal cycle:
- ``gemini_confirmed`` — the final re-review actually responded after the
re-review request (False when the final wait timed out). Defaults True so
legacy callers are unaffected; ``clean`` is only ever returned when True.
- ``likely_fixed_remaining`` — count of still-open threads that are
multi-signal "likely fixed" (see metrics.classify_finding_state).
Logic (corrections folded in):
- verification failed -> verification_failed
- unconfirmed final wait + everything resolved/likely-fixed
-> fixed_pending_confirmation
- cap reached (genuine unfixed remaining) -> capped
- no remaining + passed + gemini_confirmed -> clean
- else -> human
``regression`` / ``no_progress`` encode an agent judgment the script cannot
see, so the agent must pass them via --outcome.
"""
if verification == "failed":
return "verification_failed"
# Everything still open is multi-signal "likely fixed".
likely_all_fixed = (
remaining_actionable > 0 and likely_fixed_remaining >= remaining_actionable
)
# No open threads and the suite passed — it *looks* resolved.
looks_resolved = remaining_actionable == 0 and verification == "passed"
# Final wait timed out (Gemini never re-confirmed) but state otherwise looks
# fixed: never guess clean or capped — report pending confirmation.
if not gemini_confirmed and (likely_all_fixed or looks_resolved):
return "fixed_pending_confirmation"
if looks_resolved and gemini_confirmed:
return "clean"
if cap_reached:
return "capped"
return "human"
def pagination_warnings(pull_request: dict[str, Any]) -> list[str]:
"""Return human-readable warnings when any GraphQL page hit its limit.
Hitting the limit means the script may be silently dropping data — the
loop should surface this so the user knows to paginate or scope the PR.
"""
warnings: list[str] = []
pr_comments = (pull_request.get("comments") or {}).get("nodes") or []
reviews = (pull_request.get("reviews") or {}).get("nodes") or []
threads = (pull_request.get("reviewThreads") or {}).get("nodes") or []
if len(pr_comments) >= PAGE_LIMIT_PR_COMMENTS:
warnings.append(
f"PR comments hit page limit ({PAGE_LIMIT_PR_COMMENTS}); older comments may be missing."
)
if len(reviews) >= PAGE_LIMIT_REVIEWS:
warnings.append(
f"PR reviews hit page limit ({PAGE_LIMIT_REVIEWS}); older reviews may be missing."
)
if len(threads) >= PAGE_LIMIT_REVIEW_THREADS:
warnings.append(
f"reviewThreads hit page limit ({PAGE_LIMIT_REVIEW_THREADS}); newer threads may be missing."
)
for thread in threads:
comments = (thread.get("comments") or {}).get("nodes") or []
if len(comments) >= PAGE_LIMIT_THREAD_COMMENTS:
thread_id = thread.get("id") or "(unknown)"
warnings.append(
f"thread {thread_id} comments hit page limit "
f"({PAGE_LIMIT_THREAD_COMMENTS}); newer replies may be missing."
)
return warnings
def filter_reviews(pull_request: dict[str, Any], author: str) -> list[dict[str, Any]]:
return [
review
for review in pull_request.get("reviews", {}).get("nodes", [])
if (review.get("author") or {}).get("login") == author
]
# A reviewer can decline outright — quota exhausted, service withdrawn. That is
# a top-level PR comment, not a review, so the activity fingerprint never sees
# it and the wait would otherwise burn its whole timeout on a reviewer that
# already answered. Patterns are deliberately narrow so a review that merely
# discusses rate limiting is not mistaken for a refusal.
#
# The two kinds differ in what the user can do about it: a quota cap is
# recoverable (upgrade the account or add credits, then retry the same wait),
# a withdrawn service is not. Only the recoverable kind is worth interrupting
# the user for.
REFUSAL_QUOTA = "quota_exhausted"
REFUSAL_WITHDRAWN = "withdrawn"
REVIEWER_REFUSAL_RES = (
(REFUSAL_QUOTA, re.compile(r"reached your\b.{0,60}\blimits?\b", re.IGNORECASE | re.DOTALL)),
(REFUSAL_QUOTA, re.compile(r"usage limits?\b.{0,40}\bcode review", re.IGNORECASE | re.DOTALL)),
(REFUSAL_WITHDRAWN, re.compile(r"\bhas been sunset\b", re.IGNORECASE)),
(
REFUSAL_WITHDRAWN,
re.compile(r"review\w*\b.{0,40}\bno longer (available|supported)\b", re.IGNORECASE | re.DOTALL),
),
)
def print_reviewer_refusal(
refusal: dict[str, Any],
*,
author: str,
json_output: bool,
color_enabled: bool,
) -> None:
"""Emit the terminal stop block for a reviewer that declined to review."""
if json_output:
fields = {k: v for k, v in refusal.items() if k != "pull_request" and v is not None}
print(json.dumps({"wait": fields}, indent=2, sort_keys=True))
return
quota = refusal.get("kind") == REFUSAL_QUOTA
lines = [
f"[loop] STOP — {author} refused the review: {refusal.get('reason', '')}",
"Waiting cannot help; the reviewer already answered.",
]
if refusal.get("url"):
lines.append(f"Comment: {refusal['url']}")
if quota:
# Recoverable: the cap lifts the moment the user pays for it, so the
# decision is theirs and it has to be asked now — not after a timeout.
lines.extend(
[
"The cap is a billing limit, not a verdict — ask the user NOW how to proceed:",
" 1. Stop the loop — record with: --record-run --outcome human "
"--outcome-reason 'reviewer quota exhausted' --gemini-unconfirmed",
f" 2. Upgrade the {reviewer_resolver.display_name_for(author) or author} "
"account or add credits, then retry — "
"once the user confirms, re-run this same wait command.",
"Do not wait, do not re-ping, and do not burn a cycle re-requesting the review.",
]
)
else:
lines.append(
"Record the run with: --record-run --outcome human "
"--outcome-reason 'reviewer refused the review' --gemini-unconfirmed"
)
print(color_loop_block("\n".join(lines), enabled=color_enabled))
class ReviewerRefused(Exception):
"""The reviewer declined to review; waiting any longer cannot help."""
def __init__(self, refusal: dict[str, Any]) -> None:
super().__init__(refusal.get("reason", "reviewer refused"))
self.refusal = refusal
class WaitTimedOut(Exception):
"""A blocking ``--wait`` exhausted its ``--timeout`` budget.
Typed so the CLI can print the deterministic timed-out heartbeat instead
of a traceback — the blocking wait is designed to run as a background
task, and its captured output must end in a relayable line, not a stack.
A bare RuntimeError cannot be used: ``run_gh`` failures raise that too,
and a network error must not masquerade as a timeout.
"""
def __init__(self, elapsed_seconds: int) -> None:
super().__init__(
f"Timed out waiting for stable review activity after {elapsed_seconds} seconds."
)
self.elapsed_seconds = elapsed_seconds
def reviewer_refusal(
pull_request: dict[str, Any],
author: str,
after_iso: str | None = None,
) -> dict[str, Any] | None:
"""Return the reviewer's refusal to review, if it posted one.
Only the reviewer's own top-level comments count, and only those newer than
``after_iso`` — a refusal from a previous cycle must not end this wait.
"""
for comment in (pull_request.get("comments") or {}).get("nodes") or []:
if (comment.get("author") or {}).get("login") != author:
continue
created_at = comment.get("createdAt") or ""
if after_iso and created_at <= after_iso:
continue
body = comment.get("body") or ""
kind = next((label for label, pattern in REVIEWER_REFUSAL_RES if pattern.search(body)), None)
if kind is None:
continue
return {
"kind": kind,
"reason": " ".join(body.replace(">", " ").split())[:200],
"created_at": created_at,
"url": comment.get("url"),
}
return None
BLOB_ANCHOR_RE = re.compile(
r"https://github\.com/[^/\s]+/[^/\s]+/blob/[0-9a-f]+/(\S+?)#L(\d+)",
re.IGNORECASE,
)
def review_body_findings(
pull_request: dict[str, Any], author: str
) -> list[dict[str, Any]]:
"""Surface priority-badged findings that live in the review body.
Codex usually publishes inline review comments, but it can put a finding
directly in the review body, where no thread exists to fetch. Only the
newest review counts: a later re-review supersedes older body findings, so
these synthetic entries go stale the same way threads do.
"""
reviews = filter_reviews(pull_request, author)
if not reviews:
return []
latest = max(reviews, key=lambda review: review.get("submittedAt") or "")
body = latest.get("body") or ""
badges = list(CODEX_PRIORITY_RE.finditer(body))
if not badges:
return []
review_id = latest.get("id") or hashlib.sha1(body.encode()).hexdigest()[:16]
findings = []
for index, badge in enumerate(badges):
start = badges[index - 1].end() if index else 0
end = badges[index + 1].start() if index + 1 < len(badges) else len(body)
# The file anchor Codex prints just above a badge locates the finding.
anchors = list(BLOB_ANCHOR_RE.finditer(body, start, badge.start()))
anchor = anchors[-1] if anchors else None
path = anchor.group(1) if anchor else "(review body)"
line = int(anchor.group(2)) if anchor else None
findings.append({
"id": f"review-body:{review_id}:{index + 1}",
"isResolved": False,
"isOutdated": False,
"isReviewBodyFinding": True,
"path": path,
"line": line,
"startLine": None,
"originalLine": line,
"originalStartLine": None,
"comments": [{
"id": f"{review_id}:{index + 1}",
"author": {"login": author},
"body": body[badge.start():end].strip(),
"createdAt": latest.get("submittedAt"),
"updatedAt": latest.get("submittedAt"),
"url": latest.get("url"),
"path": path,
"line": line,
"startLine": None,
"originalLine": line,
"originalStartLine": None,
"diffHunk": "",
}],
})
return findings
def rereview_requests(
pull_request: dict[str, Any],
agent_login: str | None = None,
*,
review_trigger_mention: str | None = DEFAULT_REVIEW_TRIGGER_MENTION,
) -> list[dict[str, Any]]:
"""Count comments that ping the configured reviewer for a re-review.
When agent_login is provided, only comments authored by that login count.
This prevents arbitrary humans pinging Gemini from consuming the loop cap.
"""
if not review_trigger_mention:
return []
out = []
trigger_re = _review_trigger_re(review_trigger_mention)
for comment in pull_request.get("comments", {}).get("nodes", []):
if not trigger_re.search(comment.get("body") or ""):
continue
if agent_login is not None:
login = (comment.get("author") or {}).get("login") or ""
if login != agent_login:
continue
out.append(comment)
return out
def gh_authenticated_login() -> str | None:
"""Return the gh-authenticated user login, or None if it can't be resolved."""
try:
result = run_gh(["api", "user", "--jq", ".login"])
except RuntimeError:
return None
if isinstance(result, str) and result.strip():
return result.strip()
return None
def nonnegative_int(value: str) -> int:
try:
parsed = int(value)
except (TypeError, ValueError) as err:
raise argparse.ArgumentTypeError(
f"invalid int value: {value!r}"
) from err
if parsed < 0:
raise argparse.ArgumentTypeError("must be >= 0")
return parsed
def effective_rereview_limit(cli_value: int | None, prefs: dict[str, Any]) -> int:
if cli_value is not None:
return cli_value
value = prefs.get("max_rereview_requests", DEFAULT_REREVIEW_LIMIT)
if isinstance(value, bool):
return DEFAULT_REREVIEW_LIMIT
if isinstance(value, int) and value >= 0:
return value
if isinstance(value, str):
stripped = value.strip()
if stripped.isdigit():
return int(stripped)
return DEFAULT_REREVIEW_LIMIT
def _direct_preferences_path() -> Path:
"""Mirror ``judge.prefs_path()`` for the fallback path when ``judge`` is unavailable."""
return state_dir() / "preferences.json"
# Defaults that match the canonical ``judge.load_preferences()`` contract. The
# fallback path layers loaded values on top of this so downstream callers can
# read any documented key (``judge_mode``, ``judge_model``, etc.) without
# guarding against KeyError when the optional ``judge`` module is missing.
_FALLBACK_PREFS_DEFAULTS: dict[str, Any] = {
"schema_version": 2,
"judge_mode": "off",
"judge_model": "gpt-4o-mini",
"judge_tip_shown": False,
"max_rereview_requests": DEFAULT_REREVIEW_LIMIT,
"profiles": {},
"set_at": "",
}
def load_preferences_with_fallback() -> dict[str, Any]:
"""Load user preferences, falling back to a direct JSON read.
Primary path: ``judge.load_preferences()`` (canonical loader with full
schema validation). If the optional ``judge`` module cannot be imported
(minimal install, broken install, vendored layout), read
``preferences.json`` directly so persistent settings like
``max_rereview_requests`` still take effect. Any failure is surfaced to
stderr with a hint rather than silently dropped.
The returned dict is always populated with the documented preference keys
(defaults from ``_FALLBACK_PREFS_DEFAULTS`` overlaid by saved values), so
callers can read keys like ``judge_model`` without KeyError handling.
"""
try:
from judge import load_preferences # noqa: PLC0415
except ImportError:
print(
"warning: optional 'judge' module not importable; reading "
"preferences.json directly. Reinstall gh-review-loop "
"to restore full judge-eval support.",
file=sys.stderr,
)
else:
try:
return load_preferences()
except Exception as err:
print(
f"warning: judge.load_preferences() failed ({err!s}); "
"falling back to direct preferences.json load.",
file=sys.stderr,
)
path = _direct_preferences_path()
if not path.exists():
prefs = dict(_FALLBACK_PREFS_DEFAULTS)
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(prefs, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
except OSError:
pass
return prefs
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as err:
print(
f"warning: could not read {path} ({err!s}); ignoring saved preferences.",
file=sys.stderr,
)
return dict(_FALLBACK_PREFS_DEFAULTS)
if not isinstance(data, dict):
print(
f"warning: {path} did not contain a JSON object; ignoring saved preferences.",
file=sys.stderr,
)
return dict(_FALLBACK_PREFS_DEFAULTS)
return {**_FALLBACK_PREFS_DEFAULTS, **data}
def post_pr_comment(pr: PullRequest, body: str, *, dry_run: bool = False) -> None:
"""Post a comment on the PR via gh."""
pr_ref = f"{pr.owner}/{pr.repo}#{pr.number}"
if dry_run:
print(f"[dry-run] would post receipt to {pr_ref}:\n{body}", file=sys.stderr)
return
proc = subprocess.run(
["gh", "pr", "comment", str(pr.number), "--repo", f"{pr.owner}/{pr.repo}", "--body", body],
check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
if proc.returncode != 0:
message = proc.stderr.strip() or proc.stdout.strip()
raise RuntimeError(f"gh pr comment failed: {message}")
# ---------------------------------------------------------------------------
# Sticky receipt: a single PR comment that gets edited in place across
# multiple loop invocations on the same PR. Provides background visibility
# in the GitHub UI without spamming new comments per cycle.
# ---------------------------------------------------------------------------
# Marker embedded in the rendered body so a sticky receipt can be identified
# even if the local state file is wiped. Used as a fallback discovery key.
STICKY_RECEIPT_MARKER = "<!-- gh-review-loop:sticky-receipt -->"
# State-file primitives (sticky_state_path/load/save, active-run queries,
# the loop-active sentinel) live in loop_state so the hook gates can import
# them without paying this module's full import cost. Re-exported above.
def _state_key(pr: PullRequest) -> str:
return f"{pr.owner}/{pr.repo}#{pr.number}"
def read_reviewer_selection(pr: PullRequest) -> dict[str, Any] | None:
entry = load_sticky_state().get(_state_key(pr), {})
reviewer = entry.get("reviewer") if isinstance(entry, dict) else None
if not isinstance(reviewer, dict):
return None
login = reviewer.get("login")
if not isinstance(login, str) or not login:
return None
return dict(reviewer)
def save_reviewer_selection(pr: PullRequest, reviewer: dict[str, Any]) -> None:
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
entry["reviewer"] = dict(reviewer)
state[key] = entry
save_sticky_state(state)
def clear_reviewer_selection(pr: PullRequest) -> bool:
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
if not isinstance(entry, dict) or "reviewer" not in entry:
return False
del entry["reviewer"]
state[key] = entry
save_sticky_state(state)
return True
def read_render_baseline(pr: PullRequest, author: str) -> dict[str, str]:
"""thread_id → render_fp emitted on the previous cycle, per (pr, author).
The delta collapse (#95) compares against the *immediately previous* cycle,
never a seen-ever set — that is what makes an A→B→A thread render full on
the third cycle. Empty when never saved (cycle 1 → everything full).
"""
entry = load_sticky_state().get(_state_key(pr), {})
if not isinstance(entry, dict):
return {}
baselines = entry.get("render_baseline")
if not isinstance(baselines, dict):
return {}
fps = baselines.get(author)
if not isinstance(fps, dict):
return {}
return {k: v for k, v in fps.items() if isinstance(k, str) and isinstance(v, str)}
def save_render_baseline(pr: PullRequest, author: str, fps: dict[str, str]) -> None:
"""Replace the (pr, author) render baseline with this cycle's fingerprints.
Call ONLY after the rendered output was successfully emitted — persisting
before emission would collapse threads the agent never saw. Replacement
(not merge) keeps the baseline scoped to the previous cycle.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
baselines = entry.get("render_baseline")
baselines = dict(baselines) if isinstance(baselines, dict) else {}
baselines[author] = dict(fps)
entry["render_baseline"] = baselines
state[key] = entry
save_sticky_state(state)
def reviewer_selection_state(record: dict[str, Any], *, source: str) -> dict[str, Any]:
"""Project a persisted reviewer record into the loop's selection state.
Vendor facts are re-derived from the login rather than trusted from the
record, so selections persisted before a vendor was known (Codex records
written with ``review_trigger: null``) heal instead of permanently
disabling re-review requests for that PR.
"""
login = record["login"]
known_display_name = reviewer_resolver.display_name_for(login)
display_name = record.get("display_name") or known_display_name
if reviewer_resolver.trigger_for(login):
display_name = known_display_name
return {
"login": login,
"display_name": display_name,
"review_trigger": record.get("review_trigger") or reviewer_resolver.trigger_for(login),
"auto_reviews": reviewer_resolver.auto_reviews(login),
"source": source,
"confirmation_required": source in {"default_unconfirmed", "none_configured"},
# True whenever an actual choice backs the selection. A suggestion the
# user has not accepted is not a configuration.
"configured": source not in {"default_unconfirmed", "none_configured"},
"candidates_partial": False,
}
def resolve_reviewer_selection(args: argparse.Namespace, pr: PullRequest) -> dict[str, Any]:
login = args.reviewer or args.author
if login:
source = args.reviewer_source if args.reviewer else "explicit"
record = reviewer_resolver.make_reviewer_record(
login,
display_name=args.reviewer_name,
review_trigger=args.review_trigger_mention,
source=source,
)
save_reviewer_selection(pr, record)
return reviewer_selection_state(record, source=source)
persisted = read_reviewer_selection(pr)
if persisted:
return reviewer_selection_state(persisted, source="persisted")
# Nothing discovered, nothing persisted. Do not guess a vendor. Any default
# is wrong for someone: assuming a bot that reviews on its own turns "no
# reviewer here" into a full-timeout wait, and assuming a ping-first bot
# posts a mention on the user's PR for a reviewer they never chose. Report
# the state and let the caller ask.
#
# SUGGESTED_* is what the caller may *offer*, not what it may assume. It is
# a ping-first vendor precisely so that accepting the offer produces a real
# cycle 0 instead of a wait.
record = reviewer_resolver.make_reviewer_record(
SUGGESTED_AUTHOR,
display_name=SUGGESTED_PROVIDER_NAME,
review_trigger=SUGGESTED_REVIEW_TRIGGER_MENTION,
source="none_configured",
)
state = reviewer_selection_state(record, source="none_configured")
state["configured"] = False
state["unconfirmed"] = True
state["suggestion"] = {
"login": SUGGESTED_AUTHOR,
"display_name": SUGGESTED_PROVIDER_NAME,
"review_trigger": SUGGESTED_REVIEW_TRIGGER_MENTION,
"auto_reviews": reviewer_resolver.auto_reviews(SUGGESTED_AUTHOR),
}
state["message"] = (
"No reviewer bot has commented on this PR and none is configured. "
"Ask the user to pick one, or to ping "
f"{SUGGESTED_REVIEW_TRIGGER_MENTION} to start a review. Do not wait on "
"an unconfirmed reviewer."
)
return state
def update_run_tracking(
pr: PullRequest,
findings: list[tuple[str, str | None]],
*,
bump_session_cycle: bool = True,
) -> None:
"""Merge this invocation's findings into the run's tracking state.
``findings`` is a list of (thread_id, path) pairs. Sets ``started_at`` on
the first call of a run; unions ids/paths on every call. Stored under the
existing ``owner/repo#number`` key so it rides alongside sticky state.
``bump_session_cycle=False`` (the --cycle-summary path) keeps the
session-cycle ordinal frozen: a read-only summary re-run must not start a
new cycle.
The session-cycle ordinal advances on evidence that the previous cycle
actually *transitioned* (pushed and asked for a re-review), not merely
that it emitted a summary — see the rules below.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key, {})
run = entry.get("run", {})
if "started_at" not in run:
run["started_at"] = _now_iso()
# Monotonic counter bumped on every fetch. The loop's Stop-hook backstop
# compares it against last_summary_seq to tell whether the run advanced
# since the agent last emitted a summary (see summary_is_stale).
prev_seq = _safe_int(run.get("update_seq", 0))
run["update_seq"] = prev_seq + 1
# Script-owned session-cycle ordinal (#104): 1 on the run's first fetch,
# +1 only once the previous cycle has demonstrably ENDED. Narration copies
# this number; the model must never compute it — model-tracked counters
# drift across context compaction.
#
# A cycle ends when it emitted its summary AND transitioned through a new
# re-review request. The re-review count is the transition evidence: a
# session interrupted after `--cycle-summary` but before the push/re-review
# resumes with a `--full` recovery fetch, which is still *finishing the
# same cycle*. Treating the summary alone as the boundary mislabeled that
# fetch (and every later narration line and metric) as a new cycle (#111
# review). Summary staleness is deliberately NOT part of this test: a
# recovery fetch advances update_seq past last_summary_seq, which would
# then block the legitimate bump for the rest of the run.
# Transition evidence is LOCAL: `request_rereview.py` bumps
# `rereviews_posted` exactly once per request it actually posts. Deriving
# it from the PR's comments failed three different ways (#111 review) —
# `comments(last:100)` makes the count non-monotonic on busy PRs, the
# all-user fallback count leaks other people's pings when the agent login
# cannot be resolved, and PR history outlives `clear_run_tracking()`.
#
# The summary is CONSUMED when it advances a cycle. `last_summary_seq > 0`
# was a permanently-true flag once any summary had been emitted, so a
# second re-review arriving before the new cycle wrote its own summary
# advanced the ordinal twice off one stale summary (#111 review).
posted = _safe_int(run.get("rereviews_posted", 0))
posted_baseline = _safe_int(run.get("rereviews_posted_at_cycle_start", 0))
summary_seq = _safe_int(run.get("last_summary_seq", 0))
summary_consumed = _safe_int(run.get("summary_consumed_seq", 0))
if "session_cycle" not in run:
run["session_cycle"] = 1
run["rereviews_posted_at_cycle_start"] = posted
elif (
bump_session_cycle
and summary_seq > summary_consumed
and posted > posted_baseline
):
run["session_cycle"] = _safe_int(run.get("session_cycle")) + 1
run["rereviews_posted_at_cycle_start"] = posted
run["summary_consumed_seq"] = summary_seq
ids = set(run.get("finding_ids", []))
paths = set(run.get("finding_paths", []))
for thread_id, path in findings:
if thread_id:
ids.add(thread_id)
if path:
paths.add(path)
run["finding_ids"] = sorted(ids)
run["finding_paths"] = sorted(paths)
entry["run"] = run
state[key] = entry
save_sticky_state(state)
# Freshen the loop-active sentinel so the hooks.json shell guard lets the
# gates run for the duration of this loop (and the 24h TTL restarts).
# Without the marker the guard exits before Python starts, so the gates are
# off even though this run is active — never let that happen silently.
if not touch_sentinel():
print(
f"warning: could not write the loop-active sentinel at {sentinel_path()}. "
"The loop continues, but its edit gate, push gate, and Stop-hook "
"summary backstop stay disabled until the sentinel can be created.",
file=sys.stderr,
)
def read_run_tracking(pr: PullRequest) -> dict[str, Any]:
return load_sticky_state().get(_state_key(pr), {}).get("run", {})
def clear_run_tracking(pr: PullRequest) -> None:
"""Tear down the run accumulator and the delta baseline for this PR.
The render baseline (#95) is only meaningful while the full render it
fingerprints is still in the agent's context. Terminal completion ends that
context, so leaving the baseline behind would collapse threads a *later*
session has never seen down to URL-only stubs on its very first fetch.
Clearing both together makes the next run start from a full render.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
if isinstance(entry, dict):
stale = [field for field in ("run", "render_baseline") if field in entry]
for field in stale:
del entry[field]
if stale:
save_sticky_state(state)
# Drop the sentinel only when no loop remains anywhere — another repo's
# concurrent loop still needs the hooks live.
if not any_active_run():
clear_sentinel()
def begin_wait_chunk(pr: PullRequest, after_iso: str | None) -> dict[str, Any]:
"""Open one wait chunk: load cross-chunk wait state, applying the reset rule.
Reset rule (prevents cross-cycle leakage): if the stored anchor differs
from ``after_iso``, all wait progress (started_at, checks, settle state)
belongs to a previous cycle's wait and is discarded. ``checks`` counts
chunk invocations, incremented once per call. Fails open: state I/O
errors yield a fresh single-chunk state rather than crashing the wait.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
run = entry.get("run")
run = dict(run) if isinstance(run, dict) else {}
wait = run.get("wait")
wait = dict(wait) if isinstance(wait, dict) else {}
if wait.get("after") != after_iso:
wait = {"after": after_iso}
if not isinstance(wait.get("started_at"), str) or not wait.get("started_at"):
wait["started_at"] = metrics.now_iso()
wait["checks"] = _safe_int(wait.get("checks")) + 1
run["wait"] = wait
entry["run"] = run
state[key] = entry
try:
save_sticky_state(state)
except OSError as exc:
print(f"warning: could not persist wait state: {exc}", file=sys.stderr)
return wait
def read_wait_state(pr: PullRequest) -> dict[str, Any]:
run = read_run_tracking(pr)
wait = run.get("wait") if isinstance(run, dict) else None
return dict(wait) if isinstance(wait, dict) else {}
def _update_wait_state(pr: PullRequest, updates: dict[str, Any]) -> None:
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
run = entry.get("run")
run = dict(run) if isinstance(run, dict) else {}
wait = run.get("wait")
wait = dict(wait) if isinstance(wait, dict) else {}
wait.update(updates)
run["wait"] = wait
entry["run"] = run
state[key] = entry
try:
save_sticky_state(state)
except OSError as exc:
print(f"warning: could not persist wait state: {exc}", file=sys.stderr)
def save_wait_settle(pr: PullRequest, fingerprint: str, since_iso: str) -> None:
"""Persist the settle phase so a chunk boundary never restarts the quiet period."""
_update_wait_state(pr, {"stable_fingerprint": fingerprint, "stable_since": since_iso})
def save_wait_snapshot(pr: PullRequest, snapshot: dict[str, Any]) -> None:
"""Persist the last non-ready chunk result for --wait-heartbeat rendering."""
_update_wait_state(pr, {"last_snapshot": dict(snapshot)})
def clear_wait_state(pr: PullRequest) -> None:
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
if isinstance(entry, dict) and isinstance(entry.get("run"), dict) and "wait" in entry["run"]:
del entry["run"]["wait"]
try:
save_sticky_state(state)
except OSError as exc:
print(f"warning: could not clear wait state: {exc}", file=sys.stderr)
WAIT_FIRST_CHUNK_SECONDS = 60
WAIT_SECOND_CHUNK_SECONDS = 300
WAIT_LATER_CHUNK_SECONDS = 900
# Cadence of the one-line liveness heartbeat a blocking --wait writes to
# stderr while polling — the signal a user inspects when the wait runs as a
# background task.
WAIT_HEARTBEAT_INTERVAL_SECONDS = 60
def _parse_iso_utc(value: Any) -> _dt.datetime | None:
if not isinstance(value, str) or not value:
return None
try:
return _dt.datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(
tzinfo=_dt.timezone.utc
)
except ValueError:
return None
def wait_elapsed_seconds(
wait: dict[str, Any],
after_iso: str | None,
now: _dt.datetime | None = None,
) -> int:
"""Total wait elapsed, robust to state loss.
``max(now - started_at, now - after)``: if sticky state is corrupted or
deleted, a fresh started_at cannot silently restart the --timeout budget —
the --after anchor still bounds the total. Cycle 1 has no anchor and falls
back to started_at alone (acceptable: the cycle-1 fast path returns on
first detected activity).
"""
now = now or _dt.datetime.now(_dt.timezone.utc)
candidates = []
for value in (wait.get("started_at"), after_iso):
parsed = _parse_iso_utc(value)
if parsed is not None:
candidates.append((now - parsed).total_seconds())
return max(0, int(max(candidates))) if candidates else 0
def suggested_next_wait_seconds(checks: int) -> int:
"""Duration for the NEXT chunk, given how many chunks have already run.
Schedule: 60s → 300s → 900s. Chunked waits are the fallback for runtimes
without background-task completion notifications; every chunk costs a full
agent turn. One early heartbeat shows the loop is alive, one mid-length
chunk catches fast reviewers, then chunks stretch to 900s so a 30-minute
reviewer wait costs ~4 turns instead of ~8 (#102). Tradeoff: staler
liveness signal late in a long wait.
``checks`` follows the persisted counter in ``begin_wait_chunk``, which
increments at the START of each chunk — so at the end of the first chunk
``checks == 1``. That chunk has already served the 60s heartbeat, so the
next one is 300s. Reading ``checks <= 1`` as "still the first chunk"
silently produced 60s → 60s → 300s → 900s and burned an extra turn on
every long wait (#111 review).
"""
if checks <= 0:
return WAIT_FIRST_CHUNK_SECONDS
if checks == 1:
return WAIT_SECOND_CHUNK_SECONDS
return WAIT_LATER_CHUNK_SECONDS
def _latest_submitted_after(
pull_request: dict[str, Any], author: str, after_iso: str | None
) -> str | None:
"""Newest review submittedAt past the anchor, for the settling JSON payload."""
try:
times = [
r.get("submittedAt")
for r in filter_reviews(pull_request, author)
if isinstance(r.get("submittedAt"), str)
]
except (AttributeError, TypeError):
return None
if after_iso:
times = [t for t in times if t > after_iso]
return max(times) if times else None
def run_wait_chunk(
pr: PullRequest,
author: str,
*,
timeout_seconds: int,
interval_seconds: int,
quiet_seconds: int,
after_iso: str | None,
chunk_seconds: int,
) -> dict[str, Any]:
"""One bounded foreground wait chunk; the cross-chunk state machine's step.
Returns a dict with ``status`` in {waiting, settling, ready, timed_out}.
``ready`` carries ``pull_request`` (proceed into the fetch path exactly as
the legacy wait would); the others carry heartbeat fields and persist a
snapshot so ``--wait-heartbeat`` can render the human block later.
The quiet period is measured against the PERSISTED ``stable_since`` so a
chunk boundary never restarts settling.
"""
wait = begin_wait_chunk(pr, after_iso)
chunk_deadline = time.monotonic() + chunk_seconds
stable_fingerprint = wait.get("stable_fingerprint")
stable_since = _parse_iso_utc(wait.get("stable_since"))
while True:
pull_request = fetch_threads(pr)
refusal = reviewer_refusal(pull_request, author, after_iso=after_iso)
if refusal is not None:
clear_wait_state(pr)
return {"status": "refused", "author": author, "pull_request": None, **refusal}
fingerprint = review_activity_fingerprint(pull_request, author, after_iso=after_iso)
now_dt = _dt.datetime.now(_dt.timezone.utc)
elapsed = wait_elapsed_seconds(wait, after_iso, now=now_dt)
if fingerprint is not None and after_iso is None:
# Cycle 1 fast path: same semantics as the legacy wait.
clear_wait_state(pr)
return {"status": "ready", "pull_request": pull_request}
quiet_remaining: int | None = None
if fingerprint is not None:
if fingerprint != stable_fingerprint or stable_since is None:
stable_fingerprint = fingerprint
stable_since = now_dt
save_wait_settle(pr, fingerprint, now_dt.strftime("%Y-%m-%dT%H:%M:%SZ"))
quiet_elapsed = (now_dt - stable_since).total_seconds() if stable_since else 0.0
quiet_remaining = max(0, int(quiet_seconds - quiet_elapsed))
if quiet_remaining <= 0:
clear_wait_state(pr)
return {"status": "ready", "pull_request": pull_request}
if elapsed >= timeout_seconds:
snapshot = {
"status": "timed_out",
"author": author,
"elapsed_seconds": elapsed,
"checks": wait["checks"],
}
save_wait_snapshot(pr, snapshot)
return {**snapshot, "pull_request": None}
if time.monotonic() >= chunk_deadline:
next_wait = suggested_next_wait_seconds(wait["checks"])
snapshot = {
"status": "settling" if fingerprint is not None else "waiting",
"author": author,
"elapsed_seconds": elapsed,
"checks": wait["checks"],
"next_wait_seconds": next_wait,
}
if fingerprint is not None:
snapshot["quiet_period_remaining_seconds"] = quiet_remaining
snapshot["next_wait_seconds"] = max(
1, min(next_wait, quiet_remaining or next_wait)
)
submitted = _latest_submitted_after(pull_request, author, after_iso)
if submitted:
snapshot["submitted_at"] = submitted
save_wait_snapshot(pr, snapshot)
return {**snapshot, "pull_request": None}
time.sleep(min(interval_seconds, max(0.0, chunk_deadline - time.monotonic())))
def accumulate_fixed_markers(
pr: PullRequest,
*,
fingerprints: list[str] | None = None,
paths: list[str] | None = None,
) -> None:
"""Merge agent-supplied "fixed locally" markers into run tracking.
The agent passes ``--fixed-finding <fp>`` (finding fingerprints) and/or
fixed file paths as it remediates each cycle. They accumulate (union) under
the run entry so the terminal classification can read which findings the
agent claims to have fixed — one of the multi-signal inputs to
metrics.classify_finding_state. Stored alongside, never replacing, the
existing run-tracking ids/paths.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
run = entry.get("run")
run = dict(run) if isinstance(run, dict) else {}
fps_val = run.get("fixed_finding_fps")
fps = {x for x in fps_val if isinstance(x, str)} if isinstance(fps_val, list) else set()
fpaths_val = run.get("fixed_paths")
fpaths = {x for x in fpaths_val if isinstance(x, str)} if isinstance(fpaths_val, list) else set()
for fp in fingerprints or []:
if fp:
fps.add(fp)
for p in paths or []:
if p:
fpaths.add(p)
run["fixed_finding_fps"] = sorted(fps)
run["fixed_paths"] = sorted(fpaths)
entry["run"] = run
state[key] = entry
save_sticky_state(state)
def read_fixed_markers(pr: PullRequest) -> dict[str, set[str]]:
"""Return the accumulated fixed markers as ``{"fingerprints", "paths"}`` sets.
Empty sets when none were recorded or the state is missing/corrupt.
"""
run = read_run_tracking(pr)
if not isinstance(run, dict):
run = {}
fps = run.get("fixed_finding_fps", [])
fpaths = run.get("fixed_paths", [])
return {
"fingerprints": {x for x in fps if isinstance(x, str)} if isinstance(fps, list) else set(),
"paths": {x for x in fpaths if isinstance(x, str)} if isinstance(fpaths, list) else set(),
}
def accumulate_swept_patterns(pr: PullRequest, signatures: list[str]) -> None:
"""Union agent-supplied --swept-pattern signatures into run tracking.
Twin of accumulate_fixed_markers. A pattern the agent reports as swept this
cycle is matched against later cycles' findings to flag recurrence.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
run = entry.get("run")
run = dict(run) if isinstance(run, dict) else {}
val = run.get("swept_pattern_sigs")
sigs = {x for x in val if isinstance(x, str)} if isinstance(val, list) else set()
for s in signatures or []:
if s:
sigs.add(s)
run["swept_pattern_sigs"] = sorted(sigs)
entry["run"] = run
state[key] = entry
save_sticky_state(state)
def read_swept_patterns(pr: PullRequest) -> set[str]:
"""Return the accumulated swept pattern signatures, or empty set."""
run = read_run_tracking(pr)
val = run.get("swept_pattern_sigs", []) if isinstance(run, dict) else []
return {x for x in val if isinstance(x, str)} if isinstance(val, list) else set()
def swept_pattern_sets(
pr: PullRequest, history_swept: set[str]
) -> tuple[set[str], set[str]]:
"""Split swept signatures into ``(this_run, ever)``.
These answer different questions and must not be conflated. ``this_run`` is
what the agent reported via ``--swept-pattern`` on this run alone; it drives
the receipt's "Swept N patterns" and is what gets persisted, so a run that
swept nothing reports zero. ``ever`` folds in signatures recorded by earlier
runs of the same PR and drives recurrence detection only, because a pattern
swept in an earlier cycle that reappears now is exactly what the convergence
advisory exists to flag.
Persisting ``this_run`` rather than ``ever`` is what keeps the count
honest: ``metrics.pattern_history_for_pr`` already unions across records, so
writing the union back into each new record would compound it and the
reported count could never fall again.
"""
this_run = read_swept_patterns(pr)
return this_run, this_run | history_swept
def build_convergence(
pr: PullRequest, clusters: list[Any], history: dict[str, set[str]]
) -> dict[str, Any]:
"""Decide every pattern-related receipt field in one place.
The caller renders and persists what this returns; it must not recompute
any of it. That is the point: the swept counts were previously derived at
one site in ``main()`` and rendered at another ~85 lines away, so the two
could disagree and no test could reach either. Keeping the decision here
means a test of this function is a test of what the receipt actually says.
Returns ``stats`` (for the record's rates), ``line`` (the rendered
Convergence advisory), and ``swept_count`` / ``swept`` (this run only).
"""
# One signature entry per finding (repeat per cluster member) so the
# recurrence rate is over findings, not distinct patterns.
current_sigs = [c.signature for c in clusters for _ in range(c.count)]
this_run, ever = swept_pattern_sets(pr, history["swept"])
stats = cluster_findings.recurrence_stats(
current_sigs,
prior_sigs=prior_pattern_signatures(pr) | history["seen"],
swept_sigs=ever,
)
return {
"stats": stats,
"line": metrics.format_convergence_line(stats, swept_count=len(this_run)),
"swept_count": len(this_run),
"swept": sorted(this_run),
}
def _run_git(
args: list[str], *, cwd: str | Path | None = None
) -> subprocess.CompletedProcess[str] | None:
try:
return subprocess.run(
["git", *args],
cwd=cwd,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
except (OSError, ValueError, subprocess.SubprocessError):
return None
def _git_stdout(args: list[str], *, cwd: str | Path | None = None) -> str | None:
result = _run_git(args, cwd=cwd)
if result is None or result.returncode != 0:
return None
value = (result.stdout or "").strip()
return value or None
def repo_root(cwd: str | None = None) -> Path | None:
"""Return the working tree root, or None when it cannot be resolved.
Clustering reads the source lines findings anchor to, and thread paths are
repo-relative. Fails OPEN: any git error yields None, and the caller falls
back to prose-only clustering rather than reading the wrong files.
"""
top = _git_stdout(["rev-parse", "--show-toplevel"], cwd=cwd)
return Path(top) if top else None
def canonical_repo_full(pull_request: dict[str, Any], pr: PullRequest) -> str:
"""``owner/repo`` as GitHub spells it, for use as an exact-match store key.
``--pr OWNER/REPO#NUMBER`` keeps whatever casing the caller typed, and the
GraphQL lookup is case-insensitive, so the fetch succeeds. Verification
profiles and run metrics are keyed by an *exact* string, though: a
noncanonical spelling silently misses the existing entry and reads as "no
profile saved" / "no prior runs" (#107). The API's own ``nameWithOwner`` is
the authority; fall back to the input spelling only when it is absent.
"""
base = pull_request.get("baseRepository")
name = base.get("nameWithOwner") if isinstance(base, dict) else None
if isinstance(name, str) and name:
return name
return f"{pr.owner}/{pr.repo}"
def _canonical_pr_repositories(pull_request: dict[str, Any]) -> set[str]:
repositories = set()
for key in ("baseRepository", "headRepository"):
repository = pull_request.get(key)
name = repository.get("nameWithOwner") if isinstance(repository, dict) else None
if isinstance(name, str) and name:
repositories.add(name.casefold())
return repositories
def _checkout_repository(root: Path) -> str | None:
remote = _git_stdout(["remote", "get-url", "origin"], cwd=root)
if remote is None:
return None
match = re.search(r"github\.com(?::|/)([^/]+)/([^/#]+?)(?:\.git)?/?$", remote)
if match is None:
return None
return f"{match.group(1)}/{match.group(2)}".casefold()
def _worktree_is_clean(root: Path) -> bool:
result = _run_git(["status", "--porcelain", "--untracked-files=all"], cwd=root)
return (
result is not None
and result.returncode == 0
and not (result.stdout or "").strip()
)
def repo_root_for_pr(
pull_request: dict[str, Any], *, cwd: str | None = None
) -> Path | None:
"""Return the root only when this checkout is the selected PR head."""
expected_head = pull_request.get("headRefOid")
if not isinstance(expected_head, str) or not expected_head:
return None
root = repo_root(cwd)
if root is None or _git_stdout(["rev-parse", "HEAD"], cwd=root) != expected_head:
return None
if not _worktree_is_clean(root):
return None
checkout_repo = _checkout_repository(root)
selected_repositories = _canonical_pr_repositories(pull_request)
return root if checkout_repo in selected_repositories else None
def changed_files_in_range(
base: str | None, head: str | None, *, cwd: str | None = None
) -> set[str]:
"""Best-effort set of files changed in ``base..head`` via git.
Used to compute the ``file_changed`` signal: did the finding's path actually
change in the fixing commits? Fails OPEN — any git error (not a repo, bad
revs, git missing) yields an empty set rather than crashing the loop.
"""
if not base or not head:
return set()
try:
result = subprocess.run(
["git", "diff", "--name-only", f"{base}..{head}"],
cwd=cwd,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
except (OSError, ValueError, subprocess.SubprocessError):
return set()
if result.returncode != 0:
return set()
return {line.strip() for line in result.stdout.splitlines() if line.strip()}
LIKELY_FIXED_FINDING_STATES = {
"fixed_pushed_awaiting_review",
"fixed_pending_confirmation",
"stale_already_fixed",
}
def classify_remaining_finding_states(
threads: list[dict[str, Any]],
*,
fixed_fingerprints: set[str],
fixed_paths: set[str],
changed_paths: set[str],
prior_fingerprints: set[str],
judge_results: dict[str, dict[str, Any]],
cap_reached: bool,
) -> dict[str, str]:
"""Classify the still-actionable findings for terminal outcome derivation.
The state machine lives in ``metrics.classify_finding_state``. This adapter
only builds its explicit signal booleans from script-visible state and
agent-supplied fixed markers.
"""
states: dict[str, str] = {}
for index, thread in enumerate(threads):
if not isinstance(thread, dict):
continue
fingerprint = finding_fingerprint(thread)
path_val = thread.get("path")
path = path_val if isinstance(path_val, str) else ""
fixed_by_marker = fingerprint in fixed_fingerprints or path in fixed_paths
file_changed = path in changed_paths or path in fixed_paths
thread_id = thread.get("id")
result_key = thread_id if isinstance(thread_id, str) else fingerprint
judge_result = judge_results.get(result_key, {}) if result_key else {}
state = metrics.classify_finding_state(
{
"judge_needs_human": judge_result.get("verdict") == "needs_human",
"carried_over": fingerprint in prior_fingerprints,
"fixed_locally": fixed_by_marker,
"file_changed": file_changed,
"gemini_confirmed": False,
"cap_reached": cap_reached,
}
)
states[result_key or f"finding-{index}"] = state
return states
def count_likely_fixed_remaining(states: dict[str, str]) -> int:
return sum(1 for state in states.values() if state in LIKELY_FIXED_FINDING_STATES)
def track_finding_fingerprints(pr: PullRequest, current_fps: set[str]) -> dict[str, set[str]]:
"""Snapshot the prior cycle's finding fingerprints, then union the current.
Call ONLY on a real agent fetch (not ``--cycle-summary`` / ``--record-run``),
exactly once per cycle. It moves the running union into
``prior_seen_finding_fps`` *before* folding in this cycle's fingerprints, so
a later read-only ``--cycle-summary`` can ask "was this finding present in a
previous cycle?" without the current cycle's own fetch poisoning the answer.
Returns ``{"prior": <prior union>, "new": <current − prior>}``. Fails open:
any state I/O error yields an empty prior (everything reads as new).
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key, {})
if not isinstance(entry, dict):
entry = {}
run = entry.get("run", {})
if not isinstance(run, dict):
run = {}
prior_val = run.get("seen_finding_fps", [])
prior = {x for x in prior_val if isinstance(x, str)} if isinstance(prior_val, list) else set()
run["prior_seen_finding_fps"] = sorted(prior)
run["seen_finding_fps"] = sorted(prior | current_fps)
entry["run"] = run
state[key] = entry
save_sticky_state(state)
return {"prior": prior, "new": current_fps - prior}
def prior_finding_fingerprints(pr: PullRequest) -> set[str]:
"""Finding fingerprints seen in cycles *before* the current one.
Read by the receipt path to mark a current finding as carried-over. Empty
on cycle 1 (no prior cycle) — so every cycle-1 finding reads as new.
"""
state = load_sticky_state()
entry = state.get(_state_key(pr), {})
run = entry.get("run", {}) if isinstance(entry, dict) else {}
value = run.get("prior_seen_finding_fps", []) if isinstance(run, dict) else []
return {x for x in value if isinstance(x, str)} if isinstance(value, list) else set()
def track_pattern_signatures(pr: PullRequest, current_sigs: set[str]) -> dict[str, set[str]]:
"""Pattern-granularity twin of track_finding_fingerprints.
Moves the running union into ``prior_seen_pattern_sigs`` BEFORE folding in
this cycle's signatures, so a later read can ask "was this pattern present
in a previous cycle?" without the current fetch poisoning the answer.
Returns ``{"prior": <prior union>, "new": <current − prior>}``. Fails open.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key, {})
if not isinstance(entry, dict):
entry = {}
run = entry.get("run", {})
if not isinstance(run, dict):
run = {}
prior_val = run.get("seen_pattern_sigs", [])
prior = {x for x in prior_val if isinstance(x, str)} if isinstance(prior_val, list) else set()
run["prior_seen_pattern_sigs"] = sorted(prior)
run["seen_pattern_sigs"] = sorted(prior | current_sigs)
entry["run"] = run
state[key] = entry
save_sticky_state(state)
return {"prior": prior, "new": current_sigs - prior}
def prior_pattern_signatures(pr: PullRequest) -> set[str]:
"""Pattern signatures seen in cycles before the current one. Empty on cycle 1."""
state = load_sticky_state()
entry = state.get(_state_key(pr), {})
run = entry.get("run", {}) if isinstance(entry, dict) else {}
value = run.get("prior_seen_pattern_sigs", []) if isinstance(run, dict) else []
return {x for x in value if isinstance(x, str)} if isinstance(value, list) else set()
def detect_no_progress(pr: PullRequest, current_fingerprint: str) -> bool:
"""True when the actionable thread fingerprint is unchanged from last cycle.
Call AFTER ``update_run_tracking`` so ``update_seq`` reflects the current
fetch. Reads the fingerprint stored by the previous call, stores the new
one, and returns True only when both conditions hold:
- ``update_seq >= 2``: at least two fetches have run (one per-cycle fix
attempt has already been made without clearing the set of open threads).
- The fingerprint is identical to the one stored on the previous call.
A True return means no observable code change resolved any of the threads
the agent was supposed to fix. The agent should stop with
``--record-run --outcome no_progress``.
Fails OPEN: any state read/write error allows the push (correctness aid
must never wedge the loop).
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key, {})
if not isinstance(entry, dict):
entry = {}
run = entry.get("run", {})
if not isinstance(run, dict):
run = {}
prev_fingerprint = run.get("thread_fingerprint")
update_seq = _safe_int(run.get("update_seq", 0))
run["thread_fingerprint"] = current_fingerprint
entry["run"] = run
state[key] = entry
save_sticky_state(state)
return bool(
prev_fingerprint
and prev_fingerprint == current_fingerprint
and update_seq >= 2
)
def stamp_summary_emitted(pr: PullRequest) -> None:
"""Record that a summary covering the current run state was just emitted.
Sets ``last_summary_seq`` to the current ``update_seq`` so the Stop-hook
backstop won't re-emit a summary the agent already showed. A no-op when no
run is being tracked, so it never resurrects a cleared/absent run block.
"""
state = load_sticky_state()
if not isinstance(state, dict):
return
key = _state_key(pr)
entry = state.get(key)
if not isinstance(entry, dict):
return
run = entry.get("run")
if not isinstance(run, dict) or "update_seq" not in run:
return
seq = _safe_int(run.get("update_seq"))
if seq <= 0:
return # corrupt/non-numeric update_seq — nothing meaningful to stamp
run["last_summary_seq"] = seq
state[key]["run"] = run
save_sticky_state(state)
def accumulate_judge_results(
pr: PullRequest, judge_results: dict[str, dict[str, Any]]
) -> None:
"""Persist this cycle's judge verdicts into the run accumulator.
Verdicts are keyed by thread id so a later cycle's re-judgement of the same
thread supersedes an earlier one. Stored alongside the run's finding ids /
paths so the terminal ``--record-run`` can build a complete judge block even
after the findings themselves have been resolved. A no-op when there are no
results, so a run that never judged keeps no ``judge_ran`` flag.
"""
if not judge_results:
return
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key)
if not isinstance(entry, dict):
entry = {}
else:
entry = dict(entry)
run = entry.get("run")
if not isinstance(run, dict):
run = {}
else:
run = dict(run)
stored = run.get("judge_results")
stored_dict = dict(stored) if isinstance(stored, dict) else {}
stored_dict.update(judge_results)
run["judge_results"] = stored_dict
run["judge_ran"] = True
entry["run"] = run
state[key] = entry
save_sticky_state(state)
def merge_judge_results(
accumulated: dict[str, dict[str, Any]] | None,
current: dict[str, dict[str, Any]] | None,
) -> dict[str, dict[str, Any]]:
"""Union accumulated and current verdicts; current supersedes per thread id."""
merged = dict(accumulated or {})
merged.update(current or {})
return merged
def resolve_judge_phase(explicit_phase: str | None, *, record_run: bool) -> str:
"""Infer the judge phase from invocation context when unset.
The agent need not remember to pass --judge-phase: a terminal --record-run
invocation is the loop's completion, everything else is a per-cycle fetch.
An explicit flag always wins.
"""
if explicit_phase is not None:
return explicit_phase
return "complete" if record_run else "cycle"
def record_cycle(
pr: PullRequest,
started_at: str,
finding_count: int,
outcome: str,
finished_at: str | None = None,
) -> dict[str, Any]:
"""Append one active-cycle timing entry to the run accumulator.
A *cycle* is one active remediation attempt — fetch, analyze, edit, verify,
prepare/post the response. It measures active work only: the agent passes
``started_at`` as the moment active work for this cycle began, so waits
between cycles (ScheduleWakeup, polling, rate-limit sleeps, idle time) are
excluded by construction. ``outcome`` is "continued" for a non-terminal
cycle or the terminal outcome label for the last one. Returns the entry.
"""
finished_at = finished_at or _now_iso()
duration = metrics._duration_seconds(started_at, finished_at)
cycle = {
"started_at": started_at,
"finished_at": finished_at,
"duration_seconds": duration,
"finding_count": int(finding_count),
"outcome": outcome,
}
state = load_sticky_state()
# The state file can be hand-edited or corrupt; verify each level is the
# expected type before use (a non-dict state would AttributeError on .get,
# and dict()/list() on a string/int would raise), mirroring
# accumulate_judge_results.
state = state if isinstance(state, dict) else {}
key = _state_key(pr)
entry = state.get(key)
entry = dict(entry) if isinstance(entry, dict) else {}
run = entry.get("run")
run = dict(run) if isinstance(run, dict) else {}
cycles = run.get("cycles")
cycles = list(cycles) if isinstance(cycles, list) else []
cycles.append(cycle)
run["cycles"] = cycles
entry["run"] = run
state[key] = entry
save_sticky_state(state)
return cycle
def _now_iso() -> str:
return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def find_existing_sticky_comment(pr: PullRequest) -> int | None:
"""Look up the sticky receipt for a PR by scanning issue comments for the marker.
Used as a recovery path when the local state file is missing — the marker
embedded in the receipt body is the source of truth on GitHub.
"""
# `gh api --paginate` runs --jq on each page independently, so a per-page
# aggregator like `[...] | last` returns the last id PER PAGE — not the
# global last. Emit one id per line across all pages and pick the last
# valid one in Python.
result = run_gh(
[
"api",
f"repos/{pr.owner}/{pr.repo}/issues/{pr.number}/comments",
"--paginate",
"--jq",
f'.[] | select(.body != null and (.body | contains("{STICKY_RECEIPT_MARKER}"))) | .id',
]
)
if isinstance(result, int):
return result
if isinstance(result, str):
ids = [int(line.strip()) for line in result.splitlines() if line.strip().isdigit()]
if ids:
return ids[-1]
return None
def post_or_update_sticky_receipt(
pr: PullRequest, body: str, *, dry_run: bool = False
) -> int | None:
"""Post or in-place edit the sticky receipt comment for ``pr``.
First invocation per PR posts a fresh comment and records its id in the
state file. Subsequent invocations PATCH the same comment so the user
sees one live status block on the PR instead of N new comments.
"""
state = load_sticky_state()
key = _state_key(pr)
entry = state.get(key, {})
comment_id = entry.get("comment_id")
if comment_id is None:
# Fall back to GitHub-side discovery before posting a duplicate.
comment_id = find_existing_sticky_comment(pr)
if comment_id and not dry_run:
entry["comment_id"] = comment_id
if dry_run:
verb = "PATCH" if comment_id else "POST"
print(f"[dry-run] would {verb} sticky receipt on {key}", file=sys.stderr)
return comment_id
if comment_id:
try:
run_gh(
[
"api",
"-X",
"PATCH",
f"repos/{pr.owner}/{pr.repo}/issues/comments/{comment_id}",
"-f",
f"body={body}",
]
)
entry["updated_at"] = _now_iso()
except RuntimeError as exc:
# If the sticky comment was deleted on GitHub, PATCH returns 404.
# Drop the stale id and fall through to POST a fresh comment so
# the loop recovers without crashing.
msg = str(exc)
if "404" in msg or "Not Found" in msg:
print(
f"warning: sticky receipt {comment_id} no longer exists on GitHub "
"(deleted?). Posting a new comment.",
file=sys.stderr,
)
comment_id = None
else:
raise
if not comment_id:
result = run_gh(
[
"api",
"-X",
"POST",
f"repos/{pr.owner}/{pr.repo}/issues/{pr.number}/comments",
"-f",
f"body={body}",
]
)
if not isinstance(result, dict) or "id" not in result:
raise RuntimeError(f"Failed to post sticky receipt: {result}")
comment_id = int(result["id"])
entry["comment_id"] = comment_id
entry["started_at"] = _now_iso()
entry["updated_at"] = entry["started_at"]
state[key] = entry
save_sticky_state(state)
return comment_id
def render_receipt(
pr: PullRequest,
pull_request: dict[str, Any],
threads: list[dict[str, Any]],
*,
author: str,
resolved_outdated: int,
resolved_addressed_by_reply: int,
rereview_count: int,
rereview_limit: int,
status: str | None = None,
sticky: bool = False,
) -> str:
"""Render the markdown receipt body.
``status`` (RUNNING / DONE / STOPPED) appears in the header when set.
``sticky`` embeds the discovery marker and a "last updated" timestamp.
"""
counts = severity_counts(threads)
deferred = len(addressed_by_reply_threads(pull_request, author))
sev_line = ", ".join(
f"{k}={counts[k]}" for k in ("critical", "high", "medium", "low", "unknown") if counts.get(k)
) or "none"
header_suffix = f" — {status}" if status else ""
parts = [
f"### gh-review-loop receipt{header_suffix}",
"",
"| metric | value |",
"|---|---|",
f"| re-review requests used (cap) | {rereview_count} / {rereview_limit} |",
f"| outdated threads resolved | {resolved_outdated} |",
f"| addressed-by-reply threads resolved | {resolved_addressed_by_reply} |",
f"| addressed-by-reply still pending | {deferred} |",
f"| actionable threads remaining | {len(threads)} |",
f"| severity breakdown (actionable) | {sev_line} |",
"",
]
if sticky:
parts.append(f"_Last updated: {_now_iso()}. Receipt edits in place as the loop progresses._")
parts.append("")
parts.append(STICKY_RECEIPT_MARKER)
else:
parts.append("_Generated by `scripts/fetch_gemini_threads.py --post-receipt`._")
return "\n".join(parts) + "\n"
def build_receipt_comment_body(
blocks: list[str], *, status: str, findings_block: str = ""
) -> str:
"""Markdown body carrying the FULL cycle/terminal receipt on the PR.
This is the single authoritative delivery channel for receipt content
(#86): chat gets a one-line pointer, this comment gets everything. The
plaintext blocks ride in a code fence; the findings list stays outside it
so GitHub auto-links each finding's comment URL.
"""
parts = [f"### gh-review-loop receipt — {status}", ""]
content = "\n\n".join(block for block in blocks if block)
if content:
parts.extend(["```", content, "```", ""])
if findings_block:
parts.extend([findings_block, ""])
parts.append(
f"_Last updated: {_now_iso()}. This comment is edited in place as the loop progresses._"
)
parts.append("")
parts.append(STICKY_RECEIPT_MARKER)
return "\n".join(parts) + "\n"
def thread_fingerprint(threads: list[dict[str, Any]]) -> str:
payload = []
for thread in threads:
comments = thread.get("comments", [])
payload.append(
{
"path": thread.get("path") or first_value(thread, "path"),
"line": thread.get("line") or thread.get("originalLine") or first_value(thread, "line"),
"comments": [
{
"author": (comment.get("author") or {}).get("login"),
"body": comment.get("body"),
}
for comment in comments
],
}
)
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
def finding_fingerprint(thread: dict[str, Any]) -> str:
"""Content identity of a single finding, stable across re-posts.
Gemini re-posts the same suggestion as a *new* thread (new id) on each
re-review, so the thread id cannot tell a fresh finding from a carried-over
one. The path plus the normalized suggestion text can: we strip the leading
severity image markdown Gemini prepends and collapse whitespace, so cosmetic
differences (line-number echoes, re-wrapping) don't change the fingerprint.
"""
if not isinstance(thread, dict):
return ""
path_val = thread.get("path")
path = path_val if path_val is not None else ""
comments = _iter_comments(thread)
first_comment = comments[0].get("body") if comments and isinstance(comments[0], dict) else None
body = first_comment if isinstance(first_comment, str) else ""
body = re.sub(r"!\[[^\]]*\]\([^)]*\)", "", body) # drop severity image
body = re.sub(r"\s+", " ", body).strip().lower()
return hashlib.sha1(f"{path}\n{body[:1000]}".encode()).hexdigest()[:16]
def review_activity_fingerprint(
pull_request: dict[str, Any],
author: str,
after_iso: str | None = None,
) -> str | None:
"""Compute a hash of all Gemini review activity in ``pull_request``.
Returns ``None`` when no matching activity exists (caller keeps polling).
``after_iso`` — optional ISO-8601 lower-bound (exclusive). When set,
only reviews submitted after that timestamp and only threads whose newest
comment was created after that timestamp are counted. Pass the timestamp
of the re-review request comment so the waiter ignores prior-cycle
activity and only returns once Gemini has genuinely responded to the new
cycle's push.
"""
reviews = filter_reviews(pull_request, author)
if after_iso:
reviews = [r for r in reviews if (r.get("submittedAt") or "") > after_iso]
authored_threads = filter_threads(
pull_request,
author=author,
include_resolved=True,
include_outdated=True,
)
if after_iso:
# A thread counts as "new" when at least one of its comments was
# created after the anchor. Using the newest comment timestamp means
# a thread where Gemini added a follow-up reply after the re-review
# request is also captured correctly.
authored_threads = [
t for t in authored_threads
if any(
(c.get("createdAt") or "") > after_iso
for c in t.get("comments", [])
)
]
if not reviews and not authored_threads:
return None
payload = {"reviews": reviews, "threads": authored_threads}
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
def wait_for_stable_review(
pr: PullRequest,
author: str,
timeout_seconds: int,
interval_seconds: int,
quiet_seconds: int,
after_iso: str | None = None,
) -> dict[str, Any]:
"""Poll until ``author``'s review activity on ``pr`` is present and stable.
``after_iso`` anchors the wait to a specific point in time (ISO-8601,
exclusive). When set, review activity submitted at or before that
timestamp is ignored — the poller only returns once Gemini has posted
activity *after* that timestamp. This prevents a cycle-2 wait from
returning immediately because cycle-1's review is already stable.
Typical usage: pass the ``createdAt`` of the re-review request comment
that triggered the new cycle, so the wait is anchored to that request.
Designed to run as a background task: while polling it writes a one-line
heartbeat to stderr every ``WAIT_HEARTBEAT_INTERVAL_SECONDS`` so an
inspected task shows liveness, and a timeout raises the typed
``WaitTimedOut`` so the CLI ends the captured output with a relayable
line instead of a traceback.
"""
start = time.monotonic()
deadline = start + timeout_seconds
last_fingerprint: str | None = None
stable_since: float | None = None
last_heartbeat: float | None = None
if after_iso:
print(
f"Waiting for {author} review activity after {after_iso}...",
file=sys.stderr,
)
while True:
pull_request = fetch_threads(pr)
refusal = reviewer_refusal(pull_request, author, after_iso=after_iso)
if refusal is not None:
raise ReviewerRefused(refusal)
fingerprint = review_activity_fingerprint(pull_request, author, after_iso=after_iso)
now = time.monotonic()
if fingerprint is None:
if (
last_heartbeat is None
or now - last_heartbeat >= WAIT_HEARTBEAT_INTERVAL_SECONDS
):
print(
f"Waiting for {author} review activity... "
f"({int(now - start)}s elapsed, timeout {timeout_seconds}s)",
file=sys.stderr,
)
last_heartbeat = now
elif after_iso is None:
# Cycle 1 / initial review: activity is present, so return at once
# without waiting for it to settle. At the initial review there
# either are comments or there aren't — the quiet-period settle only
# earns its keep on cycle 2+ (after_iso set), where a freshly pushed
# fix needs Gemini's re-review to stabilize before we fetch.
print(f"Detected {author} review activity.", file=sys.stderr)
return pull_request
elif fingerprint != last_fingerprint:
print(f"Detected {author} review activity; waiting for it to settle...", file=sys.stderr)
last_fingerprint = fingerprint
stable_since = now
elif stable_since is not None and now - stable_since >= quiet_seconds:
print(f"{author} review activity is stable.", file=sys.stderr)
return pull_request
if now >= deadline:
raise WaitTimedOut(int(now - start))
time.sleep(min(interval_seconds, max(0.0, deadline - now)))
def format_judge_verdict_summary(
judge_results: dict[str, dict[str, Any]],
phase: str,
) -> str:
"""One-line verdict breakdown for agent narration relay.
Printed after the markdown thread list so the agent can copy it verbatim
into its text response. Only threads where the judge actually produced a
verdict (status == "ok") are counted; skipped threads are excluded.
Output format (intended for literal copy-paste into agent text):
[loop] judge (cycle): 3 thread(s) — valid_actionable: 2, false_positive: 1
"""
verdicts: Counter[str] = Counter(
r["verdict"]
for r in judge_results.values()
if r.get("status") == "ok" and r.get("verdict")
)
if not verdicts:
return f"[loop] judge ({phase}): evaluated {len(judge_results)} thread(s) — all skipped/errored"
total = sum(verdicts.values())
breakdown = ", ".join(
f"{verdict}: {count}"
for verdict, count in sorted(verdicts.items(), key=lambda kv: -kv[1])
)
return f"[loop] judge ({phase}): {total} thread(s) evaluated — {breakdown}"
def format_judge_thread_table(
threads: list[dict[str, Any]],
judge_results: dict[str, dict[str, Any]],
phase: str,
) -> str:
"""Per-thread judge decision table for agent narration relay.
Printed after the markdown thread list so the agent can copy it verbatim
into its text response. Each row shows the thread location, severity,
verdict, recommended action, and confidence — all in one place, outside
the collapsed thread markdown block.
Format:
[loop] judge eval (cycle): 3 thread(s)
1 src/foo.py:42 [high] — valid_actionable · fix · conf 0.91
2 src/bar.py:15 [medium] — needs_human · escalate · conf 0.84
3 src/baz.py:7 [low] — skipped (no API key)
"""
rows = []
for i, thread in enumerate(threads, 1):
path = thread.get("path") or "?"
line = thread.get("line") or thread.get("originalLine") or "?"
sev = thread_severity(thread)
tid = thread.get("id", "")
jr = judge_results.get(tid)
if jr and jr.get("status") == "ok":
verdict = jr.get("verdict", "?")
action = jr.get("recommended_action", "?")
conf = jr.get("confidence", 0.0)
try:
conf_str = f"{float(conf):.2f}"
except (TypeError, ValueError):
conf_str = "?"
rows.append(
f" {i} {path}:{line} [{sev}] — {verdict} · {action} · conf {conf_str}"
)
elif jr and jr.get("status") == "skipped":
reason = jr.get("skip_reason", "unknown")
rows.append(f" {i} {path}:{line} [{sev}] — skipped ({reason})")
else:
rows.append(f" {i} {path}:{line} [{sev}] — not evaluated")
n = len(threads)
header = f"[loop] judge eval ({phase}): {n} thread(s)"
return "\n".join([header] + rows)
def load_profile_for_repo(repo: str) -> dict[str, Any] | None:
"""Best-effort saved verification profile lookup for formatter commands."""
try:
from judge import get_profile # noqa: PLC0415
except ImportError:
return None
try:
return get_profile(repo)
except Exception:
return None
def judge_eval_requested(judge_status: dict[str, Any]) -> bool:
if not isinstance(judge_status, dict) or judge_status.get("ran"):
return False
mode = judge_status.get("mode")
reason = str(judge_status.get("skip_reason") or "")
return mode in {"on_cycle", "on_complete", "once"} and "no-op" not in reason
def color_loop_block(block: str, *, enabled: bool) -> str:
"""Color only human-readable loop-owned blocks."""
if isinstance(block, str) and block.startswith("[loop]"):
return color_loop(block, enabled=enabled)
return block
def print_clustering_blocks(clusters: list[Any]) -> None:
"""Print the shared pattern view and any degenerate-clustering warning."""
patterns_block = metrics.format_patterns_block(clusters)
if patterns_block:
print(patterns_block)
clustering_advisory = metrics.format_degenerate_clustering_advisory(clusters)
if clustering_advisory:
print(clustering_advisory)
# Diff hunks are convenience context, not the source of truth — the agent has
# the checkout and reads files before editing. Cap them so verbose reviewers
# don't dominate the fetch output.
HUNK_MAX_LINES = 10
def truncate_diff_hunk(hunk: str) -> str:
lines = hunk.rstrip().splitlines()
if len(lines) <= HUNK_MAX_LINES:
return "\n".join(lines)
dropped = len(lines) - HUNK_MAX_LINES
return "\n".join(lines[:HUNK_MAX_LINES]) + f"\n… (+{dropped} more diff lines)"
def _thread_header_parts(thread: dict[str, Any]) -> tuple[str, Any, str]:
"""(path, line, status_text) shared by the full block and the collapsed line."""
path = thread.get("path") or first_value(thread, "path") or "(unknown path)"
line = thread.get("line") or first_value(thread, "line") or thread.get("originalLine") or ""
status = []
severity = thread_severity(thread)
if severity != "unknown":
status.append(severity)
if thread.get("isResolved"):
status.append("resolved")
if thread.get("isOutdated"):
status.append("outdated")
status_text = f" [{' '.join(status)}]" if status else ""
return path, line, status_text
def render_thread_block(
thread: dict[str, Any],
judge_result: dict[str, Any] | None = None,
*,
index: int | None = None,
include_confidence: bool = True,
) -> list[str]:
"""The full rendered block for one thread.
Two forms from one renderer, so the delta collapse (#95) can never diverge
from what is actually shown:
- Output form (``index=N``, ``include_confidence=True``) — what
``render_markdown`` emits.
- Canonical form (``index=None``, ``include_confidence=False``) — what
``thread_render_fingerprint`` hashes. Index-free because the positional
number shifts whenever an earlier thread resolves; confidence-free
because the judge's confidence decimal jitters across re-runs. Both are
volatile decoration, not semantic change.
"""
path, line, status_text = _thread_header_parts(thread)
judge_line = ""
if judge_result and judge_result.get("status") == "ok":
conf = f"conf {judge_result['confidence']:.2f}, " if include_confidence else ""
judge_line = (
f" \n > **Judge:** `{judge_result['verdict']}` ({conf}"
f"severity_override={judge_result['severity_override']}, "
f"action={judge_result['recommended_action']}). {judge_result.get('reason', '')}"
)
elif judge_result and judge_result.get("status") == "skipped":
judge_line = f" \n > **Judge:** skipped — {judge_result.get('skip_reason', '')}"
prefix = f"## {index}. " if index is not None else "## "
lines = [f"{prefix}{path}:{line}{status_text}{judge_line}"]
for comment in thread["comments"]:
body = (comment.get("body") or "").strip()
url = comment.get("url") or ""
created = comment.get("createdAt") or ""
lines.extend(["", f"- {created} {url}".strip(), "", body, ""])
hunk = comment.get("diffHunk")
if hunk:
lines.extend(["```diff", truncate_diff_hunk(hunk), "```", ""])
return lines
def thread_render_fingerprint(
thread: dict[str, Any], judge_result: dict[str, Any] | None = None
) -> str:
"""Hash of the canonical rendered block — render first, fingerprint second.
Distinct from ``finding_fingerprint`` (a deliberately fuzzy identity for
carried-over tracking): this is an exact content hash whose only job is
"would this thread's block re-render byte-identical to last cycle's?".
Any renderer change automatically invalidates stale baselines.
"""
canonical = "\n".join(
render_thread_block(thread, judge_result, index=None, include_confidence=False)
)
return hashlib.sha256(canonical.encode()).hexdigest()
def render_collapsed_thread_line(thread: dict[str, Any], *, index: int) -> list[str]:
"""One-line stand-in for a thread whose full block is unchanged (#95).
Keeps everything needed to act without the body: anchor, severity, and the
comment URL for recovery after context loss.
"""
path, line, status_text = _thread_header_parts(thread)
comments = thread.get("comments") or []
first = comments[0] if comments and isinstance(comments[0], dict) else {}
url = first.get("url") or ""
suffix = f" ({url})" if url else ""
return [f"## {index}. {path}:{line}{status_text} — unchanged since last cycle{suffix}", ""]
def render_markdown(
pr: dict[str, Any],
threads: list[dict[str, Any]],
author: str,
*,
reviewer_name: str = DEFAULT_PROVIDER_NAME,
review_trigger_mention: str = DEFAULT_REVIEW_TRIGGER_MENTION,
judge_results: dict[str, dict[str, Any]] | None = None,
baseline: dict[str, str] | None = None,
full: bool = False,
) -> str:
reviews = filter_reviews(pr, author)
rereviews = rereview_requests(pr, review_trigger_mention=review_trigger_mention)
outdated = outdated_unresolved_threads(pr, author)
deferred = addressed_by_reply_threads(pr, author)
lines = [
f"# {reviewer_name} Threads for PR #{pr.get('number')}",
"",
f"PR: {pr.get('url')}",
f"Author filter: `{author}`",
f"Reviews by author: {len(reviews)}",
f"Re-review requests: {len(rereviews)}",
f"Unresolved outdated threads: {len(outdated)}",
f"Addressed-by-reply threads: {len(deferred)}",
f"Threads: {len(threads)}",
"",
]
if reviews:
lines.append("## Reviews")
for review in reviews:
state = review.get("state") or "UNKNOWN"
submitted = review.get("submittedAt") or ""
url = review.get("url") or ""
lines.append(f"- {state} {submitted} {url}".strip())
lines.append("")
for index, thread in enumerate(threads, start=1):
# Optional judge annotation — only present when the user opted in via
# --judge-mode (or saved preference) AND --judge-phase matched.
jr = judge_results.get(thread.get("id")) if judge_results else None
thread_id = thread.get("id")
if (
not full
and baseline is not None
and isinstance(thread_id, str)
and baseline.get(thread_id) == thread_render_fingerprint(thread, jr)
):
lines.extend(render_collapsed_thread_line(thread, index=index))
continue
lines.extend(render_thread_block(thread, jr, index=index))
return "\n".join(lines).rstrip() + "\n"
def first_value(thread: dict[str, Any], key: str) -> Any:
for comment in thread.get("comments", []):
value = comment.get(key)
if value is not None:
return value
return None
def main() -> int:
# Post-rename housekeeping: move ~/.config/gh-gemini-review-loop to
# ~/.config/gh-review-loop once, leaving a compatibility symlink for
# pre-rename plugin versions. No-op after the first run. Done here (the
# loop's single entry point) so the hook gates never pay for it.
migrate_legacy_state_dir()
parser = argparse.ArgumentParser(
description=__doc__,
epilog=(
"Verification profiles: the loop detects a per-repo check profile on "
"first run (pytest/ruff, npm scripts, cargo, go) and stores it in "
"preferences.json. Say \"set up a verification profile for this repo\" "
"to configure, or \"skip verification profile\" to opt out."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--pr", help="PR URL or OWNER/REPO#NUMBER. Defaults to current branch PR.")
parser.add_argument("--repo", default=None, help="OWNER/REPO for formatter-only commands.")
parser.add_argument("--author", default=None, help=f"Reviewer author login. Default: {DEFAULT_AUTHOR}")
parser.add_argument(
"--reviewer",
default=None,
help="Reviewer bot login to select and persist for this PR.",
)
parser.add_argument(
"--reviewer-source",
choices=["explicit", "confirmed"],
default="explicit",
help=(
"Selection source to persist with --reviewer. Use 'confirmed' for "
"the first-run prompt confirmation path."
),
)
parser.add_argument(
"--reviewer-name",
default=None,
help=f"Human-readable reviewer name for output. Default: {DEFAULT_PROVIDER_NAME!r}.",
)
parser.add_argument(
"--review-trigger-mention",
default=None,
help=(
"Mention phrase that requests a re-review and counts toward the loop cap. "
f"Default: {DEFAULT_REVIEW_TRIGGER_MENTION!r}."
),
)
parser.add_argument(
"--list-reviewers",
action="store_true",
help="Discover AI reviewer bot candidates on this PR and exit without mutating state.",
)
parser.add_argument(
"--reset-reviewer",
action="store_true",
help="Clear the persisted reviewer selection for this PR and exit.",
)
parser.add_argument("--format", choices=["markdown", "json"], default="markdown")
parser.add_argument(
"--no-color",
action="store_true",
help="Disable ANSI color for human-readable [loop] blocks.",
)
parser.add_argument(
"--profile-intro",
action="store_true",
help="Print the saved verification-profile intro block for --repo and exit.",
)
parser.add_argument(
"--planned-verification",
action="store_true",
help="Print the planned repo-aware verification block for --repo and exit.",
)
parser.add_argument(
"--wait-heartbeat",
action="store_true",
help=(
"Print the human heartbeat block for the PR's in-progress chunked "
"wait (from persisted state) and exit. Use after a --format json "
"wait chunk so JSON stdout stays machine-only."
),
)
parser.add_argument("--include-resolved", action="store_true")
parser.add_argument("--include-outdated", action="store_true")
parser.add_argument(
"--full",
action="store_true",
help=(
"Render every thread in full, ignoring the delta baseline. Use after "
"context loss or compaction to re-establish the baseline. Diff hunks "
"still use the standard truncation cap."
),
)
parser.add_argument("--wait", action="store_true", help="Poll until reviewer activity appears and is stable.")
parser.add_argument("--timeout", type=int, default=900, help="Seconds to wait with --wait. Default: 900.")
parser.add_argument("--interval", type=int, default=20, help="Polling interval in seconds with --wait. Default: 20.")
parser.add_argument("--quiet-period", type=int, default=45, help="Stable activity period in seconds with --wait. Default: 45.")
parser.add_argument(
"--wait-chunk-seconds",
type=int,
default=None,
help=(
"With --wait, return after at most this many seconds with a "
"deterministic waiting/settling/timed_out status instead of "
"blocking until --timeout. --timeout stays the TOTAL wait budget "
"across chunks. Omit for legacy blocking behavior."
),
)
parser.add_argument(
"--after",
metavar="ISO8601",
default=None,
help=(
"With --wait, ignore reviewer activity submitted at or before this "
"ISO-8601 timestamp. Pass the createdAt of the re-review request "
"comment so the waiter only returns once the reviewer has genuinely "
"responded to the new cycle's push, not to prior-cycle reviews."
),
)
parser.add_argument(
"--resolve-outdated",
dest="resolve_outdated",
action="store_true",
default=True,
help="Resolve unresolved outdated reviewer threads before printing current feedback. Enabled by default.",
)
parser.add_argument(
"--no-resolve-outdated",
dest="resolve_outdated",
action="store_false",
help="Do not resolve outdated reviewer threads; use for read-only inspection.",
)
parser.add_argument(
"--max-rereview-requests",
type=nonnegative_int,
default=None,
help=(
"Warn when prior reviewer re-review requests reach this limit. "
"Overrides max_rereview_requests in ~/.config/gh-review-loop/preferences.json "
f"(default: {DEFAULT_REREVIEW_LIMIT})."
),
)
# Cleanup (resolve-outdated, resolve-addressed-by-reply) now always runs
# regardless of the re-review cap. --resolve-past-cap / --ignore-loop-limit
# are kept as accepted no-ops for backward compatibility with older scripts.
parser.add_argument(
"--resolve-past-cap",
dest="ignore_loop_limit",
action="store_true",
help=argparse.SUPPRESS,
)
parser.add_argument(
"--ignore-loop-limit",
dest="ignore_loop_limit",
action="store_true",
help=argparse.SUPPRESS,
)
parser.add_argument(
"--agent-login",
default=None,
help=(
"GitHub login of the agent posting re-review requests. If omitted, the script "
"auto-detects via `gh api user`. Used to count only the agent's own re-reviews "
"toward the cap (humans pinging the reviewer do not consume cycles)."
),
)
parser.add_argument(
"--no-agent-filter",
action="store_true",
help="Disable agent-login filtering; count ANY configured reviewer re-review comment.",
)
parser.add_argument(
"--post-receipt",
action="store_true",
help=(
"Post a summary 'loop receipt' comment to the PR after fetch/filter. "
"Includes re-review requests used, threads resolved, severity breakdown, and remaining actionable count. "
"Posts a NEW comment each invocation; for one live comment edited in place, use --sticky-receipt."
),
)
parser.add_argument(
"--sticky-receipt",
action="store_true",
help=(
"Like --post-receipt, but maintain ONE comment per PR that is edited in place across loop "
"invocations. State persists in ~/.config/gh-review-loop/state.json "
"(override with GGRL_STATE_DIR env var). Provides background visibility in the PR UI."
),
)
parser.add_argument(
"--receipt-status",
choices=["running", "done", "stopped"],
default=None,
help=(
"Tag the receipt with a status header (RUNNING / DONE / STOPPED). Used with --sticky-receipt "
"to communicate loop phase. Defaults to RUNNING for sticky, none for one-shot."
),
)
parser.add_argument(
"--resolve-addressed-by-reply",
dest="resolve_addressed_by_reply",
action="store_true",
default=True,
help=(
"Resolve unresolved threads where a non-bot maintainer posted a substantive reply "
"(>=%d chars). Enabled by default." % ADDRESSED_BY_REPLY_MIN_CHARS
),
)
parser.add_argument(
"--no-resolve-addressed-by-reply",
dest="resolve_addressed_by_reply",
action="store_false",
help="Leave addressed-by-reply threads unresolved (they remain hidden from actionable output).",
)
parser.add_argument(
"--include-addressed-by-reply",
action="store_true",
help="Include addressed-by-reply threads in actionable output (default: hidden).",
)
parser.add_argument(
"--min-severity",
choices=["critical", "high", "medium", "low"],
default=None,
help=(
"Drop actionable threads below this reviewer-assigned severity. "
"Threads without a recognized severity marker ('unknown') are kept regardless "
"(use --drop-unknown-severity to remove them too)."
),
)
parser.add_argument(
"--drop-unknown-severity",
action="store_true",
help="Drop threads with no severity marker (can be used alone or with --min-severity).",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Do not call any GraphQL mutations; log intended writes to stderr instead.",
)
parser.add_argument(
"--judge-mode",
choices=["off", "on_cycle", "on_complete", "once"],
default=None,
help=(
"Override the saved OpenAI-judge preference for this invocation. "
"Without this flag the script reads ~/.config/gh-review-loop/preferences.json "
"(default: 'off'). Requires an OpenAI API key resolved by key_resolver.py "
"(env var / dotfile / OS keystore); gracefully skips otherwise."
),
)
parser.add_argument(
"--judge-phase",
choices=["cycle", "complete"],
default=None,
help=(
"Which loop phase this invocation represents (the agent supplies this). "
"'cycle' = before fixes each cycle; 'complete' = after the loop stops. "
"Together with --judge-mode, controls whether the judge actually runs."
),
)
parser.add_argument(
"--judge-model",
default=None,
help="Override the OpenAI model used by the judge for this invocation.",
)
parser.add_argument(
"--record-run",
action="store_true",
help=(
"Write one run-metrics record to runs.jsonl and print the [loop] Summary. "
"Use once at loop end. Combine with --fixed-count and --verification."
),
)
parser.add_argument(
"--cycle-summary",
action="store_true",
help=(
"Print the [loop] Summary block for the current cycle from accumulated "
"state, WITHOUT writing a record or clearing run tracking. Safe to call "
"every cycle (unlike --record-run, which is terminal and destructive)."
),
)
parser.add_argument(
"--auto-snapshot",
action="store_true",
help=(
"Render the lean auto-snapshot receipt instead of the full one. For "
"the Stop-hook backstop: shows only GitHub-observable state (threads "
"seen/resolved/open, cycles) and omits agent-only fields (fixed "
"count, verification, outcome) the hook cannot know."
),
)
parser.add_argument("--fixed-count", type=int, default=0, help="Agent-claimed fixes this run.")
parser.add_argument(
"--fixed-finding",
action="append",
default=[],
help=(
"Finding fingerprint the agent fixed locally. Repeatable; stored in "
"run tracking and used for terminal classification."
),
)
parser.add_argument(
"--fixed-path",
action="append",
default=[],
help=(
"Path the agent fixed locally. Repeatable; stored in run tracking and "
"used as path-level fixed/change evidence."
),
)
parser.add_argument(
"--swept-pattern",
action="append",
default=[],
metavar="SIG",
help="Pattern signature (from the Patterns receipt 'sig:' token) the agent "
"swept across changed files this cycle. Repeatable. Accumulates for "
"the convergence advisory.",
)
parser.add_argument(
"--changed-base",
default=None,
help="Optional git base ref for changed-file evidence (`git diff --name-only base..head`).",
)
parser.add_argument(
"--changed-head",
default=None,
help="Optional git head ref for changed-file evidence (`git diff --name-only base..head`).",
)
parser.add_argument(
"--verification",
choices=["passed", "failed", "skipped"],
default="skipped",
help="Result of the verification step this run.",
)
parser.add_argument(
"--verification-details",
default=None,
help="Optional JSON object with structured verification context.",
)
parser.add_argument(
"--outcome",
choices=list(metrics.VALID_OUTCOMES),
default=None,
help="Terminal outcome of the loop. If omitted, derived from state.",
)
parser.add_argument("--outcome-reason", default=None, help="One-line reason for --outcome.")
parser.add_argument(
"--gemini-confirmed",
dest="gemini_confirmed",
action="store_true",
default=True,
help="The final Gemini wait/re-review completed. Default for legacy compatibility.",
)
parser.add_argument(
"--gemini-unconfirmed",
"--no-gemini-confirmed",
dest="gemini_confirmed",
action="store_false",
help="The final Gemini wait timed out or otherwise did not confirm the latest fixes.",
)
parser.add_argument(
"--semantic-risk",
action="append",
default=[],
help=(
"Manual/heuristic semantic risk note for this cycle. Repeatable; "
"rendered only in cycle/run summary output."
),
)
parser.add_argument(
"--record-cycle",
action="store_true",
help=(
"Append one active-cycle timing entry to the run accumulator and exit. "
"Pass --cycle-started-at (when active work for the cycle began), "
"--finding-count, and --cycle-outcome. Measures active work only; waits "
"between cycles are excluded. Does not touch runs.jsonl."
),
)
parser.add_argument(
"--cycle-started-at",
default=None,
help="ISO-8601 UTC timestamp (YYYY-MM-DDThh:mm:ssZ) when this cycle's active work began.",
)
parser.add_argument(
"--finding-count",
type=int,
default=0,
help="Number of actionable findings handled in this cycle.",
)
parser.add_argument(
"--cycle-outcome",
default="continued",
help="Cycle outcome: 'continued' for a non-terminal cycle, else the terminal label.",
)
parser.add_argument(
"--stats",
action="store_true",
help="Print local Gemini-loop stats for this repo from runs.jsonl and exit.",
)
parser.add_argument(
"--stats-window",
type=int,
default=metrics.DEFAULT_WINDOW,
help=f"Number of most-recent runs to aggregate. Default: {metrics.DEFAULT_WINDOW}.",
)
parser.add_argument(
"--stats-all-repos",
action="store_true",
help="Aggregate across all repos instead of only the current one.",
)
args = parser.parse_args()
color_enabled = colors_enabled(no_color=args.no_color)
if args.profile_intro or args.planned_verification:
if not args.repo:
print("error: --profile-intro/--planned-verification require --repo OWNER/REPO.", file=sys.stderr)
return 1
profile = load_profile_for_repo(args.repo)
blocks = []
if args.profile_intro:
blocks.append(metrics.format_profile_intro_block(profile, args.repo))
if args.planned_verification:
blocks.append(metrics.format_planned_verification_block(profile))
print(color_loop_block("\n".join(blocks), enabled=color_enabled))
return 0
if args.wait_heartbeat:
pr = resolve_pr(args.pr)
snapshot = read_wait_state(pr).get("last_snapshot")
if not isinstance(snapshot, dict) or not snapshot:
print(
color_loop_block(
"[loop] no reviewer wait in progress for this PR.",
enabled=color_enabled,
)
)
return 0
print(
color_loop_block(
metrics.format_wait_heartbeat(
str(snapshot.get("status", "")),
author=str(snapshot.get("author", DEFAULT_AUTHOR)),
elapsed_seconds=snapshot.get("elapsed_seconds", 0),
checks=snapshot.get("checks", 0),
next_wait_seconds=snapshot.get("next_wait_seconds", 0),
quiet_period_remaining_seconds=snapshot.get(
"quiet_period_remaining_seconds"
),
)
or "[loop] no reviewer wait in progress for this PR.",
enabled=color_enabled,
)
)
return 0
if args.stats:
try:
if args.pr:
pr = resolve_pr(args.pr)
repo_full = f"{pr.owner}/{pr.repo}"
elif args.stats_all_repos:
repo_full = "(all repos)"
else:
repo_full = resolve_current_repo()
except RuntimeError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
records, skipped = metrics.load_records()
selected = select_stats_records(
records, repo=repo_full, window=args.stats_window, all_repos=args.stats_all_repos
)
agg = metrics.aggregate(selected)
if args.format == "json":
print(json.dumps({"repo": repo_full, "stats": agg, "skipped": skipped}, indent=2, sort_keys=True))
else:
print(metrics.format_stats(repo_full, agg, skipped=skipped))
return 0
if args.record_cycle:
if not args.cycle_started_at:
print("error: --record-cycle requires --cycle-started-at.", file=sys.stderr)
return 1
try:
pr = resolve_pr(args.pr)
except RuntimeError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
try:
cycle = record_cycle(
pr,
started_at=args.cycle_started_at,
finding_count=args.finding_count,
outcome=args.cycle_outcome,
)
except OSError as exc:
# Metrics state I/O is best-effort: warn and exit 0 so a failed
# write never breaks the loop. Return here (do not fall through to
# the success print, which would reference an unbound `cycle`).
print(f"warning: could not record cycle timing: {exc}", file=sys.stderr)
return 0
print(
color_loop_block(
f"[loop] recorded cycle: {metrics.format_duration(cycle['duration_seconds'])} "
f"active ({cycle['finding_count']} finding(s), outcome={cycle['outcome']}).",
enabled=color_enabled,
)
)
return 0
try:
prefs = load_preferences_with_fallback()
args.max_rereview_requests = effective_rereview_limit(
args.max_rereview_requests, prefs
)
pr = resolve_pr(args.pr)
if args.reset_reviewer:
cleared = clear_reviewer_selection(pr)
status = "cleared" if cleared else "not_set"
if args.format == "json":
print(json.dumps({"reviewerSelection": {"status": status}}, indent=2, sort_keys=True))
else:
message = (
"[loop] Reviewer selection cleared for this PR."
if cleared
else "[loop] Reviewer selection was not set for this PR."
)
print(color_loop_block(message, enabled=color_enabled))
return 0
if args.list_reviewers:
discovery = fetch_reviewer_discovery(pr)
pull_request_for_discovery = discovery["pull_request"]
self_login = None if args.no_agent_filter else (args.agent_login or gh_authenticated_login())
candidates = reviewer_resolver.discover_candidates(
pull_request_for_discovery,
self_login=self_login,
)
partial = bool(discovery.get("partial"))
warnings = list(discovery.get("warnings") or [])
if args.format == "json":
print(
json.dumps(
{
"reviewers": [candidate.to_dict() for candidate in candidates],
"partial": partial,
"warnings": warnings,
},
indent=2,
sort_keys=True,
)
)
else:
for warning in warnings:
print(f"warning: {warning}", file=sys.stderr)
if candidates:
for candidate in candidates:
trigger = candidate.review_trigger or "(no safe trigger known)"
print(f"- {candidate.display_name} `{candidate.login}` trigger={trigger}")
elif partial:
print("[loop] Reviewer discovery incomplete; retry or select a reviewer manually.")
else:
print("[loop] No AI reviewer threads found on this PR.")
return 0
reviewer_selection = resolve_reviewer_selection(args, pr)
args.author = reviewer_selection["login"]
args.reviewer_name = reviewer_selection["display_name"]
args.review_trigger_mention = reviewer_selection.get("review_trigger")
if args.fixed_finding or args.fixed_path:
try:
accumulate_fixed_markers(
pr,
fingerprints=args.fixed_finding,
paths=args.fixed_path,
)
except OSError as exc:
print(f"warning: could not persist fixed markers: {exc}", file=sys.stderr)
if args.swept_pattern:
try:
accumulate_swept_patterns(pr, args.swept_pattern)
except OSError as exc:
print(f"warning: could not persist swept patterns: {exc}", file=sys.stderr)
if args.wait and args.wait_chunk_seconds is not None:
if args.wait_chunk_seconds <= 0:
parser.error("--wait-chunk-seconds must be a positive integer.")
chunk = run_wait_chunk(
pr,
args.author,
timeout_seconds=args.timeout,
interval_seconds=args.interval,
quiet_seconds=args.quiet_period,
after_iso=args.after,
chunk_seconds=args.wait_chunk_seconds,
)
if chunk["status"] != "ready":
wait_fields = {
k: v
for k, v in chunk.items()
if k not in ("pull_request", "author") and v is not None
}
if chunk["status"] == "refused":
print_reviewer_refusal(
chunk,
author=args.author,
json_output=args.format == "json",
color_enabled=color_enabled,
)
return 0
if args.format == "json":
print(json.dumps({"wait": wait_fields}, indent=2, sort_keys=True))
else:
print(
color_loop_block(
metrics.format_wait_heartbeat(
chunk["status"],
author=args.author,
elapsed_seconds=chunk.get("elapsed_seconds", 0),
checks=chunk.get("checks", 0),
next_wait_seconds=chunk.get("next_wait_seconds", 0),
quiet_period_remaining_seconds=chunk.get(
"quiet_period_remaining_seconds"
),
),
enabled=color_enabled,
)
)
return 0
pull_request = chunk["pull_request"]
elif args.wait:
try:
pull_request = wait_for_stable_review(
pr,
author=args.author,
timeout_seconds=args.timeout,
interval_seconds=args.interval,
quiet_seconds=args.quiet_period,
after_iso=args.after,
)
except ReviewerRefused as refused:
print_reviewer_refusal(
{"status": "refused", **refused.refusal},
author=args.author,
json_output=args.format == "json",
color_enabled=color_enabled,
)
return 0
except WaitTimedOut as timed_out:
# Mirror the chunked-mode timed_out contract: deterministic
# relayable output, exit 0. Matters most when this wait ran as
# a background task and this is the tail of its captured log.
if args.format == "json":
print(
json.dumps(
{"wait": {
"status": "timed_out",
"elapsed_seconds": timed_out.elapsed_seconds,
}},
indent=2,
sort_keys=True,
)
)
else:
print(
color_loop_block(
metrics.format_wait_heartbeat(
"timed_out",
author=args.author,
elapsed_seconds=timed_out.elapsed_seconds,
),
enabled=color_enabled,
)
)
return 0
else:
pull_request = fetch_threads(pr)
resolved_outdated = 0
resolved_addressed_by_reply = 0
if args.no_agent_filter:
agent_login: str | None = None
else:
agent_login = args.agent_login or gh_authenticated_login()
if agent_login:
print(f"Counting only re-reviews posted by '{agent_login}' toward the cap.", file=sys.stderr)
else:
print(
"warning: could not detect agent login via `gh api user`; "
"falling back to counting all re-review pings.",
file=sys.stderr,
)
rereviews = rereview_requests(
pull_request,
agent_login,
review_trigger_mention=args.review_trigger_mention,
)
_limit_reached = len(rereviews) >= args.max_rereview_requests
if args.resolve_outdated:
resolved_outdated = resolve_outdated_threads(
pull_request, args.author, dry_run=args.dry_run
)
if resolved_outdated:
tag = "[dry-run] would resolve" if args.dry_run else "Resolved"
print(f"{tag} {resolved_outdated} outdated {args.author} thread(s).", file=sys.stderr)
if not args.dry_run:
pull_request = fetch_threads(pr)
if args.resolve_addressed_by_reply:
resolved_addressed_by_reply = resolve_addressed_by_reply(
pull_request, args.author, dry_run=args.dry_run
)
if resolved_addressed_by_reply:
tag = "[dry-run] would resolve" if args.dry_run else "Resolved"
print(
f"{tag} {resolved_addressed_by_reply} addressed-by-reply "
f"{args.author} thread(s).",
file=sys.stderr,
)
if not args.dry_run:
pull_request = fetch_threads(pr)
threads = filter_threads(
pull_request,
author=args.author,
include_resolved=args.include_resolved,
include_outdated=args.include_outdated,
include_addressed_by_reply=args.include_addressed_by_reply,
)
# Findings the reviewer wrote into the review body have no thread to
# fetch, so they are appended as synthetic current feedback.
threads.extend(review_body_findings(pull_request, args.author))
threads = sort_by_severity(threads)
if not args.record_run and not args.stats:
try:
update_run_tracking(
pr,
[(t["id"], t.get("path", "")) for t in threads],
bump_session_cycle=not args.cycle_summary,
)
except OSError as exc:
print(f"warning: could not update run tracking: {exc}", file=sys.stderr)
if args.min_severity or args.drop_unknown_severity:
before = len(threads)
threads = filter_by_min_severity(
threads, args.min_severity, keep_unknown=not args.drop_unknown_severity
)
dropped = before - len(threads)
if dropped:
if args.min_severity:
tag = " (kept unknown-severity)" if not args.drop_unknown_severity else ""
msg = f"--min-severity {args.min_severity}: dropped {dropped} thread(s){tag}."
else:
msg = f"--drop-unknown-severity: dropped {dropped} unknown-severity thread(s)."
print(msg, file=sys.stderr)
page_warnings = pagination_warnings(pull_request)
for warning in page_warnings:
print(f"warning: {warning}", file=sys.stderr)
# ---- No-progress detection ------------------------------------------
# Compare the actionable thread fingerprint to the previous cycle's.
# Only runs during a real fetch (not --cycle-summary / --record-run /
# --stats), so the comparison reflects actual loop cycles, not
# mid-cycle read-only calls.
no_progress_flag = False
if not args.record_run and not args.cycle_summary and not args.stats:
current_fp = thread_fingerprint(threads)
try:
no_progress_flag = detect_no_progress(pr, current_fp)
except OSError as exc:
print(f"warning: could not check progress: {exc}", file=sys.stderr)
# Snapshot prior-cycle finding fingerprints so the receipt can mark
# carried-over findings. Real-fetch only, once per cycle.
try:
track_finding_fingerprints(
pr, {finding_fingerprint(t) for t in threads if isinstance(t, dict)}
)
except OSError as exc:
print(f"warning: could not track finding fingerprints: {exc}", file=sys.stderr)
try:
track_pattern_signatures(
pr,
{cluster_findings.pattern_signature(t) for t in threads if isinstance(t, dict)}
- {""},
)
except OSError as exc:
print(f"warning: could not track pattern signatures: {exc}", file=sys.stderr)
# ---- Judge (optional, opt-in via prefs file or --judge-mode) -------
# The script is the single source of truth for whether the judge
# runs. The agent supplies --judge-phase; we read the saved mode
# from ~/.config/gh-review-loop/preferences.json (or
# --judge-mode override) and decide.
try:
from judge import ( # noqa: PLC0415
JudgeClient, JudgeError, should_judge_run, skipped_result,
)
judge_available = True
except ImportError:
judge_available = False
judge_results: dict[str, dict[str, Any]] = {}
judge_status = {
"ran": False,
"skip_reason": "judge.py not importable (plugin install path issue)",
}
if judge_available:
effective_mode = (
args.judge_mode
if args.judge_mode is not None
else prefs.get("judge_mode", "off")
)
effective_phase = resolve_judge_phase(
args.judge_phase, record_run=bool(args.record_run)
)
judge_results = {}
if should_judge_run(mode=effective_mode, phase=effective_phase):
client = JudgeClient(model=args.judge_model or prefs["judge_model"])
ready, skip_reason = client.is_ready()
if not ready:
judge_status = {
"ran": False,
"mode": effective_mode,
"phase": effective_phase,
"skip_reason": skip_reason,
}
print(
f"info: judge skipped — {skip_reason}. "
"Loop continues unchanged.",
file=sys.stderr,
)
else:
judge_status = {
"ran": True,
"mode": effective_mode,
"phase": effective_phase,
"model": client.model,
"judged_count": 0,
"errors": 0,
}
for thread in threads:
# Build a single judge input from the first matching
# bot comment in this filtered thread.
body = (thread.get("comments") or [{}])[0].get("body") or ""
finding_payload = {
"severity": thread_severity(thread),
"path": thread.get("path"),
"line": thread.get("line") or thread.get("originalLine"),
"body": body,
"diff_hunk": (thread.get("comments") or [{}])[0].get("diffHunk") or "",
}
try:
jr = client.judge(finding_payload)
judge_results[thread["id"]] = dataclasses.asdict(jr)
if jr.status == "ok":
judge_status["judged_count"] += 1
except JudgeError as exc:
judge_results[thread["id"]] = dataclasses.asdict(
skipped_result(f"judge error: {exc}")
)
judge_status["errors"] += 1
else:
judge_status = {
"ran": False,
"mode": effective_mode,
"phase": effective_phase,
"skip_reason": (
f"mode={effective_mode!r} + phase={effective_phase!r} → no-op"
),
}
# --------------------------------------------------------------------
# Persist this cycle's verdicts into the run accumulator so the terminal
# --record-run can report the eval even after findings are resolved.
if not args.record_run and not args.stats and judge_results:
try:
accumulate_judge_results(pr, judge_results)
except OSError as exc:
print(f"warning: could not persist judge results: {exc}", file=sys.stderr)
rereviews = rereview_requests(
pull_request,
agent_login,
review_trigger_mention=args.review_trigger_mention,
)
if len(rereviews) >= args.max_rereview_requests:
print(
f"warning: {len(rereviews)} reviewer re-review request(s) already exist; "
f"the configured loop cap is {args.max_rereview_requests}.",
file=sys.stderr,
)
if args.record_run or args.cycle_summary:
try:
# --record-run owns accumulation here because it skips the
# per-fetch tracking above; --cycle-summary just reads what that
# path already accumulated this cycle.
if args.record_run:
update_run_tracking(
pr,
[(t["id"], t.get("path", "")) for t in threads],
bump_session_cycle=False,
)
run = read_run_tracking(pr)
except OSError as exc:
print(f"warning: could not update or read run tracking: {exc}", file=sys.stderr)
run = {}
baseline_ids = set(run.get("finding_ids", []))
finding_paths = run.get("finding_paths", [])
current_actionable_ids = {t["id"] for t in threads}
addressed_by_reply_ids = {
t["id"] for t in addressed_by_reply_threads(pull_request, args.author)
}
# Merge verdicts accumulated across cycles with this invocation's,
# so the record reflects the whole run even when the terminal pass
# has no live findings to re-judge. Current invocation supersedes.
if not isinstance(run, dict):
run = {}
accumulated_results = run.get("judge_results")
if not isinstance(accumulated_results, dict):
accumulated_results = {}
merged_judge_results = merge_judge_results(accumulated_results, judge_results)
judge_ran = bool(run.get("judge_ran")) or bool(judge_status.get("ran"))
cap_reached = len(rereviews) >= args.max_rereview_requests
fixed_markers = read_fixed_markers(pr)
changed_paths = changed_files_in_range(args.changed_base, args.changed_head)
remaining_states = classify_remaining_finding_states(
threads,
fixed_fingerprints=fixed_markers["fingerprints"],
fixed_paths=fixed_markers["paths"],
changed_paths=changed_paths,
prior_fingerprints=prior_finding_fingerprints(pr),
judge_results=merged_judge_results,
cap_reached=cap_reached,
)
likely_fixed_remaining = count_likely_fixed_remaining(remaining_states)
outcome = args.outcome or _derive_outcome(
len(current_actionable_ids),
args.verification,
cap_reached,
gemini_confirmed=args.gemini_confirmed,
likely_fixed_remaining=likely_fixed_remaining,
)
derived = derive_record_fields(
baseline_ids=baseline_ids,
current_actionable_ids=current_actionable_ids,
addressed_by_reply_ids=addressed_by_reply_ids,
outcome=outcome,
judge_ran=judge_ran,
judge_results=merged_judge_results,
)
terminal_breakdown = {
"confirmed_fixed_outdated": derived["observed_fixed_count"],
"fixed_pending_confirmation": likely_fixed_remaining,
"remaining_valid_actionable": max(
0,
derived["remaining_actionable"]
- derived["needs_human"]
- likely_fixed_remaining,
),
"needs_human": derived["needs_human"],
}
verification_details: dict[str, Any] = {}
if args.verification_details:
try:
verification_details = json.loads(args.verification_details)
except json.JSONDecodeError:
print(
"warning: --verification-details is not valid JSON; storing {}.",
file=sys.stderr,
)
# Pass the repo root so clustering can merge findings the reviewer
# worded differently but which anchor to the same code shape. Falls
# back to prose-only clustering when the root cannot be resolved.
clusters = cluster_findings.cluster(
[t for t in threads if isinstance(t, dict)],
root=repo_root_for_pr(pull_request),
)
# Cross-run history: --record-run clears the live accumulator, so a
# resumed loop folds in pattern signatures recorded by prior runs of
# this PR (runs.jsonl) — otherwise recurrence resets to 0 on resume.
# Canonical on both sides: records are matched by exact repo
# string, so the read key and the write key must agree regardless
# of how the caller spelled --pr.
metrics_repo = canonical_repo_full(pull_request, pr)
history = metrics.pattern_history_for_pr(metrics_repo, pr.number)
convergence = build_convergence(pr, clusters, history)
conv_stats = convergence["stats"]
record = metrics.build_record(
repo=metrics_repo,
pr=pr.number,
provider=args.author,
fixed_count=args.fixed_count,
cycles_used=len(rereviews),
cycle_cap=args.max_rereview_requests,
verification=args.verification,
verification_details=verification_details,
outcome=outcome,
outcome_reason=args.outcome_reason or f"outcome: {outcome}",
session_cycle=_safe_int(run.get("session_cycle")) or None,
started_at=run.get("started_at"),
finding_paths=finding_paths,
judge=metrics.build_judge_block(judge_ran, merged_judge_results),
cycles=run.get("cycles", []),
terminal_breakdown=terminal_breakdown,
patterns={
"distinct_patterns": conv_stats["distinct_patterns"],
"max_cluster_size": max((c.count for c in clusters), default=0),
"pattern_recurrence_rate": round(conv_stats["recurrence_rate"], 3),
"swept_count": convergence["swept_count"],
# Persisted so later runs of this PR can compute cross-run
# recurrence after --record-run clears the live accumulator.
# Only this run's own sweeps: pattern_history_for_pr unions
# across records, so writing the union back here would
# compound it into every later record and the count could
# never fall again.
"signatures": sorted({c.signature for c in clusters}),
"swept": convergence["swept"],
} if clusters else None,
**derived,
)
# Read the carried-over classification BEFORE the terminal branch:
# --record-run calls clear_run_tracking(), which deletes the very
# run block prior_finding_fingerprints() reads. Reading it after
# would mark every finding on a capped/human/stopped receipt "new".
prior_fps = prior_finding_fingerprints(pr)
# --cycle-summary is read-only: print the block but never append a
# record or clear the accumulator, so it is safe to call every cycle.
if args.record_run:
try:
metrics.append_record(record)
except OSError as exc:
print(f"warning: could not record run metrics: {exc}", file=sys.stderr)
else:
try:
clear_run_tracking(pr)
except OSError as exc:
print(f"warning: could not clear run tracking: {exc}", file=sys.stderr)
else:
# --cycle-summary leaves the accumulator intact but stamps that a
# summary covering the current run state was emitted, so the
# Stop-hook backstop won't duplicate it.
try:
stamp_summary_emitted(pr)
except OSError as exc:
print(f"warning: could not stamp summary state: {exc}", file=sys.stderr)
if args.auto_snapshot:
print(color_loop_block(metrics.format_auto_snapshot(record), enabled=color_enabled))
return 0
# Single-channel receipt delivery (#86): the full receipt goes to
# the sticky PR comment, chat gets a one-line pointer. stdout keeps
# the full receipt only as the fail-open fallback (dry-run, or the
# comment write failed) so a receipt can never be lost.
summary_block = metrics.format_run_summary(record, terminal=bool(args.record_run))
semantic_risk_block = metrics.format_semantic_risk_block(args.semantic_risk)
suite_block = metrics.format_suite_block(verification_details)
patterns_block = metrics.format_patterns_block(clusters)
clustering_advisory = metrics.format_degenerate_clustering_advisory(clusters)
convergence_line = convergence["line"]
findings_view = []
for thread in threads:
comments = _iter_comments(thread)
line_val = thread.get("line")
findings_view.append({
"path": thread.get("path"),
"line": line_val if line_val is not None else thread.get("originalLine"),
"severity": thread_severity(thread),
"url": comments[0].get("url") if comments and isinstance(comments[0], dict) else None,
"carried": finding_fingerprint(thread) in prior_fps,
})
findings_block = metrics.format_findings_block(findings_view)
receipt_url = None
if not args.dry_run:
receipt_status = "RUNNING" if not args.record_run else (
"DONE" if record.get("outcome") == "clean" else "STOPPED"
)
body = build_receipt_comment_body(
[
summary_block,
semantic_risk_block,
suite_block,
patterns_block,
clustering_advisory,
convergence_line,
],
status=receipt_status,
findings_block=findings_block,
)
try:
comment_id = post_or_update_sticky_receipt(pr, body)
except (RuntimeError, OSError) as exc:
comment_id = None
print(
f"warning: could not deliver the receipt to the PR comment: {exc}. "
"Printing the full receipt instead.",
file=sys.stderr,
)
if comment_id:
receipt_url = (
f"https://github.com/{pr.owner}/{pr.repo}/pull/"
f"{pr.number}#issuecomment-{comment_id}"
)
if receipt_url:
new_count = sum(1 for f in findings_view if not f.get("carried"))
print(
color_loop_block(
metrics.format_compact_receipt_line(
record,
terminal=bool(args.record_run),
receipt_url=receipt_url,
findings_new=new_count,
findings_carried=len(findings_view) - new_count,
),
enabled=color_enabled,
)
)
if judge_eval_requested(judge_status):
print(
color_loop_block(
metrics.format_judge_skip(judge_status.get("skip_reason", "")),
enabled=color_enabled,
)
)
# The clustering advisory changes what the agent must do BEFORE
# fixing (manual sweep), so it stays in chat alongside the
# pointer; everything else lives in the PR comment.
if clustering_advisory:
print(clustering_advisory)
else:
print(color_loop_block(summary_block, enabled=color_enabled))
if judge_eval_requested(judge_status):
print(
color_loop_block(
metrics.format_judge_skip(judge_status.get("skip_reason", "")),
enabled=color_enabled,
)
)
if semantic_risk_block:
print(color_loop_block(semantic_risk_block, enabled=color_enabled))
if suite_block:
print(suite_block)
# Printed directly (like suite_block / findings_block): these
# start with "Patterns ("/"Convergence:", not "[loop]", so
# color_loop_block would be a no-op wrap.
if patterns_block:
print(patterns_block)
if clustering_advisory:
print(clustering_advisory)
if convergence_line:
print(convergence_line)
if findings_block:
print(findings_block)
return 0
if args.post_receipt or args.sticky_receipt:
sticky = args.sticky_receipt
status = args.receipt_status.upper() if args.receipt_status else (
"RUNNING" if sticky else None
)
receipt = render_receipt(
pr, pull_request, threads,
author=args.author,
resolved_outdated=resolved_outdated,
resolved_addressed_by_reply=resolved_addressed_by_reply,
rereview_count=len(rereviews),
rereview_limit=args.max_rereview_requests,
status=status,
sticky=sticky,
)
if sticky:
post_or_update_sticky_receipt(pr, receipt, dry_run=args.dry_run)
else:
post_pr_comment(pr, receipt, dry_run=args.dry_run)
except RuntimeError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
# Compute once for both human and machine fetch output. The advisory is a
# pre-fix ordering guard, so JSON automation must receive it too.
fetch_clusters = cluster_findings.cluster(
[thread for thread in threads if isinstance(thread, dict)],
root=repo_root_for_pr(pull_request),
)
clustering_advisory = metrics.format_degenerate_clustering_advisory(fetch_clusters)
# Delta collapse baseline (#95): thread_id → render_fp emitted last cycle.
# Read fails open to {} (everything renders full — the safe direction).
try:
render_baseline = read_render_baseline(pr, args.author)
except OSError as exc:
print(f"warning: could not read render baseline: {exc}", file=sys.stderr)
render_baseline = {}
current_render_fps: dict[str, str] = {}
for thread in threads:
thread_id = thread.get("id")
if isinstance(thread_id, str):
jr = judge_results.get(thread_id) if judge_results else None
current_render_fps[thread_id] = thread_render_fingerprint(thread, jr)
# Profile blocks ride along with the fetch (#99) so the agent doesn't pay
# two extra script invocations per cycle for --profile-intro and
# --planned-verification.
#
# They can only describe a profile that already exists. On a first run the
# decision happens after this fetch, so blocks built here would say "none
# saved / ad hoc" while the agent goes on to save a profile and run it —
# the relayed text would contradict the suite actually used (#107). Emit
# the regenerate notice for that path and keep the standalone flags as the
# post-decision source. A `skipped` or malformed profile is *not* this
# case: a decision exists, and its fallback wording is the final answer.
profile_repo = canonical_repo_full(pull_request, pr)
try:
saved_profile = load_profile_for_repo(profile_repo)
except Exception: # pragma: no cover - load_profile_for_repo already swallows
saved_profile = None
profile_blocks_provisional = saved_profile is None
if profile_blocks_provisional:
profile_intro_block = metrics.format_pending_profile_block(
profile_repo, script=str(Path(__file__).resolve())
)
planned_verification_block = ""
else:
profile_intro_block = metrics.format_profile_intro_block(saved_profile, profile_repo)
planned_verification_block = metrics.format_planned_verification_block(saved_profile)
if args.format == "json":
# The machine path stays lossless: full threads, annotated with the
# delta bit. It never commits the baseline — only markdown emission
# (what the agent actually consumed) moves it.
threads_payload = []
for thread in threads:
copied = dict(thread)
thread_id = thread.get("id")
copied["changedSinceLastCycle"] = not (
isinstance(thread_id, str)
and render_baseline.get(thread_id) == current_render_fps.get(thread_id)
)
threads_payload.append(copied)
print(
json.dumps(
{
"pullRequest": pull_request,
"threads": threads_payload,
"clustering": {
"clusterCount": len(fetch_clusters),
"maxClusterSize": max(
(cluster.count for cluster in fetch_clusters), default=0
),
"advisory": clustering_advisory,
},
"reviewerSelection": reviewer_selection,
"loopStatus": {
# Script-owned narration ordinal (#104): copy it into
# "session cycle N" lines; never model-compute it.
"sessionCycle": _safe_int(
read_run_tracking(pr).get("session_cycle")
) or None,
"reReviewRequests": len(rereviews),
"reReviewLimit": args.max_rereview_requests,
"limitReached": len(rereviews) >= args.max_rereview_requests,
"agentLogin": agent_login,
"unresolvedOutdatedThreads": len(outdated_unresolved_threads(pull_request, args.author)),
"addressedByReplyThreads": len(addressed_by_reply_threads(pull_request, args.author)),
"resolvedOutdatedThreads": resolved_outdated,
"resolvedAddressedByReplyThreads": resolved_addressed_by_reply,
"severityCounts": severity_counts(threads),
"pageLimitWarnings": page_warnings,
"dryRun": args.dry_run,
"actionableThreadFingerprint": thread_fingerprint(threads),
"noProgressDetected": no_progress_flag,
"judge": judge_status,
},
"judgeResults": judge_results,
"humanBlocks": {
"profileIntro": profile_intro_block,
# Empty (not a fallback string) while provisional, so
# automation cannot relay a suite that has not been
# chosen yet.
"plannedVerification": planned_verification_block,
"profileBlocksProvisional": profile_blocks_provisional,
},
},
indent=2,
sort_keys=True,
)
)
else:
# Script-owned narration ordinal (#104): the agent copies this N into
# its "session cycle N" lines instead of tracking a counter that
# drifts across context compaction.
session_cycle_n = _safe_int(read_run_tracking(pr).get("session_cycle"))
if session_cycle_n:
print(
color_loop_block(
f"[loop] session cycle {session_cycle_n}", enabled=color_enabled
)
)
print(
render_markdown(
pull_request,
threads,
args.author,
reviewer_name=args.reviewer_name,
review_trigger_mention=args.review_trigger_mention,
judge_results=judge_results,
baseline=render_baseline,
full=args.full,
),
end="",
)
# The clustering warning is an ordering guard, so show it on the
# initial fetch while the agent is deciding what to fix. Summary-mode
# output repeats it later as an audit trail.
print_clustering_blocks(fetch_clusters)
if judge_status.get("ran") and judge_results:
print(
color_loop_block(
format_judge_thread_table(
threads, judge_results, judge_status.get("phase", "cycle")
),
enabled=color_enabled,
),
)
elif judge_eval_requested(judge_status):
print(
color_loop_block(
metrics.format_judge_skip(judge_status.get("skip_reason", "")),
enabled=color_enabled,
)
)
if no_progress_flag:
print(
color_loop_block(
"[loop] no_progress: actionable thread set is unchanged since the previous "
"cycle — no fix landed. Stop with: "
"--record-run --outcome no_progress "
"--outcome-reason 'no code change resolved any open thread'",
enabled=color_enabled,
)
)
print(color_loop_block(profile_intro_block, enabled=color_enabled))
if planned_verification_block:
print(color_loop_block(planned_verification_block, enabled=color_enabled))
# Commit the delta baseline ONLY after the rendered output has been
# emitted (#95): render → print → flush → persist. Persisting first
# would collapse threads the agent never saw if this process dies
# mid-emission. Save failure fails open — next cycle renders full.
sys.stdout.flush()
try:
save_render_baseline(pr, args.author, current_render_fps)
except OSError as exc:
print(f"warning: could not save render baseline: {exc}", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 399257f25a683d04b0adef718e1f6e575905476fba3ff01e9e9b63737296a0fd