"""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 .reconciliation import ( GetIssueStateCallback, GetPRStateCallback, ReconciliationReport, run_reconciliation_tick, ) 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") ) 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 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: (owner, repo, get_pr_state, get_issue_state) — the # reconciliation tick fires every reconciliation_interval_s. None # disables reconciliation (useful for tests that don't need it). ) -> 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 iteration = 0 logger.info( "master loop starting: tick=%.1fs reaper=%.1fs reconcile=%.1fs max_pickups=%d", cfg.tick_interval_s, cfg.reaper_interval_s, cfg.reconciliation_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: owner, repo, get_pr_state, get_issue_state = reconciliation_args reconciliation_report = run_reconciliation_tick( engine, owner=owner, repo=repo, get_pr_state=get_pr_state, get_issue_state=get_issue_state, ) 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, ) if on_iteration is not None: try: on_iteration(MasterTickReport( tick=tick_report, reaper=reaper_report, pickup_guard=pickup_report, reconciliation=reconciliation_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", ]