cleveragents-core/tools/controller/master/ci_poll.py

"""AWAITING_CI poll-exhaustion handler.

Per plan v9 + the round-2 review (item N2): without this, workflows
that enter ``AWAITING_CI`` and never receive ``ci_green`` / ``ci_red_*``
events (CI runner outage, broken integration, etc.) hang indefinitely
— only ``operator_unstick`` could rescue them.

This module ships the minimum-viable escape: a periodic scan that
finds workflows whose ``entered_state_at`` for AWAITING_CI exceeds
the configured poll-exhaustion threshold + fires the
``ci_polling_exhausted`` event → STUCK.

What this module DOES NOT do (yet):
- Actual CI status polling against Forgejo. The reconciliation tick
  + the prefetch-driven CI summarizer cover that path; this handler
  is the EXIT for workflows that have been polling-without-progress
  for too long.
- Differentiating "CI never reported" from "CI reported but the
  controller missed it". Both fall under the same timeout.

Default threshold: 2 hours (``CONTROLLER_AWAITING_CI_TIMEOUT_S``).
Operators can tune per repo via env.
"""
from __future__ import annotations

import json
import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone

from sqlalchemy import text
from sqlalchemy.engine import Engine

from ..db.session import session_scope
from ..state_machine import IllegalTransitionError, apply_event

logger = logging.getLogger(__name__)


DEFAULT_AWAITING_CI_TIMEOUT_S = int(
    os.environ.get("CONTROLLER_AWAITING_CI_TIMEOUT_S", "7200")
)


@dataclass
class CIPollExhaustionReport:
    """Per-sweep summary."""

    workflows_scanned: int = 0
    workflows_exhausted: int = 0
    exhausted_workflow_ids: list[int] = field(default_factory=list)


def run_ci_poll_exhaustion_tick(
    engine: Engine, *,
    timeout_s: int | None = None,
) -> CIPollExhaustionReport:
    """One sweep: STUCK any AWAITING_CI workflow whose entered_state_at
    is older than ``timeout_s``.

    Composes with the master's tick layers (the loop calls this on
    the same cadence as reconciliation by default — see loop.py).
    """
    threshold = timeout_s or DEFAULT_AWAITING_CI_TIMEOUT_S
    report = CIPollExhaustionReport()
    now = datetime.now(timezone.utc)

    with session_scope(engine) as session:
        dialect = session.bind.dialect.name if session.bind else "sqlite"
        if dialect == "postgresql":
            select_sql = text(
                "SELECT workflow_id, current_state, entered_state_at "
                "  FROM workflows "
                " WHERE current_state = 'AWAITING_CI' "
                "   AND entered_state_at IS NOT NULL "
                "   AND entered_state_at + (:threshold || ' seconds')::interval < :now"
            )
        else:
            # SQLite: TIMESTAMP arithmetic via julianday.
            select_sql = text(
                "SELECT workflow_id, current_state, entered_state_at "
                "  FROM workflows "
                " WHERE current_state = 'AWAITING_CI' "
                "   AND entered_state_at IS NOT NULL "
                "   AND (julianday(:now) - julianday(entered_state_at)) "
                "       * 86400 > :threshold"
            )
        rows = session.execute(
            select_sql, {"threshold": threshold, "now": now},
        ).all()
        report.workflows_scanned = len(rows)
        for row in rows:
            try:
                new_state = apply_event(
                    row.current_state, "ci_polling_exhausted",
                )
            except (IllegalTransitionError, ValueError) as exc:
                # IllegalTransitionError: state doesn't accept the event.
                # ValueError: state name isn't in KNOWN_STATES (DB row
                # corruption or an unknown-state guard miss). Both
                # should skip this row rather than aborting the tick.
                logger.warning(
                    "ci_polling_exhausted: workflow_id=%s state=%r "
                    "rejected the event (%s); skipping",
                    row.workflow_id, row.current_state, exc,
                )
                continue
            session.execute(
                text(
                    "UPDATE workflows SET "
                    "  current_state = :to_state, "
                    "  last_transition_at = :now, "
                    "  entered_state_at = :now "
                    "WHERE workflow_id = :wf_id"
                ),
                {
                    "to_state": new_state, "now": now,
                    "wf_id": row.workflow_id,
                },
            )
            session.execute(
                text(
                    "INSERT INTO controller_events "
                    "(workflow_id, ts, event_type, from_state, to_state, "
                    " payload, forgejo_write_pending, replay_attempts) "
                    "VALUES (:wf_id, :ts, 'ci_poll_exhausted', "
                    "        :from_state, :to_state, :payload, 0, 0)"
                ),
                {
                    "wf_id": row.workflow_id, "ts": now,
                    "from_state": row.current_state,
                    "to_state": new_state,
                    "payload": json.dumps({
                        "reason": "awaiting_ci_timeout",
                        "threshold_seconds": threshold,
                        # text() SELECTs return TIMESTAMP as a string on
                        # SQLite (and a datetime on Postgres). Normalize.
                        "entered_state_at": (
                            row.entered_state_at.isoformat()
                            if hasattr(row.entered_state_at, "isoformat")
                            else (str(row.entered_state_at)
                                  if row.entered_state_at else None)
                        ),
                        "source": "ci_poll_exhaustion",
                    }),
                },
            )
            report.workflows_exhausted += 1
            report.exhausted_workflow_ids.append(row.workflow_id)

    if report.workflows_exhausted:
        logger.warning(
            "ci_poll_exhaustion: %d workflow(s) STUCK after %ds in "
            "AWAITING_CI (ids: %s)",
            report.workflows_exhausted, threshold,
            report.exhausted_workflow_ids,
        )
    return report


__all__ = [
    "CIPollExhaustionReport",
    "DEFAULT_AWAITING_CI_TIMEOUT_S",
    "run_ci_poll_exhaustion_tick",
]