orchestrator/src/labels.py

"""ORCH-089: auto-mode by Plane labels — autoApprove + autoDeploy (pure logic).

Leaf module — pure, unit-testable logic over the config flags + the Plane label
helpers in ``plane_sync``. Mirrors the leaf pattern of ``src/serial_gate.py`` /
``src/self_deploy.py``: imports only ``config`` (and lazily ``plane_sync`` /
``qg.checks`` / ``projects``), never ``stage_engine`` / ``launcher``.

What it decides (ADR-001 D1):
  * Whether the auto-mode is in scope for a repo (``auto_approve_applies`` /
    ``auto_deploy_applies``) — a LOCAL, network-free check evaluated FIRST.
  * Whether a given Plane label is present on an issue (``has_label``) — the only
    network call, made ONLY after ``applies()`` is True, so a disabled kill-switch
    costs zero network and yields zero regression (AC-8).

never-raise contract (BR-6/AC-6, fail-safe to the MANUAL gate): every public
function degrades to "no auto" on ANY error / ambiguity / Plane unavailability.
There is NO fail-open here — the conservative default is always "no auto"
(human gate stays), so an error can never auto-pass a gate.
"""
from __future__ import annotations

import logging

from .config import settings

logger = logging.getLogger("orchestrator.labels")


# ---------------------------------------------------------------------------
# Scope / kill-switch (mirrors self_deploy_applies / serial_gate_applies)
# ---------------------------------------------------------------------------
def _auto_label_applies(repo: str) -> bool:
    """Shared scope check for both auto-modes (ADR-001 D5).

      * ``auto_label_enabled=False`` -> always False (kill-switch; both gates 1:1
        as before ORCH-089, and — crucially — ``has_label`` is never consulted, so
        no new network call on the gate, AC-8).
      * ``auto_label_repos`` (CSV) non-empty -> real only for the listed repos.
      * empty CSV -> self-hosting only (``orchestrator``) — the safe default
        (the autoDeploy insertion lives in Phase A, which only exists for the
        self-hosting repo).
    Never raises -> False on error (degrade to "no auto" = manual gate).
    """
    try:
        if not getattr(settings, "auto_label_enabled", False):
            return False
        raw = (getattr(settings, "auto_label_repos", "") or "").strip()
        if raw:
            allowed = {r.strip().lower() for r in raw.split(",") if r.strip()}
            return (repo or "").strip().lower() in allowed
        # Lazy import keeps this module a leaf (avoids importing qg at load).
        from .qg.checks import is_self_hosting_repo
        return is_self_hosting_repo(repo)
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.warning("_auto_label_applies error for %s: %s", repo, e)
        return False


def auto_approve_applies(repo: str) -> bool:
    """Whether the autoApprove (BRD gate) auto-mode is in scope for ``repo``."""
    return _auto_label_applies(repo)


def auto_deploy_applies(repo: str) -> bool:
    """Whether the autoDeploy (prod-deploy gate) auto-mode is in scope for ``repo``."""
    return _auto_label_applies(repo)


# ---------------------------------------------------------------------------
# Label presence (the ONLY network call; ADR-001 D1)
# ---------------------------------------------------------------------------
def has_label(work_item_id: str, label_name: str, project_id: str | None = None) -> bool:
    """True iff the issue carries a label whose name == ``label_name`` (normalized).

    Resolution (all inside one ``try/except -> False``):
      1. ``plane_sync.fetch_issue_labels`` — the issue's label uuids (None on error
         -> False);
      2. ``plane_sync.get_project_labels`` — {normalized_name -> uuid} project map
         (TTL-cached);
      3. normalize the sought name and look it up in the project map;
      4. no match, OR an ambiguous name (the project map maps it to the
         ``__AMBIGUOUS__`` sentinel) -> False (fail-safe);
      5. ``return target_uuid in set(labels)``.

    Any error / unavailability / ambiguity -> **False** (never auto on doubt).
    """
    if not label_name:
        return False
    try:
        from . import plane_sync
        labels = plane_sync.fetch_issue_labels(work_item_id, project_id)
        if labels is None:
            # Could not read the issue's labels -> fail-safe to manual.
            return False
        if not labels:
            return False
        name_map = plane_sync.get_project_labels(
            plane_sync._resolve_project_id(work_item_id, project_id)
        )
        if not name_map:
            return False
        normalized = plane_sync._normalize_label(label_name)
        target_uuid = name_map.get(normalized)
        if not target_uuid or target_uuid == "__AMBIGUOUS__":
            return False
        return target_uuid in set(labels)
    except Exception as e:  # noqa: BLE001 - never-raise -> no auto
        logger.warning(
            "has_label error for %s/%s -> fail-safe (no auto): %s",
            work_item_id, label_name, e,
        )
        return False


# ---------------------------------------------------------------------------
# Observability snapshot for GET /queue (ADR-001 D7)
# ---------------------------------------------------------------------------
def snapshot() -> dict:
    """Read-only auto-label summary for GET /queue (additive block). never-raise."""
    try:
        enabled = bool(getattr(settings, "auto_label_enabled", False))
    except Exception:  # noqa: BLE001
        enabled = False
    try:
        return {
            "enabled": enabled,
            "approve_label": getattr(settings, "auto_approve_label", ""),
            "deploy_label": getattr(settings, "auto_deploy_label", ""),
            "repos": getattr(settings, "auto_label_repos", "") or "",
        }
    except Exception as e:  # noqa: BLE001 - never-raise -> minimal dict
        logger.warning("labels snapshot error: %s", e)
        return {"enabled": enabled, "approve_label": "", "deploy_label": "", "repos": ""}