Files
cleveragents-core/tools/_pr_classification_cache.py
T
drew 2658deee94 feat(auto-agents): PR State Warmer substrate + supporting infra
Adds a long-lived sidecar (pr_state_warmer.py) that polls Forgejo's
/pulls endpoint every 30s and writes the full PR snapshot to a
shared SQLite store, eliminating the dispatcher's per-cycle
cold-cache stalls (24-30s rebuilds on flaky cycles) and the silent
50-PR pagination cap on the legacy single-page fetch.

Substrate
- tools/_pr_state_cache.py  — SQLite store with (owner, repo) PK,
  WAL mode, additive v2→v3 migration (comments_refreshed_updated_at),
  bounded fcntl.flock migration lock, threading.Lock for per-process
  init, @_with_reheal decorator (catches OperationalError no-such-
  table + DatabaseError corruption with file quarantine), atomic
  TEMP-table chunking for >32k seen-set, _normalize_updated_at to
  canonicalize Forgejo tz-marker drift
- tools/pr_state_warmer.py — poll/upsert/vanish/comments-refresh
  loop with fcntl.flock singleton (rejects second warmer), bounded
  comments-refresh cap, persistent deferral via SQL pending query,
  PermissionError-tolerant lock setup, cold-start log suppression
- tools/_pr_classification_cache.py — three-layer fall-through
  (warmer cache → list cache → live fetch) with staleness gate
  (PR_STATE_WARMER_STALE_AFTER_S floored at 30s in prod)

Comments cache hardening
- Bot-filter at write time drops bot status/claim/release/sentinel
  while preserving **Implementation Attempt** markers (94.6%
  reduction on bot-heavy PRs like #30's 19k-comment thread)
- _normalize_since_cursor strips microsecond precision before
  building ?since= query (fixes the live-observed Forgejo HTTP 422
  bug on PRs #25 + #28); handles uppercase Z, lowercase z, ±HH:MM
  offsets (including non-zero like +05:30), naive ISO
- Lazy migration of legacy null-key by_author entries on _read_cache
- _newest_cursor walks tail-back skipping malformed entries

Supporting infrastructure (cumulative dmpipeline-v2 work)
- Telemetry server: SSE live tail, run-sessions enumeration,
  cost/token tracking, app.js UI rewrite with collapsible sections
- MCP servers (mcp_ci_server, mcp_forgejo_server, mcp_git_server,
  mcp_handoff_server, mcp_graphify_server) for opencode worker
  context access
- Live log writer (tools/live_log_writer.py) — SSE-streaming
  dispatcher event log
- Tier-dispatcher escalation flow with prompts trimmed for budget
- Shared bot-logins resolver (tools/_bot_logins.py) replacing two
  drift-prone copies
- token_usage_audit.py for opencode cost analysis

Tests
- 2259 passing across 65 changed/new files
- New suites: test_pr_state_cache, test_pr_state_warmer,
  test_pr_state_warmer_integration, test_pr_classification_cache,
  test_pr_list_cache_backoff, test_mcp_* (5 servers),
  test_live_log_writer_sse, test_telemetry_run_sessions,
  test_review_post_ready_label
- Test_pr_comments_cache expanded with bot-filter coverage,
  cursor-normalization regression pins, format-drift, atomicity,
  failed-comments-not-stamped (silent-data-loss class)
- Parametrized @_with_reheal coverage across 7 wrapped APIs
- Real fault-inject atomicity test for chunked mark_vanished path
  via Connection wrapper class
- Subprocess-based singleton flock test (cross-process contract)
- Event-driven SIGTERM-mid-poll test (no fixed-sleep flake)

Architecture notes
- Schema v3 migration is additive (ALTER ADD COLUMN); v0/v1 still
  need destructive rebuild because pre-v2 column shape lacks
  owner/repo. Cross-process drop-table-ping-pong prevented by the
  fcntl migration lock + per-process _initialized flag.
- Comments-refresh deferral is persistent via
  comments_refreshed_updated_at column — survives warmer restart,
  picks up next cycle even if PR didn't change again. Replaces
  in-memory changed_numbers list.
- Rollback path: PR_STATE_WARMER_PREFER=0 bypasses the warmer
  cache and reverts to live-fetch behavior. PR_STATE_CACHE_DISABLE=1
  short-circuits the warmer process at startup.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-17 10:02:53 -04:00

785 lines
30 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Delta-cached PR enumeration for the reviewer dispatcher.
Replaces the 5 per-cycle ``list_prs_*.ts`` subprocess calls (the
flaky path that timed out at 120 s in runs 14 / 15 / 16) with native
Python that:
- Fetches the open-PR list ONCE per cycle, sorted by ``updated_at``
descending.
- For each PR, checks the on-disk cache (``ForgejoCache.pr_classifications``):
if ``head_sha`` matches, ``updated_at`` hasn't advanced, schema
version matches, and last-checked is within TTL → reuse the cached
classification with ZERO per-PR API calls.
- Re-classifies (fetching CI status + reviews + commits) only for
PRs that actually changed since the last check.
- Applies the 5 reviewer-filter predicates to all classifications.
See ``.drew/planning/fix list_prs_by_filter.md`` for the full plan,
including phased migration (Phase 1 ships this module + the MCP tool
with the dispatcher unchanged; Phase 2 cuts the dispatcher over
behind a feature flag).
Cache hit predicate (all four must hold):
cached.head_sha == pr.head.sha
cached.updated_at >= pr.updated_at
cached.classification_schema == current
now - cached.last_checked_at < ttl_seconds
The schema_version axis is the lever for invalidating the entire
cache en masse when a new classification axis is added (e.g. adding
``has_unmerged_dependency`` would bump from v1 → v2 and all v1 rows
would cache-miss-and-refetch on first read).
This module is pure logic over the cache + the existing
``_review_fetch`` helpers — no agent-side I/O, no LLM calls.
"""
from __future__ import annotations
import datetime as _dt
import json
import logging
import os
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _loader import load_sibling # noqa: E402
_claim_runtime = load_sibling("_claim_runtime", "_claim_runtime.py")
_review_fetch = load_sibling("_review_fetch", "_review_fetch.py")
_forgejo_cache = load_sibling("_forgejo_cache", "_forgejo_cache.py")
_backoff = load_sibling("_backoff", "_backoff.py")
_pr_state_cache = load_sibling("_pr_state_cache", "_pr_state_cache.py")
_logger = logging.getLogger(__name__)
# The 5 reviewer-side filter names. Lifted verbatim from
# ``dispatch_review._make_work_group`` calls so a typo here surfaces
# as a clean ValueError, not a silent no-match.
FILTER_NAMES = (
"addressed_changes_ci_passing",
"addressed_changes_ci_failing",
"no_active_review_ci_passing",
"no_active_review_ci_failing",
"missing_ci_checks",
)
DEFAULT_TTL_SECONDS = 300
# Forgejo /pulls flakiness resilience (R1, 2026-05-16). Live tests
# saw the ``/repos/{owner}/{repo}/pulls?state=open&...`` endpoint
# time out at the default 30 s request_timeout_s on roughly 1 in 5
# cycles (4 timeouts observed in run-18 alone). Each timeout
# triggered a 30 s wait followed by a fallback to the legacy
# ``npx --yes tsx list_prs_*.ts`` script that ALSO times out at
# 120 s in npx-cold-start conditions, costing 30-150 s per affected
# cycle. Two cheap fixes:
#
# - **Shorter per-call timeout** (default 8 s) for the listing
# fetch. Forgejo serves a clean list in <500 ms on the happy
# path; 8 s is plenty of headroom while bounding the worst case
# so we fall back fast.
# - **Last-good listing cache + exponential backoff** so the next
# N cycles inside the backoff window short-circuit to the cached
# listing instead of paying the timeout again.
# 30s default chosen after run-21 probe (2026-05-16): Forgejo's
# ``/pulls`` endpoint takes ~24s on a cold cache because the response
# inlines full ``head``/``base`` repo+owner metadata (90 KB of 127 KB
# total for 8 PRs). The original 8s default fail-fast caused the
# dispatcher to time out before the cold-cache rebuild completed,
# triggering backoff while Forgejo's already-in-flight rebuild ran
# to completion server-side. Bumping to 30s lets a single attempt
# succeed; subsequent attempts within Forgejo's hot-cache window
# return in <1s.
_LIST_TIMEOUT_S_DEFAULT = 30
_LIST_TIMEOUT_ENV = "REVIEW_DISPATCHER_PR_LIST_TIMEOUT_S"
_LIST_CACHE_DIR_DEFAULT = "/tmp/cleveragents-pr-list-cache"
_LIST_CACHE_DIR_ENV = "REVIEW_DISPATCHER_PR_LIST_CACHE_DIR"
_LIST_BACKOFF = _backoff.Backoff(
base_env="REVIEW_DISPATCHER_PR_LIST_BACKOFF_BASE_S",
max_env="REVIEW_DISPATCHER_PR_LIST_BACKOFF_MAX_S",
base_default=60,
max_default=1800,
)
_LIST_DISABLE_ENV = "REVIEW_DISPATCHER_PR_LIST_CACHE_DISABLE"
_LIST_CACHE_SCHEMA_VERSION = 1
# Label names that mark a PR as claimed by another agent. Mirrors
# ``list_prs.ts``'s CLAIM_LABELS set. Coordinated with
# ``tools/setup_auto_labels.py`` and the worker-side ``claim_pr.ts``;
# mutating this set is a breaking change against the 4 agents that
# read these labels.
CLAIM_LABELS = frozenset(
(
"auto/claimed-merge",
"auto/claimed-implementer",
"auto/claimed-reviewer",
)
)
def refresh_then_filter(
cfg: Any,
filter_name: str,
ttl_seconds: int = DEFAULT_TTL_SECONDS,
cache: Any = None,
) -> list[dict[str, Any]]:
"""Return open PRs matching ``filter_name``, populating the cache
as needed.
Order is the order Forgejo returned (sorted by ``updated_at`` desc
— most-recently-touched PRs first), so the most-actionable PRs
surface first in the dispatcher's iteration.
Pass ``cache`` to share a connection (the dispatcher should open
one per cycle to avoid SQLite open/close overhead); omit it for
standalone calls (e.g. from the MCP tool) and the function will
open/close its own.
"""
if filter_name not in FILTER_NAMES:
raise ValueError(
f"unknown filter {filter_name!r}; "
f"valid: {list(FILTER_NAMES)}"
)
own_cache = cache is None
if own_cache:
cache = _forgejo_cache.ForgejoCache.open()
try:
prs = _list_open_prs_sorted_by_updated(cfg)
results: list[dict[str, Any]] = []
for pr in prs:
try:
classification = _ensure_classified(
cache, cfg, pr, ttl_seconds,
)
except Exception as exc: # noqa: BLE001
# Per-PR classification failure (transient API blip
# on this specific PR's reviews/CI) does NOT abort the
# whole enumeration — log + skip the PR. Better to
# return a slightly-incomplete list than to fail the
# whole reviewer cycle on one bad PR.
_logger.warning(
"classify PR #%s failed; skipping for this cycle: %s",
pr.get("number"), exc,
)
continue
if _evaluate_filter(filter_name, classification):
results.append(_project_pr(pr, classification))
return results
finally:
if own_cache:
cache.close()
def _list_timeout_s() -> int:
raw = os.environ.get(_LIST_TIMEOUT_ENV)
if raw:
try:
return max(1, int(raw))
except ValueError:
pass
return _LIST_TIMEOUT_S_DEFAULT
def _list_cache_dir() -> Path:
return Path(os.environ.get(_LIST_CACHE_DIR_ENV) or _LIST_CACHE_DIR_DEFAULT)
def _list_cache_path(cfg: Any) -> Path:
"""Per-(owner, repo) cache file. We sanitize defensively (the
same path-traversal check the ci-logs cache uses) so a malformed
owner/repo couldn't be persuaded to write outside the cache dir."""
import re as _re
owner = _re.sub(r"[^a-zA-Z0-9_.-]", "", str(getattr(cfg, "owner", "")))[:64]
repo = _re.sub(r"[^a-zA-Z0-9_.-]", "", str(getattr(cfg, "repo", "")))[:64]
return _list_cache_dir() / f"{owner or 'INVALID'}.{repo or 'INVALID'}.json"
def _list_cache_disabled() -> bool:
raw = os.environ.get(_LIST_DISABLE_ENV, "").strip().lower()
return raw in {"1", "true", "yes", "on"}
# Thin wrappers preserved for the existing test suite + any direct
# imports; delegate to the shared :class:`_backoff.Backoff`.
def _list_compute_next_attempt_after(
failures: int, now_dt: _dt.datetime,
) -> str | None:
return _LIST_BACKOFF.next_attempt_after(failures, now_dt)
def _list_backoff_active(
cached: dict[str, Any] | None, now_dt: _dt.datetime,
) -> bool:
return _LIST_BACKOFF.is_active(cached, now_dt)
def _read_list_cache(cfg: Any) -> dict[str, Any] | None:
path = _list_cache_path(cfg)
if not path.exists():
return None
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
_logger.warning(
"PR-list cache read failed for %s: %s", path, exc,
)
return None
if not isinstance(payload, dict):
return None
if payload.get("schema_version") != _LIST_CACHE_SCHEMA_VERSION:
return None
if not isinstance(payload.get("prs"), list):
return None
return payload
def _write_list_cache(cfg: Any, payload: dict[str, Any]) -> None:
path = _list_cache_path(cfg)
tmp = path.with_suffix(path.suffix + ".tmp")
try:
path.parent.mkdir(parents=True, exist_ok=True)
tmp.write_text(
json.dumps(payload, indent=2, default=str),
encoding="utf-8",
)
tmp.replace(path)
except (OSError, TypeError, ValueError) as exc:
_logger.warning(
"PR-list cache write failed for %s: %s", path, exc,
)
try:
tmp.unlink(missing_ok=True)
except OSError:
pass
def _cfg_with_short_timeout(cfg: Any) -> Any:
"""Shallow shadow of cfg with a shorter ``request_timeout_s`` for
the /pulls listing only. Keeps the rest of the dispatcher's call
surface on the standard (longer) timeout. Returns the original
cfg unchanged when no shortening is in effect.
Uses ``copy.copy`` rather than a ``dir(cfg)`` walk: the latter
triggered every descriptor / property getter for side effects we
don't want, and was slower than the stdlib copy anyway."""
import copy as _copy
short = _list_timeout_s()
current = int(getattr(cfg, "request_timeout_s", 30) or 30)
if short >= current:
return cfg
shadow = _copy.copy(cfg)
try:
shadow.request_timeout_s = short
except (AttributeError, TypeError):
# AttributeError: slotted shape with no __dict__.
# TypeError: frozen dataclass raises dataclasses.FrozenInstanceError
# which subclasses AttributeError on 3.13 but TypeError on
# older builds — catch the union for safety. Fall back to
# the dir-walk reconstruction for both cases.
attrs = {
k: getattr(cfg, k) for k in dir(cfg)
if not k.startswith("_")
and not callable(getattr(cfg, k, None))
}
attrs["request_timeout_s"] = short
shadow = SimpleNamespace(**attrs)
return shadow
_WARMER_PREFER_ENV = "PR_STATE_WARMER_PREFER"
# How old the warmer's latest write can get before we treat the cache
# as stale and fall through to a live fetch. Default 5 min ≈ 10× the
# warmer's 30s interval — generous enough to absorb a slow warmer
# cycle without false-positive staleness, tight enough to catch a
# dead warmer process within a couple of dispatcher cycles.
_WARMER_STALE_AFTER_S_DEFAULT = 300
_WARMER_STALE_AFTER_S_ENV = "PR_STATE_WARMER_STALE_AFTER_S"
# Floor below which env-provided values are ignored as obvious config
# errors. Production should never let an operator set this below
# the warmer's interval (config drift would make every cycle fall
# through to a live fetch). Tests that need sub-floor values to
# drive the staleness gate end-to-end monkeypatch
# ``_WARMER_STALE_FLOOR_S`` to 1.
_WARMER_STALE_FLOOR_S = 30
def _warmer_preferred() -> bool:
"""Whether to read from :mod:`_pr_state_cache` before falling back
to a live Forgejo fetch. Default ON; ``=0`` rolls back to the
pre-warmer code path."""
raw = os.environ.get(_WARMER_PREFER_ENV)
if raw is None:
return True
return raw.strip().lower() in {"1", "true", "yes", "on"}
def _warmer_stale_after_s() -> int:
"""Staleness threshold in seconds. Default 300s (5 min ≈ 10× the
warmer's 30s interval).
Floored at ``_WARMER_STALE_FLOOR_S`` (30s in prod) so operator
config drift can't accidentally produce sub-interval thresholds
that fall through to live fetch every cycle. Tests that need
sub-floor values to drive the gate end-to-end monkeypatch the
floor symbol directly."""
raw = os.environ.get(_WARMER_STALE_AFTER_S_ENV)
if raw:
try:
return max(_WARMER_STALE_FLOOR_S, int(raw))
except ValueError:
pass
return _WARMER_STALE_AFTER_S_DEFAULT
def _warmer_cache_fresh(latest_at: str | None) -> bool:
"""True iff ``latest_at`` (ISO-8601) is recent enough to trust.
None / unparseable → False (treat as stale, fall through).
A naive ``latest_at`` (no tz suffix — e.g. a test that monkeypatched
``_now`` or a future schema-drift) is normalized to UTC before the
aware-vs-aware comparison so this never raises ``TypeError`` into
the dispatcher cycle.
"""
if not latest_at:
return False
try:
parsed = _dt.datetime.fromisoformat(str(latest_at))
except (ValueError, TypeError):
return False
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=_dt.timezone.utc)
deadline = parsed + _dt.timedelta(seconds=_warmer_stale_after_s())
return _dt.datetime.now(_dt.timezone.utc) < deadline
def _list_open_prs_sorted_by_updated(cfg: Any) -> list[dict[str, Any]]:
"""List open PRs sorted by most-recently-updated, with three
layered sources (preferred → fallback):
1. **Warmer cache** (:mod:`_pr_state_cache`) — populated by the
``pr_state_warmer`` sidecar every ~30s. Local SQLite read,
~0ms, always paginated across all open PRs, never blocks on
Forgejo flakes. Default path when warmer is running.
2. **Last-good list cache** — populated by this module on a
successful live fetch. Stored at ``/tmp/cleveragents-pr-list-cache``.
Used when the warmer's cache is empty (warmer not running
yet / disabled) AND when in active backoff window.
3. **Live Forgejo fetch** — single-page ``/pulls?state=open&sort=newest&limit=50&page=1``
call. Used when both caches are empty. Has the 50-PR hard cap
and cold-cache flakiness this module was originally built to
compensate for; the warmer is the real fix.
``PR_STATE_WARMER_PREFER=0`` skips layer 1 for rollback.
"""
# Layer 1: warmer cache (preferred). If the warmer has populated
# the local SQLite at any point, prefer it — it's always fresher
# than the request-time fallback below (30s poll cadence beats
# 120s dispatcher cycle) AND it doesn't block on Forgejo.
#
# Staleness check: if the latest warmer write is older than
# ``PR_STATE_WARMER_STALE_AFTER_S`` (default 5 min ≈ 10× the
# default 30s interval), the warmer process is likely dead and
# the cache is stale — fall through to the live path so the
# dispatcher doesn't blindly serve hour-old data.
if _warmer_preferred() and not _pr_state_cache.is_disabled():
try:
warmer_prs = _pr_state_cache.list_open_prs(
owner=cfg.owner, repo=cfg.repo,
)
latest_at = _pr_state_cache.latest_write_at(
owner=cfg.owner, repo=cfg.repo,
)
except _pr_state_cache.PRStateCacheError as exc:
_logger.warning("warmer-cache read failed; falling through: %s", exc)
warmer_prs = []
latest_at = None
# Freshness is keyed on ``latest_at`` (the warmer's heartbeat),
# NOT on whether ``warmer_prs`` is non-empty. A repo with zero
# open PRs and a healthy warmer ``latest_at`` IS fresh — return
# the empty list directly instead of falling through to live
# fetch every cycle.
fresh = bool(latest_at) and _warmer_cache_fresh(latest_at)
if fresh:
return warmer_prs
if latest_at and not fresh:
_logger.warning(
"warmer cache for %s/%s is stale (latest write %s); "
"warmer may be dead — falling through to live fetch",
cfg.owner, cfg.repo, latest_at,
)
# Cold warmer cache (no rows ever written): fall through to the
# live path below — it doubles as the bootstrap for cold starts.
path = (
f"/repos/{cfg.owner}/{cfg.repo}/pulls"
f"?state=open&sort=newest&limit=50&page=1"
)
if _list_cache_disabled():
return _do_live_pr_list_fetch(cfg, path)
cached = _read_list_cache(cfg)
now_dt = _dt.datetime.now(_dt.timezone.utc)
if cached is not None and _list_backoff_active(cached, now_dt):
_logger.info(
"PR-list cache for %s/%s in backoff "
"(failures=%s, next_attempt_after=%s); serving cached list "
"(count=%s)",
cfg.owner, cfg.repo,
cached.get("consecutive_failures", 0),
cached.get("next_attempt_after"),
len(cached.get("prs") or []),
)
return list(cached.get("prs") or [])
short_cfg = _cfg_with_short_timeout(cfg)
try:
live = _do_live_pr_list_fetch(short_cfg, path)
except Exception as exc: # noqa: BLE001
# Live fetch failed (timeout / network). Bump failures + serve
# cached list if we have one — caller still gets actionable
# data. If no cache exists, re-raise so the dispatcher's
# outer exception handler can fall back to the legacy TS
# path (slow but a different code path that occasionally
# succeeds when /pulls is flaky).
prior_failures = int((cached or {}).get("consecutive_failures") or 0)
next_failures = prior_failures + 1
payload = {
"schema_version": _LIST_CACHE_SCHEMA_VERSION,
"fetched_at": _now_iso(),
"prs": list((cached or {}).get("prs") or []),
"consecutive_failures": next_failures,
"next_attempt_after": _list_compute_next_attempt_after(
next_failures, now_dt,
),
"last_error": f"{type(exc).__name__}: {exc}",
}
_write_list_cache(cfg, payload)
if cached is None:
_logger.warning(
"PR-list live fetch failed for %s/%s with no cached "
"fallback; re-raising for legacy TS fallback: %s",
cfg.owner, cfg.repo, exc,
)
raise
_logger.warning(
"PR-list live fetch failed for %s/%s (consecutive_failures=%s); "
"serving cached list (count=%s); next attempt deferred "
"until %s",
cfg.owner, cfg.repo, next_failures,
len(payload["prs"]), payload["next_attempt_after"],
)
return list(payload["prs"])
# Live success — persist (full overwrite) and clear backoff.
_write_list_cache(cfg, {
"schema_version": _LIST_CACHE_SCHEMA_VERSION,
"fetched_at": _now_iso(),
"prs": live,
"consecutive_failures": 0,
"next_attempt_after": None,
})
return live
def _now_iso() -> str:
return _dt.datetime.now(_dt.timezone.utc).isoformat()
def _do_live_pr_list_fetch(cfg: Any, path: str) -> list[dict[str, Any]]:
"""The raw Forgejo /pulls listing fetch — same shape as the
pre-R1 code. Lifted out of ``_list_open_prs_sorted_by_updated``
so the cache wrapper can call it once on the happy path and
once for the disabled-mode bypass without duplicating logic."""
resp = _claim_runtime.idempotent_get(path, cfg)
status = int(resp.get("status") or 0)
if status != 200:
_logger.warning(
"PR list fetch returned status=%s; returning empty list",
status,
)
return []
body = resp.get("body")
if not isinstance(body, list):
_logger.warning(
"PR list fetch returned non-list body (type=%s); "
"returning empty list",
type(body).__name__,
)
return []
return body
def _ensure_classified(
cache: Any,
cfg: Any,
pr: dict[str, Any],
ttl_seconds: int,
) -> dict[str, Any]:
"""Cache-aware classification: return the cached row when fresh,
else re-classify and write back."""
pr_number = int(pr.get("number") or 0)
cached = cache.get_pr_classification(pr_number)
if cached and _is_cache_fresh(cached, pr, ttl_seconds):
return cached
fresh = _classify_pr(cfg, pr)
cache.upsert_pr_classification(fresh)
return fresh
def _is_cache_fresh(
cached: dict[str, Any],
pr: dict[str, Any],
ttl_seconds: int,
) -> bool:
"""Four-axis freshness check (per the plan + module docstring)."""
head_sha = (pr.get("head") or {}).get("sha") or ""
pr_updated = pr.get("updated_at") or ""
if cached.get("head_sha") != head_sha:
return False
# ``>=`` because the cache may have been refreshed AFTER the PR's
# updated_at (e.g. two cycles back-to-back). Only when the PR's
# updated_at moves strictly past the cache's recorded value do
# we re-classify.
if (cached.get("updated_at") or "") < pr_updated:
return False
expected_version = (
_forgejo_cache.ForgejoCache.PR_CLASSIFICATION_SCHEMA_VERSION
)
if cached.get("classification_schema_version") != expected_version:
return False
last = cached.get("last_checked_at") or ""
if not last:
return False
try:
last_dt = datetime.fromisoformat(last.replace("Z", "+00:00"))
except (TypeError, ValueError):
return False
age = datetime.now(timezone.utc) - last_dt
if age > timedelta(seconds=ttl_seconds):
return False
return True
def _classify_pr(cfg: Any, pr: dict[str, Any]) -> dict[str, Any]:
"""Per-PR fetch + classification. Returns a dict with all the
schema fields populated."""
pr_number = int(pr.get("number") or 0)
head_sha = (pr.get("head") or {}).get("sha") or ""
ci_status = _classify_ci_status(cfg, head_sha)
reviews, _completed = _review_fetch.fetch_existing_reviews(cfg, pr_number)
approvals_count = _count_active_approvals(reviews)
has_active_rc = (
_review_fetch.count_active_request_changes(reviews) > 0
)
has_unaddressed_rc = _has_unaddressed_request_changes(
cfg, pr_number, reviews,
)
labels = [(l.get("name") or "") for l in (pr.get("labels") or [])]
is_claimed = any(name in CLAIM_LABELS for name in labels)
mergeable = pr.get("mergeable") # bool or None
# stale_state: not used by any of the 5 reviewer filters today,
# so we punt — set to "not_stale" without fetching the base branch.
# When a future filter needs real stale-state, add a one-call-per-
# base-ref lookup (typically just origin/master once per cycle).
stale_state = "not_stale"
return {
"pr_number": pr_number,
"head_sha": head_sha,
"updated_at": pr.get("updated_at") or "",
"last_checked_at": datetime.now(timezone.utc).isoformat(),
"ci_status": ci_status,
"approvals_count": approvals_count,
"has_active_request_changes": has_active_rc,
"has_unaddressed_request_changes": has_unaddressed_rc,
"is_claimed": is_claimed,
"is_mergeable": mergeable,
"stale_state": stale_state,
"labels_json": json.dumps(labels),
"classification_schema_version": (
_forgejo_cache.ForgejoCache.PR_CLASSIFICATION_SCHEMA_VERSION
),
}
def _classify_ci_status(cfg: Any, head_sha: str) -> str:
"""Map Forgejo's combined-status state into the 4-value
classification ``list_prs.ts`` uses. Match its mapping exactly so
the Python output is parity with the TS output."""
if not head_sha:
return "unknown"
status = _review_fetch.fetch_ci_status(cfg, head_sha)
if not status:
return "unknown"
state = (status.get("state") or "").lower()
if state == "success":
return "passing"
# list_prs.ts treats failure/error/warning all as "failing".
if state in ("failure", "error", "warning"):
return "failing"
if state == "pending":
return "pending"
# Empty string or anything else is "no CI checks reported yet".
return "unknown"
def _count_active_approvals(reviews: list[dict[str, Any]]) -> int:
"""Count non-dismissed APPROVE reviews using one-per-author
semantics (only the author's latest review counts).
Matches ``list_prs.ts``'s approvals_count: a reviewer who first
APPROVEs then later REQUEST_CHANGES gives zero approvals (the
APPROVE is superseded). Stale reviews (those marked ``stale``
after a new commit) still count if not dismissed — that's the
Forgejo default and matches the TS semantic.
"""
latest_per_user: dict[str, dict[str, Any]] = {}
for r in reviews:
if not isinstance(r, dict):
continue
user_obj = r.get("user")
login = (user_obj.get("login") if isinstance(user_obj, dict) else None) or ""
submitted_at = r.get("submitted_at") or ""
current = latest_per_user.get(login)
if current is None or submitted_at > (current.get("submitted_at") or ""):
latest_per_user[login] = r
count = 0
for r in latest_per_user.values():
if r.get("dismissed"):
continue
if (r.get("state") or "").upper() == "APPROVED":
count += 1
return count
def _has_unaddressed_request_changes(
cfg: Any,
pr_number: int,
reviews: list[dict[str, Any]],
) -> bool:
"""True iff there's at least one active REQUEST_CHANGES review
that has NOT been followed by a new commit.
Pulls the PR's commit list to compare timestamps against the
latest active RC review. ``list_prs.ts`` uses the same shape.
The commit fetch is the most expensive part of classification (PR
commit list can be long), so the cache-hit path SKIPS this entirely.
"""
rc_timestamps: list[str] = []
for r in reviews:
if not isinstance(r, dict):
continue
if r.get("dismissed"):
continue
if (r.get("state") or "").upper() != "REQUEST_CHANGES":
continue
ts = r.get("submitted_at") or ""
if ts:
rc_timestamps.append(ts)
if not rc_timestamps:
return False
latest_rc = max(rc_timestamps)
commits, _completed = _review_fetch.fetch_pr_commits(cfg, pr_number)
for c in commits:
if not isinstance(c, dict):
continue
commit = c.get("commit") or {}
committer = commit.get("committer") if isinstance(commit, dict) else {}
commit_ts = (
committer.get("date") if isinstance(committer, dict) else ""
) or ""
if commit_ts > latest_rc:
return False
return True
def _evaluate_filter(filter_name: str, row: dict[str, Any]) -> bool:
"""Apply the named filter predicate to a classification row.
Predicates are mechanical translations of the 5 TS wrapper scripts'
hard-coded args in
``.opencode/skills/auto-agents-system/scripts/list_prs_*.ts``.
Verified against the scripts' comment blocks 2026-05-16.
"""
if filter_name == "addressed_changes_ci_passing":
return (
row["ci_status"] == "passing"
and row["approvals_count"] == 0
and bool(row["has_active_request_changes"])
and not bool(row["has_unaddressed_request_changes"])
and not bool(row["is_claimed"])
)
if filter_name == "addressed_changes_ci_failing":
return (
row["ci_status"] == "failing"
and row["approvals_count"] == 0
and bool(row["has_active_request_changes"])
and not bool(row["has_unaddressed_request_changes"])
and not bool(row["is_claimed"])
)
if filter_name == "no_active_review_ci_passing":
return (
row["ci_status"] == "passing"
and row["approvals_count"] == 0
and not bool(row["has_active_request_changes"])
and not bool(row["is_claimed"])
)
if filter_name == "no_active_review_ci_failing":
return (
row["ci_status"] == "failing"
and row["approvals_count"] == 0
and not bool(row["has_active_request_changes"])
and not bool(row["is_claimed"])
)
if filter_name == "missing_ci_checks":
return (
row["ci_status"] == "unknown"
and row["approvals_count"] == 0
and not bool(row["has_unaddressed_request_changes"])
and not bool(row["is_claimed"])
)
# Unreachable — refresh_then_filter validates filter_name first.
raise ValueError(f"unknown filter {filter_name!r}")
def _project_pr(
pr: dict[str, Any],
cls: dict[str, Any],
) -> dict[str, Any]:
"""Trim the Forgejo PR object to the fields downstream callers
actually use, plus the classification flags. Matches the shape
the TS scripts emitted so the dispatcher cutover (Phase 2) is a
drop-in."""
return {
"number": pr.get("number"),
"title": pr.get("title"),
"head_sha": cls["head_sha"],
"head_ref": (pr.get("head") or {}).get("ref"),
"base_ref": (pr.get("base") or {}).get("ref"),
"updated_at": pr.get("updated_at"),
"labels": [(l.get("name") or "") for l in (pr.get("labels") or [])],
"ci_status": cls["ci_status"],
"approvals_count": cls["approvals_count"],
"has_active_request_changes": bool(cls["has_active_request_changes"]),
"has_unaddressed_request_changes": bool(
cls["has_unaddressed_request_changes"]
),
"is_claimed": bool(cls["is_claimed"]),
}
__all__ = (
"FILTER_NAMES",
"DEFAULT_TTL_SECONDS",
"CLAIM_LABELS",
"refresh_then_filter",
)