"""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": ""}