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

"""Controller-managed opt-in label gate (Phase 1k).

For migration safety, the controller only manages PRs and issues that
carry a configurable opt-in label (default: ``controller-managed``).
This lets operators:

- Opt one PR in for a parallel-run trial (Phase 2 migration mechanism).
- Roll out gradually without exposing the controller to PRs that
  human reviewers are actively driving.
- Pause management of any PR mid-flight by removing the label.

The label name comes from ``CONTROLLER_OPT_IN_LABEL`` env var; default
is ``controller-managed``. Operators can override per-deployment.

What this module ships:

- ``opt_in_label_name()`` — single source of truth for the configured
  label name.
- ``has_opt_in_label(entity_dict, label_name)`` — predicate over a
  Forgejo PR/issue dict's ``labels`` field. Defensive about shape.
- ``filter_by_opt_in_label(entities, label_name)`` — list comprehension
  wrapper for discovery/backfill.

Discovery (Phase 1d) + backfill (Phase 1f) are wired to use this; the
reconciliation tick (Phase 1g) checks for label removal post-discovery
and transitions opted-out workflows to ABANDONED with reason
``opt-in-label-removed``.
"""
from __future__ import annotations

import os
from typing import Any

DEFAULT_OPT_IN_LABEL = "controller-managed"


def opt_in_label_name() -> str:
    """Return the configured opt-in label name.

    Reads ``CONTROLLER_OPT_IN_LABEL`` env var; falls back to
    ``controller-managed``. Empty or whitespace-only env value
    falls back to the default (operators MUST set a non-empty name)."""
    raw = os.environ.get("CONTROLLER_OPT_IN_LABEL", "").strip()
    return raw or DEFAULT_OPT_IN_LABEL


def has_opt_in_label(entity: dict | None, label_name: str | None = None) -> bool:
    """True iff ``entity['labels']`` contains a label with the given
    name (case-sensitive, since Forgejo label names are case-sensitive).

    Defensive shape handling — returns False for any of:
    - None entity
    - non-dict entity
    - missing or non-list ``labels`` field
    - label entries that aren't dicts
    - label entries with a missing or non-string ``name`` field
    """
    name = label_name if label_name is not None else opt_in_label_name()
    if not isinstance(entity, dict):
        return False
    labels = entity.get("labels")
    if not isinstance(labels, list):
        return False
    for lbl in labels:
        if not isinstance(lbl, dict):
            continue
        lbl_name = lbl.get("name")
        if isinstance(lbl_name, str) and lbl_name == name:
            return True
    return False


def filter_by_opt_in_label(
    entities: list[dict], label_name: str | None = None,
) -> list[dict]:
    """Return only the entities carrying the opt-in label.

    Logs the skip count at INFO level via the caller's choice — this
    is a pure function. Callers (discovery, backfill) log the
    filtered-out total themselves."""
    name = label_name if label_name is not None else opt_in_label_name()
    return [e for e in entities if has_opt_in_label(e, name)]


def count_filtered(entities: list[dict], label_name: str | None = None) -> int:
    """Count entities filtered OUT by the opt-in label gate. Useful
    for discovery/backfill reports."""
    name = label_name if label_name is not None else opt_in_label_name()
    if not isinstance(entities, list):
        return 0
    return sum(1 for e in entities if not has_opt_in_label(e, name))


__all__ = [
    "DEFAULT_OPT_IN_LABEL",
    "count_filtered",
    "filter_by_opt_in_label",
    "has_opt_in_label",
    "opt_in_label_name",
]