cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity
Files
d4e1219577e01f270aca90e2706cfdcd523100cd
cleveragents-core/tools/controller/master/loop.py
T
drew 14e592ddd5 feat(controller): zombie-CI detection — stop waiting on dead CI runs 
A CI gate stuck `pending` is ambiguous: the job may genuinely be
running, or the run may be dead (a crashed runner, an Actions job whose
terminal commit-status was never posted — `CI / status-check` zombies
routinely here). The "wait for the whole run to finish" fix then waited
forever on the dead case (PR #36: a `status-check` gate pending for 8 h
while the run had actually finished RED 8 h earlier).

New `ci_run_status.classify_ci_run` resolves a still-pending run to
`complete` / `running` / `stale` via two checks, authoritative-first:

  1. ACTIVE-RUN — `get_action_tasks` asks Forgejo's Actions API
     directly whether a task for the commit is still running; catches a
     dead run immediately, regardless of age.
  2. AGE — if no gate has updated in > CONTROLLER_CI_STALE_AFTER_MIN
     (default 90) the run has stopped; the fallback when the Actions
     API is unavailable.

A `stale` run is no longer waited on: the verdict is taken from the
gates that DID finish (`terminal_verdict`) — any failure → red, all
pass → green, fully-dead → red. Applied in both `ci_status_poll` (the
AWAITING_CI verdict) and `ci_summarize` (the implementer's summary —
zombie pending gates drop out of `gates_pending`/`overall_state`) so
the two agree and never ping-pong.

New `get_action_tasks` Forgejo callback wired through forgejo_http →
__main__ → the poll and the prefetch path.

23 new tests; full controller suite (1222) green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 02:21:12 -04:00

507 lines
22 KiB
Python
Raw Blame History

"""Master main loop — composes the per-tick work (tick + reaper +
pickup guard) and runs it on a configurable cadence until a stop
event fires.
Per plan v9: the master is a long-running singleton per (owner, repo).
This module is the orchestrator that ties together the deterministic
pieces shipped in Phase 1d-1/1d-2 + later additions.
Currently composes:
- ``tick.run_tick()`` — advance state for any completed attempts
- ``reaper.reap_stale_attempts()`` — reset stale-heartbeat in_progress rows
- ``pickup_guard.transition_exhausted_to_stuck()`` — STUCK workflows
whose attempts have been re-pended too many times
Deferred to Phase 1d-3+:
- Discovery (Forgejo poll for new PRs/issues + insert DISCOVERED rows)
- Per-workflow scheduling (after a state transition, enqueue the
next attempt's workflow_attempts row with input_payload prefetched)
- Forgejo writes (status comments, labels, merges)
- MERGING state's actual Forgejo merge call
- Periodic reconciliation tick
- Backfill at startup
- Operator CLI server (HTTP / unix socket for controller-cli)
"""
from __future__ import annotations
import logging
import os
import threading
from collections.abc import Callable
from dataclasses import dataclass
from sqlalchemy.engine import Engine
from ..pickup_guard import (
DEFAULT_MAX_PICKUPS,
PickupGuardReport,
transition_exhausted_to_stuck,
)
from ..reaper import ReaperReport, reap_stale_attempts
from .ci_gate import CIGateReport, run_ci_gate_tick
from .ci_poll import CIPollExhaustionReport, run_ci_poll_exhaustion_tick
from .ci_status_poll import (
CIStatusPollReport,
GetCIStatusCallback,
run_ci_status_poll_tick,
)
from .discovery import DiscoveryReport, run_discovery
from .merging import MergeCallback, MergingHandlerReport, run_merging_tick
from .promote import PromoteDiscoveredReport, run_promote_discovered_tick
from .reconciliation import (
GetIssueStateCallback,
GetPRStateCallback,
ReconciliationReport,
run_reconciliation_tick,
)
from .scheduler import (
PrefetchCallback,
SchedulerReport,
schedule_next_attempts,
)
from .tick import TickReport, run_tick
logger = logging.getLogger(__name__)
@dataclass
class MasterConfig:
"""Per-master config; tunable via env vars."""
tick_interval_s: float = float(
os.environ.get("CONTROLLER_MASTER_TICK_INTERVAL_S", "30")
)
reaper_interval_s: float = float(
os.environ.get("CONTROLLER_REAPER_INTERVAL_S", "60")
)
reconciliation_interval_s: float = float(
os.environ.get("CONTROLLER_RECONCILIATION_INTERVAL_S", "300")
)
# CI poll-exhaustion runs on the same cadence as reconciliation
# by default — both are slow ticks that scan all non-terminal
# workflows.
ci_poll_exhaustion_interval_s: float = float(
os.environ.get("CONTROLLER_CI_POLL_EXHAUSTION_INTERVAL_S", "300")
)
# CI status polling — how often the master checks Forgejo for CI
# results on AWAITING_CI workflows. Default 60s; faster than
# reconciliation since CI usually finishes in minutes.
ci_status_poll_interval_s: float = float(
os.environ.get("CONTROLLER_CI_STATUS_POLL_INTERVAL_S", "60")
)
# Periodic discovery — how often the master polls Forgejo for
# new PRs/issues. Without this, only PRs that existed at master
# startup get discovered (backfill is one-shot). Default 30s.
discovery_interval_s: float = float(
os.environ.get("CONTROLLER_DISCOVERY_INTERVAL_S", "30")
)
pickup_guard_max_pickups: int = DEFAULT_MAX_PICKUPS
@dataclass
class MasterTickReport:
"""Summary of one composite master tick."""
tick: TickReport
reaper: ReaperReport
pickup_guard: PickupGuardReport
reconciliation: ReconciliationReport | None = None
ci_poll_exhaustion: CIPollExhaustionReport | None = None
promote_discovered: PromoteDiscoveredReport | None = None
scheduler: SchedulerReport | None = None
merging: MergingHandlerReport | None = None
discovery: DiscoveryReport | None = None
ci_status_poll: CIStatusPollReport | None = None
ci_gate: CIGateReport | None = None
def run_master_iteration(
engine: Engine,
*,
max_pickups: int = DEFAULT_MAX_PICKUPS,
) -> MasterTickReport:
"""Run one composite iteration: tick + reaper + pickup guard.
Order matters:
1. ``tick`` first: advance state machine for completed attempts;
may produce new transitions that the reaper / pickup guard
then notice.
2. ``reaper`` next: reset stale-heartbeat in_progress rows.
Post-reap, those attempts return to the pending pool +
pickup_count is preserved (the guard uses it).
3. ``pickup_guard`` last: STUCK any workflows whose pending
attempts have hit MAX_PICKUPS. Runs AFTER the reaper so a
just-reaped attempt's pickup_count is visible.
"""
return MasterTickReport(
tick=run_tick(engine),
reaper=reap_stale_attempts(engine),
pickup_guard=transition_exhausted_to_stuck(engine, max_pickups=max_pickups),
)
def master_main_loop(
engine: Engine,
*,
config: MasterConfig | None = None,
stop_event: threading.Event | None = None,
on_iteration: Callable[[MasterTickReport], None] | None = None,
reconciliation_args: tuple | None = None,
# If set: 4- OR 5-tuple — (owner, repo, get_pr_state, get_issue_state)
# OR (owner, repo, get_pr_state, get_issue_state, recon_kwargs_dict).
# The reconciliation tick fires every reconciliation_interval_s.
# None disables reconciliation (useful for tests that don't need it).
# recon_kwargs_dict (Phase 1k+) is passed through as keyword args
# to run_reconciliation_tick — used for opt_in_label /
# require_opt_in_label settings.
prefetch: PrefetchCallback | None = None,
# When set, the loop runs the DISCOVERED→ANALYZING promoter +
# the per-workflow scheduler every iteration. Without it, the
# scheduler is skipped — workflows would still transition between
# states but never get fresh worker attempts enqueued. Production
# __main__.py always passes a prefetch callback; tests may pass
# None to skip the scheduling layer.
merging_args: tuple | None = None,
# If set: (owner, repo, merge_callback). The MERGING handler
# fires every iteration (cheap if no workflows in MERGING).
# Without this, workflows that transition to MERGING never have
# the Forgejo merge call invoked — they sit in MERGING forever.
discovery_args: tuple | None = None,
# If set: 4- OR 5-tuple — (owner, repo, list_prs, list_issues)
# OR (owner, repo, list_prs, list_issues, discovery_kwargs).
# Periodic discovery fires every discovery_interval_s. Without
# this, only PRs that existed at master startup (via backfill)
# are ever managed — PRs created after master startup wait until
# the master restarts.
ci_status_poll_args: tuple | None = None,
# If set: (owner, repo, get_ci_status) OR the same with a
# trailing get_failure_logs and/or get_pr_details. The CI status
# poller fires every ci_status_poll_interval_s. Without this,
# workflows in AWAITING_CI exit only via the polling-exhaustion
# timeout (operator-intervention path); with it, ci_green /
# ci_red / ci_infra_recheck transitions fire autonomously.
# ``get_failure_logs`` (optional 4th element) lets the poll
# classify a failure infra-vs-real and route an infra-broken
# reran-CI back to DISCOVERED for the gate's rerun budget.
# ``get_pr_details`` (optional 5th element) feeds the pre-review
# mergeable gate: a green-CI PR that no longer merges cleanly is
# routed to CONFLICT_RESOLVING instead of REVIEWING, skipping a
# doomed LLM review.
ci_gate_args: tuple | None = None,
# If set: (owner, repo, get_ci_status, get_pr_details,
# trigger_ci_rerun) OR the same with a trailing get_failure_logs.
# The CI-freshness gate runs every iteration BEFORE the
# DISCOVERED→ANALYZING promoter, so a DISCOVERED PR with
# stale/infra-broken CI is diverted (CI rerun → AWAITING_CI, or
# STUCK once the rerun budget is exhausted) before it burns an
# estimator + implementer attempt. ``get_failure_logs`` (optional
# 6th element) gives the gate's classifier the real CI job-log
# content it needs to detect infra failures. Without this, the
# incident-class bug (stale infra-failed CI dead-ends a workflow
# at STUCK) reappears.
local_ci_in_flight: Callable[[], bool] | None = None,
# RUN_CI_LOCAL only: a predicate reporting whether a local CI run
# is currently executing. When set, the ci_poll_exhaustion sweep is
# skipped while local CI is busy — an on-demand verdict is minutes
# away and AWAITING_CI workflows queue behind it, so the timeout
# (sized for remote CI) must not STUCK them. See
# run_ci_poll_exhaustion_tick.
) -> None:
"""Run the master loop until ``stop_event`` is set.
Different ticks at different cadences:
- tick (state machine + transitions): every tick_interval_s (30s)
- reaper (stale-heartbeat reset): every reaper_interval_s (60s)
- reconciliation (Forgejo sync): every reconciliation_interval_s (300s)
- pickup guard: every iteration (cheap)
"""
cfg = config or MasterConfig()
stop = stop_event or threading.Event()
last_reap_at_iteration = 0
last_reconcile_at_iteration = 0
last_ci_poll_exhaustion_at_iteration = 0
last_discovery_at_iteration = 0
last_ci_status_poll_at_iteration = 0
iteration = 0
logger.info(
"master loop starting: tick=%.1fs reaper=%.1fs reconcile=%.1fs "
"ci_poll_exh=%.1fs discovery=%.1fs max_pickups=%d",
cfg.tick_interval_s,
cfg.reaper_interval_s,
cfg.reconciliation_interval_s,
cfg.ci_poll_exhaustion_interval_s,
cfg.discovery_interval_s,
cfg.pickup_guard_max_pickups,
)
try:
while not stop.is_set():
iteration += 1
# Always run tick + pickup guard. Run reaper + reconciliation
# less often per their own intervals.
tick_report = run_tick(engine)
should_reap = (
iteration - last_reap_at_iteration
) * cfg.tick_interval_s >= cfg.reaper_interval_s
reaper_report = (
reap_stale_attempts(engine) if should_reap else ReaperReport()
)
if should_reap:
last_reap_at_iteration = iteration
should_reconcile = (
reconciliation_args is not None
and (iteration - last_reconcile_at_iteration) * cfg.tick_interval_s
>= cfg.reconciliation_interval_s
)
reconciliation_report: ReconciliationReport | None = None
if should_reconcile and reconciliation_args is not None:
try:
n = len(reconciliation_args)
if n == 5:
(owner, repo, get_pr_state, get_issue_state, extra_kwargs) = (
reconciliation_args
)
elif n == 4:
(owner, repo, get_pr_state, get_issue_state) = (
reconciliation_args
)
extra_kwargs = {}
else:
# R-round4 P4: explicit length check. A 6+ tuple
# used to silently fall to the `else` branch +
# raise ValueError("too many values to unpack")
# which the outer except Exception swallowed
# silently every iteration.
raise TypeError(
f"reconciliation_args must be a 4- or 5-tuple; "
f"got length {n}"
)
reconciliation_report = run_reconciliation_tick(
engine,
owner=owner,
repo=repo,
get_pr_state=get_pr_state,
get_issue_state=get_issue_state,
**(extra_kwargs or {}),
)
except Exception:
logger.exception("reconciliation tick raised; continuing")
last_reconcile_at_iteration = iteration
pickup_report = transition_exhausted_to_stuck(
engine,
max_pickups=cfg.pickup_guard_max_pickups,
)
# CI poll-exhaustion: STUCK any AWAITING_CI workflow whose
# entered_state_at is older than the threshold. Runs on
# its own cadence (default = reconciliation cadence).
should_ci_poll_exh = (
iteration - last_ci_poll_exhaustion_at_iteration
) * cfg.tick_interval_s >= cfg.ci_poll_exhaustion_interval_s
ci_poll_report: CIPollExhaustionReport | None = None
if should_ci_poll_exh:
try:
ci_poll_report = run_ci_poll_exhaustion_tick(
engine, local_ci_in_flight=local_ci_in_flight
)
except Exception:
logger.exception("ci_poll_exhaustion tick raised; continuing")
last_ci_poll_exhaustion_at_iteration = iteration
# CI-freshness gate: divert DISCOVERED PRs whose CI is
# stale/infra-broken BEFORE the promoter runs. Ordered
# first so a stale-CI workflow is re-triggered (→
# AWAITING_CI) or STUCK'd (rerun budget exhausted) instead
# of being promoted to ANALYZING and burning a worker. Both
# ticks use guarded UPDATEs on current_state='DISCOVERED',
# so they're race-safe even out of order — but ordering
# this first makes the divert the common case. Runs every
# iteration (cheap when no DISCOVERED PRs).
ci_gate_report: CIGateReport | None = None
if ci_gate_args is not None:
try:
(cg_owner, cg_repo, cg_get_ci, cg_get_pr, cg_trigger) = (
ci_gate_args[:5]
)
# Optional 6th element: the CI job-log fetcher.
cg_get_logs = ci_gate_args[5] if len(ci_gate_args) > 5 else None
ci_gate_report = run_ci_gate_tick(
engine,
owner=cg_owner,
repo=cg_repo,
get_ci_status=cg_get_ci,
get_pr_details=cg_get_pr,
trigger_ci_rerun=cg_trigger,
get_failure_logs=cg_get_logs,
)
except Exception:
logger.exception("ci_gate tick raised; continuing")
# Phase 1k+++ (real-run): promote DISCOVERED → ANALYZING +
# schedule next attempts. Without these two ticks the
# master never enqueues anything for workers to pick up.
# Runs every iteration (both are cheap).
promote_report: PromoteDiscoveredReport | None = None
scheduler_report: SchedulerReport | None = None
if prefetch is not None:
try:
promote_report = run_promote_discovered_tick(engine)
except Exception:
logger.exception("promote_discovered tick raised; continuing")
try:
scheduler_report = schedule_next_attempts(
engine,
prefetch=prefetch,
)
except Exception:
logger.exception("scheduler tick raised; continuing")
# Phase 1k+++ (real-run): MERGING handler. For workflows
# in MERGING state, call the Forgejo merge endpoint via
# the injected callback. Without this, workflows that
# transition to MERGING (via reviewer approval) never
# have the actual merge call invoked. Cheap when no
# workflows are in MERGING.
merging_report: MergingHandlerReport | None = None
if merging_args is not None:
try:
owner, repo, merge_cb = merging_args
merging_report = run_merging_tick(
engine,
merge=merge_cb,
owner=owner,
repo=repo,
)
except Exception:
logger.exception("merging tick raised; continuing")
# Phase 1k+++ (real-run): periodic discovery. Backfill is
# one-shot at startup; new PRs created later need this
# tick to be picked up. Runs every discovery_interval_s.
discovery_report: DiscoveryReport | None = None
should_discover = (
discovery_args is not None
and (iteration - last_discovery_at_iteration) * cfg.tick_interval_s
>= cfg.discovery_interval_s
)
if should_discover and discovery_args is not None:
try:
n = len(discovery_args)
if n == 5:
(d_owner, d_repo, d_list_prs, d_list_issues, d_kwargs) = (
discovery_args
)
elif n == 4:
(d_owner, d_repo, d_list_prs, d_list_issues) = discovery_args
d_kwargs = {}
else:
raise TypeError(
f"discovery_args must be a 4- or 5-tuple; got length {n}"
)
discovery_report = run_discovery(
engine,
owner=d_owner,
repo=d_repo,
list_prs=d_list_prs,
list_issues=d_list_issues,
**(d_kwargs or {}),
)
except Exception:
logger.exception("discovery tick raised; continuing")
last_discovery_at_iteration = iteration
# CI status poll: scan AWAITING_CI workflows + apply
# ci_green / ci_red transitions based on Forgejo's
# combined-status. Without this the workflow only exits
# AWAITING_CI via the ci_poll_exhaustion timeout. Runs
# on ci_status_poll_interval_s cadence.
ci_status_report: CIStatusPollReport | None = None
should_ci_status_poll = (
ci_status_poll_args is not None
and (iteration - last_ci_status_poll_at_iteration) * cfg.tick_interval_s
>= cfg.ci_status_poll_interval_s
)
if should_ci_status_poll and ci_status_poll_args is not None:
try:
csp_owner, csp_repo, csp_get_ci = ci_status_poll_args[:3]
# Optional 4th element: the CI job-log fetcher.
csp_get_logs = (
ci_status_poll_args[3] if len(ci_status_poll_args) > 3 else None
)
# Optional 5th element: the PR-detail fetcher for
# the pre-review mergeable gate.
csp_get_pr_details = (
ci_status_poll_args[4] if len(ci_status_poll_args) > 4 else None
)
# Optional 6th element: the Actions-task fetcher for
# the zombie-CI active-run check.
csp_get_action_tasks = (
ci_status_poll_args[5] if len(ci_status_poll_args) > 5 else None
)
ci_status_report = run_ci_status_poll_tick(
engine,
owner=csp_owner,
repo=csp_repo,
get_ci_status=csp_get_ci,
get_failure_logs=csp_get_logs,
get_pr_details=csp_get_pr_details,
get_action_tasks=csp_get_action_tasks,
)
except Exception:
logger.exception("ci_status_poll tick raised; continuing")
last_ci_status_poll_at_iteration = iteration
if on_iteration is not None:
try:
on_iteration(
MasterTickReport(
tick=tick_report,
reaper=reaper_report,
pickup_guard=pickup_report,
reconciliation=reconciliation_report,
ci_poll_exhaustion=ci_poll_report,
promote_discovered=promote_report,
scheduler=scheduler_report,
merging=merging_report,
discovery=discovery_report,
ci_status_poll=ci_status_report,
ci_gate=ci_gate_report,
)
)
except Exception:
logger.exception("on_iteration callback raised")
if (
tick_report.transitions_applied
or reaper_report.rows_reaped
or pickup_report.workflows_stuck
or (
reconciliation_report
and reconciliation_report.workflows_transitioned
)
):
logger.info(
"master iteration %d: transitions=%d reaped=%d "
"stuck=%d reconciled=%d",
iteration,
tick_report.transitions_applied,
reaper_report.rows_reaped,
pickup_report.workflows_stuck,
reconciliation_report.workflows_transitioned
if reconciliation_report
else 0,
)
stop.wait(cfg.tick_interval_s)
finally:
logger.info("master loop stopped after %d iterations", iteration)
__all__ = [
"MasterConfig",
"MasterTickReport",
"master_main_loop",
"run_master_iteration",
]
Reference in New Issue View Git Blame Copy Permalink