orchestrator/src/bug_fast_track.py

"""ORCH-019: bug-fast-track — a cheaper/shorter pipeline route for bug-fix tasks.

Leaf module — pure, unit-testable logic over the config flags + the proven Plane
label apparatus (``labels.has_label`` -> ``plane_sync``, ORCH-089). Mirrors the
leaf pattern of ``src/labels.py`` / ``src/serial_gate.py``: imports only
``config`` (and lazily ``labels`` / ``db`` / ``qg.checks``), never
``stage_engine`` / ``launcher``.

What it decides (ADR-001):
  * Whether the bug-fast-track is in scope for a repo (``bug_fast_track_applies``)
    — a LOCAL, network-free check evaluated FIRST.
  * Whether a given Plane issue carries the ``Bug`` label (``is_bug_task``) — the
    only network call, made ONLY after ``applies()`` is True, so a disabled
    kill-switch costs zero network and yields zero regression (AC-6).
  * Whether a task's stored track skips the ``architecture`` stage
    (``skips_architecture``) — a pure predicate over the DB-stored ``track``,
    read in the hot ``advance_stage`` path WITHOUT any network call (NFR-4).

never-raise contract (BR-6/AC-6, fail-safe to the FULL cycle): every public
function degrades to "full cycle" on ANY error / ambiguity / Plane
unavailability / disabled flag. There is NO fail-open here — the conservative
default is always the full pipeline (with ``architecture``), so an error can
never silently skip a stage.
"""
from __future__ import annotations

import logging

from .config import settings

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


# ---------------------------------------------------------------------------
# Scope / kill-switch (mirrors _auto_label_applies / serial_gate_applies)
# ---------------------------------------------------------------------------
def bug_fast_track_applies(repo: str) -> bool:
    """Whether the bug-fast-track is REAL for ``repo`` (ADR-001 D6 / AC-6).

      * ``bug_fast_track_enabled=False`` -> always False (kill-switch; start and
        routing are 1:1 as before ORCH-019, and — crucially — ``has_label`` is
        never consulted, so no new network call on start, AC-6).
      * ``bug_fast_track_repos`` (CSV) non-empty -> real only for the listed repos.
      * empty CSV -> self-hosting only (``orchestrator``) — the safe default (the
        track is first burnt in on the orchestrator itself, where the `Bug` label
        is guaranteed to exist; enduro opts in via an explicit CSV entry).
    Checked FIRST (local, network-free); never raises -> False on error (degrade
    to "full cycle", which matches the kill-switch-off behaviour).
    """
    try:
        if not getattr(settings, "bug_fast_track_enabled", False):
            return False
        raw = (getattr(settings, "bug_fast_track_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 -> full cycle
        logger.warning("bug_fast_track_applies error for %s: %s", repo, e)
        return False


# ---------------------------------------------------------------------------
# Classification (the ONLY network call; ADR-001 D1)
# ---------------------------------------------------------------------------
def is_bug_task(work_item_id: str, project_id: str | None = None) -> bool:
    """True iff the issue carries the configured ``Bug`` label (Plane API source).

    ``bug_fast_track_applies`` is assumed already True (checked by the caller —
    the gate idiom ``applies(repo) and is_bug_task(...)`` short-circuits before any
    network call when the kill-switch is off). Delegates to the proven
    ``labels.has_label`` (fetch_issue_labels + get_project_labels, normalization,
    TTL-cache, source-of-truth = Plane API, not the webhook payload).

    Any error / ambiguity / Plane unavailability -> **False** (fail-safe -> full
    cycle, never silently fast-track on doubt).
    """
    try:
        label = (getattr(settings, "bug_fast_track_label", "") or "").strip()
        if not label:
            return False
        from . import labels
        return bool(labels.has_label(work_item_id, label, project_id))
    except Exception as e:  # noqa: BLE001 - never-raise -> full cycle
        logger.warning(
            "is_bug_task error for %s -> fail-safe (full cycle): %s", work_item_id, e
        )
        return False


# ---------------------------------------------------------------------------
# Routing predicate (pure, DB-backed; hot path — NO network, NFR-4) — ADR-001 D3
# ---------------------------------------------------------------------------
def skips_architecture(track: str | None) -> bool:
    """Whether a task with stored ``track`` skips the ``architecture`` stage.

    Pure predicate (no I/O): True iff the kill-switch is on AND ``track == 'bug'``.
    Used by ``advance_stage`` on the analysis-exit edge to map
    ``analysis -> architecture`` to ``analysis -> development`` for a bug task.
    A disabled flag -> always False (1:1 prior routing); any error -> False
    (fail-safe -> full cycle).
    """
    try:
        if not getattr(settings, "bug_fast_track_enabled", False):
            return False
        return (track or "").strip().lower() == "bug"
    except Exception as e:  # noqa: BLE001 - never-raise -> full cycle
        logger.warning("skips_architecture error for track=%r: %s", track, e)
        return False


# ---------------------------------------------------------------------------
# Observability snapshot for GET /queue (ADR-001 D7)
# ---------------------------------------------------------------------------
def snapshot() -> dict:
    """Read-only bug-fast-track summary for GET /queue (additive block). never-raise.

    Surfaces the flags + a savings metric derived from the existing telemetry: the
    count of tasks on the bug track and the number of ``architecture`` agent runs
    those tasks structurally skipped (one per bug task = ``est_saved_architecture_runs``).
    Any error -> a minimal dict with the flags (never crashes the endpoint).
    """
    try:
        enabled = bool(getattr(settings, "bug_fast_track_enabled", False))
    except Exception:  # noqa: BLE001
        enabled = False
    try:
        label = getattr(settings, "bug_fast_track_label", "Bug") or "Bug"
    except Exception:  # noqa: BLE001
        label = "Bug"
    try:
        repos_cfg = getattr(settings, "bug_fast_track_repos", "") or ""
    except Exception:  # noqa: BLE001
        repos_cfg = ""
    active_bug_tasks = 0
    total_bug_tasks = 0
    try:
        from . import db
        conn = db.get_db()
        try:
            # ORCH-090 terminal set {done,cancelled}: "active" = not terminal.
            row = conn.execute(
                "SELECT "
                "  COUNT(*) AS total, "
                "  SUM(CASE WHEN stage NOT IN ('done','cancelled') THEN 1 ELSE 0 END) AS active "
                "FROM tasks WHERE track = 'bug'"
            ).fetchone()
            if row:
                total_bug_tasks = int(row["total"] or 0)
                active_bug_tasks = int(row["active"] or 0)
        finally:
            conn.close()
    except Exception as e:  # noqa: BLE001
        logger.warning("bug_fast_track snapshot count error: %s", e)
    return {
        "enabled": enabled,
        "label": label,
        "repos": repos_cfg,
        "active_bug_tasks": active_bug_tasks,
        "total_bug_tasks": total_bug_tasks,
        # Each bug task skips exactly one `architecture` stage (one architect agent
        # run + ADR). This is the structural savings the track buys (FR-7 / AC-7).
        "est_saved_architecture_runs": total_bug_tasks,
    }