drew
251eeb21ff
Continuing the round-3 deep-pass cleanup. Three more run-blockers + one robustness fix. RB5 — MERGING handler never invoked from master loop: ``run_merging_tick`` was exported by the master package but no caller fired it. Workflows that transition to MERGING (via reviewer approval) would sit there indefinitely with no Forgejo merge call. Fix: - ``master/loop.py`` accepts a ``merging_args=(owner, repo, merge_callback)`` kwarg. When set, the tick fires every iteration (cheap if no workflows in MERGING). - ``MasterTickReport`` gains ``merging: MergingHandlerReport | None``. - ``master/__main__.py`` wires it from the Forgejo callback bundle. RB6 — periodic discovery never fires: ``run_discovery`` was only called at startup via ``run_startup_backfill`` + the ``--discovery-only-once`` smoke flag. PRs created after master startup would not be discovered until the master restarted. Fix: - ``master/loop.py`` accepts ``discovery_args=(owner, repo, list_prs, list_issues)`` or the 5-tuple with kwargs. Periodic tick on its own cadence (``CONTROLLER_DISCOVERY_INTERVAL_S``, default 30s). - ``MasterTickReport`` gains ``discovery: DiscoveryReport | None``. - ``master/__main__.py`` wires it + threads ``require_opt_in_label`` through. RB-robust — worker calls create_all defensively: Master is normally responsible for schema creation (workers run After= it via systemd ordering). But if the worker is started in isolation (test / local dev / unit ordering broken), it'd crash on the first query against missing tables. Fix: - ``worker/__main__.py`` calls ``create_all(engine)`` after ``build_engine``. ``create_all`` is idempotent (CREATE TABLE IF NOT EXISTS); safe to call from both master + worker. - ``cleveragents-controller-worker@.service`` adds ``After=cleveragents-controller-master.service`` + ``Wants=cleveragents-controller-master.service`` so systemd enforces the start ordering in production. Total: 703 controller tests pass (no test changes; all new wiring is exercised by master_main_loop tests via the new kwargs). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
360 lines
15 KiB
Python
360 lines
15 KiB
Python
"""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_poll import CIPollExhaustionReport, run_ci_poll_exhaustion_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")
|
|
)
|
|
# 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
|
|
|
|
|
|
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.
|
|
) -> 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
|
|
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:
|
|
if len(reconciliation_args) == 5:
|
|
(owner, repo, get_pr_state, get_issue_state,
|
|
extra_kwargs) = reconciliation_args
|
|
else:
|
|
(owner, repo, get_pr_state,
|
|
get_issue_state) = reconciliation_args
|
|
extra_kwargs = {}
|
|
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)
|
|
except Exception:
|
|
logger.exception(
|
|
"ci_poll_exhaustion tick raised; continuing"
|
|
)
|
|
last_ci_poll_exhaustion_at_iteration = iteration
|
|
|
|
# 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:
|
|
if len(discovery_args) == 5:
|
|
(d_owner, d_repo, d_list_prs, d_list_issues,
|
|
d_kwargs) = discovery_args
|
|
else:
|
|
(d_owner, d_repo, d_list_prs,
|
|
d_list_issues) = discovery_args
|
|
d_kwargs = {}
|
|
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
|
|
|
|
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,
|
|
))
|
|
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",
|
|
]
|