cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity
Files
a91df787d77073e558a11e9903dae7ed284845a3
cleveragents-core/tools/controller/master/loop.py
T
drew a91df787d7 feat(controller): Phase 2 — Gate 2 estimator-abandon 
Adds the second of three abandon gates: the estimator (Gate 2) can
mark a work item fundamentally unworkable, transitioning the
workflow ANALYZING -> ABANDONED and triggering a Forgejo close via
Phase 1's decomposed close_act orchestrator. Catches abandon cases
at the cheapest LLM stage, before implementer/reviewer tiers fire.

Substantive:
- EstimatorOutputV1: additive verdict + abandon_reason_category +
  abandon_reason_detail fields (pre-Phase-2 outputs still parse).
  @model_validator enforces abandon-requires-category atomicity at
  parse time — third defense layer beyond MCP setter + outcomes
  mapper
- state_machine: estimator_abandon event + (ANALYZING,
  estimator_abandon) -> ABANDONED. 57 transitions; invariants clean
- mcp/estimator_builder: estimator_set_verdict setter validates
  verdict enum + 9-category whitelist (scope_intractable,
  intent_wrong, security_regression, deprecated_dependency,
  breaks_protected_invariants, out_of_scope, low_value,
  unmaintained_path, policy_violation) + cross-field rules
- outcomes._map_estimator_outcome: dispatch verdict='abandon'
  -> estimator_abandon, with confidence-low downgrade to
  estimator_done (honors the agent prompt's documented "high or
  medium" requirement)
- estimator_abandon_side_effects.py: per-state side-effect tick
  modeled on grooming_side_effects.py; invokes close_act with
  cause=Cause.ESTIMATOR_ABANDON + event_type='estimator_abandon'
- _events.py: shared latest_transition_event +
  workflows_with_latest_transition_in helpers; dialect-aware
  payload['event'] extraction (SQLite json_extract +
  PostgreSQL ->>); centralizes the event_type='transition' +
  payload['event'] convention that side-effect ticks consume
- gate2_abandon_config.py: CONTROLLER_GATE2_ABANDON_ENABLED kill
  switch (default false). Fresh Phase 2 deploys are audit-only
  until operator explicitly enables; dry_run shared with grooming
  for unified safe-rollout staging
- .opencode/agents/estimator-implementation.md: GATE 2 ABANDON
  section with 9-category criteria + low_value disqualifier ("PR
  cites an issue/ticket -> route to reviewer instead")

Round-2 adversarial-review fixes (all required pre-commit):
- forgejo_writes.close_issue / close_act: NEW cause + event_type
  kwargs (defaults preserve Phase 1 grooming behavior; Phase 2
  callsite overrides). Fixes audit-trail attribution: telemetry
  queries SELECT WHERE cause='estimator_abandon' now return the
  right rows. Phase 1 regression test pins the grooming defaults
- tick.py operator_unstick lookback: dialect-aware json_extract
  fix (Phase 1 carry-over bug; would silently no-op on PostgreSQL)
- grooming_side_effects.py: idempotency filter now keys on
  check_name set (grooming check_names only) so a Phase 1 close
  and a Phase 2 close on the same workflow don't cross-cancel

Tests (+50): TestEstimatorOutputV1Phase2,
TestEstimatorAbandonStateMachine, TestMapEstimatorOutcomePhase2
(including confidence-low downgrade), TestEstimatorSetVerdict
(all 9 categories + cross-field rules), TestEventsHelper,
TestEstimatorAbandonSideEffectTick (including
test_close_writes_estimator_abandon_cause_and_event_type pinning
the audit-trail attribution, and Phase 1 regression guard).
Doc-contract test asserts all 9 categories appear in the agent
prompt. 1509/1509 passing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 15:44:09 -04:00

594 lines
27 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 .estimator_abandon_side_effects import (
EstimatorAbandonCallbacks,
EstimatorAbandonSideEffectReport,
run_estimator_abandon_side_effects_tick,
)
from .grooming_side_effects import (
GroomingCallbacks,
GroomingSideEffectReport,
run_grooming_side_effects_tick,
)
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
grooming_side_effects: GroomingSideEffectReport | None = None
estimator_abandon_side_effects: EstimatorAbandonSideEffectReport | 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,
estimator_abandon_callbacks: EstimatorAbandonCallbacks | None = None,
# When set, the estimator-abandon side-effect tick fires every
# iteration: it finds workflows whose state-machine just
# transitioned via ``estimator_abandon`` (ANALYZING → ABANDONED)
# and performs the Forgejo close (PATCH state:closed + audit
# comment) via ``forgejo_writes.close_act``. Cheap when no
# workflows are in the just-abandoned state. None disables the
# tick — the state transition still happens, but Forgejo never
# sees the close. Pass an ``EstimatorAbandonCallbacks`` instance
# with the Forgejo HTTP closures + dry_run flag (sources from
# ``CONTROLLER_GROOMING_DRY_RUN`` to share the safe-rollout
# toggle with grooming).
grooming_callbacks: GroomingCallbacks | None = None,
# When set, the grooming side-effect tick fires every iteration:
# it finds workflows whose state-machine just transitioned via
# ``groom_verdict_defer`` / ``groom_verdict_close`` and performs
# the Forgejo writes (audit comment + label swap / PATCH closed)
# via the decomposed ``forgejo_writes.close_act`` / ``defer_act``.
# Cheap when nothing is in a verdict-just-fired state. Without
# this (None), the grooming worker's verdicts transition workflow
# state but Forgejo never sees the action. Pass a
# ``GroomingCallbacks`` instance with the Forgejo HTTP closures +
# the dry_run flag.
# 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 1 corrected dispatch (worker-shape) — grooming
# side-effect tick. For workflows whose state machine just
# transitioned via ``groom_verdict_defer`` /
# ``groom_verdict_close``, perform the Forgejo writes
# (audit comment + label swap or PATCH closed) via the
# decomposed ``forgejo_writes.close_act`` / ``defer_act``.
# Idempotent: the ``grooming_decisions.executed=1`` filter
# skips already-completed workflows. Cheap when no
# workflows are pending side-effects.
grooming_side_effects_report: GroomingSideEffectReport | None = None
if grooming_callbacks is not None:
try:
grooming_side_effects_report = (
run_grooming_side_effects_tick(
engine=engine,
list_comments=grooming_callbacks.list_comments,
post_comment=grooming_callbacks.post_comment,
patch_pr_state=grooming_callbacks.patch_pr_state,
get_labels=grooming_callbacks.get_labels,
add_label=grooming_callbacks.add_label,
remove_label=grooming_callbacks.remove_label,
dry_run=grooming_callbacks.dry_run,
)
)
except Exception:
logger.exception(
"grooming_side_effects tick raised; continuing"
)
# Phase 2 corrected dispatch (2026-05-25) — estimator-abandon
# side-effect tick. Analogous to the grooming tick above;
# close-only (no defer path for Gate 2 abandons).
estimator_abandon_side_effects_report: (
EstimatorAbandonSideEffectReport | None
) = None
if estimator_abandon_callbacks is not None:
try:
estimator_abandon_side_effects_report = (
run_estimator_abandon_side_effects_tick(
engine=engine,
callbacks=estimator_abandon_callbacks,
)
)
except Exception:
logger.exception(
"estimator_abandon_side_effects 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,
grooming_side_effects=grooming_side_effects_report,
estimator_abandon_side_effects=(
estimator_abandon_side_effects_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