server/heuristics.py

"""
Deterministic heuristics for bug-triage inference and action suggestion.

This module is the single source of truth for all rule-based logic that
maps ticket attributes to severity, priority, component, and team.  Both
the standalone ``inference.py`` runner and the FastAPI ``server/app.py``
import from here so that fixes only need to be applied in one place.
"""

from __future__ import annotations

# ---------------------------------------------------------------------------
# Lookup tables
# ---------------------------------------------------------------------------

#: Maps a component slug to the engineering team responsible for it.
COMPONENT_TEAM_MAP: dict[str, str] = {
    "api-gateway": "backend-api",
    "auth-service": "backend-api",
    "user-service": "backend-api",
    "payment-service": "backend-api",
    "web-app": "frontend-web",
    "ios-app": "mobile-ios",
    "android-app": "mobile-android",
    "database": "data-platform",
    "cache": "infrastructure",
    "cdn": "infrastructure",
}

#: Free-text keywords that strongly suggest a component when found in a ticket.
COMPONENT_KEYWORDS: dict[str, tuple[str, ...]] = {
    "api-gateway": ("api gateway", "gateway", "edge", "proxy", "routing", "/api/"),
    "auth-service": ("auth", "authentication", "login", "signin", "token", "session"),
    "user-service": ("user service", "users endpoint", "profile", "identity", "account"),
    "payment-service": ("payment", "checkout", "charge", "billing", "tax", "order"),
    "web-app": ("web app", "web-app", "browser", "dashboard", "frontend", "page"),
    "ios-app": ("ios", "iphone", "ipad", "apple"),
    "android-app": ("android",),
    "database": ("database", "query", "sql", "db", "index"),
    "cache": ("cache", "redis", "memcache"),
    "cdn": ("cdn", "image", "asset", "static content"),
}

#: A service name may hint that certain components are likely candidates.
SERVICE_COMPONENT_HINTS: dict[str, tuple[str, ...]] = {
    "api": ("api-gateway", "user-service"),
    "auth": ("auth-service",),
    "identity": ("user-service",),
    "payments": ("payment-service",),
    "web-app": ("web-app", "cdn", "database", "cache"),
    "mobile-app": ("ios-app", "android-app", "auth-service"),
}


# ---------------------------------------------------------------------------
# Helper — build a single lowercase string from a ticket for keyword search
# ---------------------------------------------------------------------------

def ticket_text(ticket) -> str:
    """Return a single lowercase string combining ticket fields for keyword search."""
    return " ".join(
        str(part)
        for part in (
            ticket.title,
            ticket.description,
            ticket.service,
            " ".join(ticket.component_candidates),
        )
    ).lower()


# ---------------------------------------------------------------------------
# Component inference
# ---------------------------------------------------------------------------

def infer_component(ticket, available_components: list[str]) -> str:
    """
    Pick the most likely component for *ticket* from *available_components*.

    Scoring heuristic (higher wins):
    - Position bonus: earlier in ``component_candidates`` ← up to +3
    - Service hint match                                   ← +3
    - Component name appears in ticket text                ← +4
    - Keyword match                                        ← +3 per keyword
    - Ad-hoc tie-breakers for common patterns              ← +1
    """
    candidates = [c for c in ticket.component_candidates if c in available_components]
    if not candidates:
        return available_components[0] if available_components else "api-gateway"

    text = ticket_text(ticket)
    service_hints = SERVICE_COMPONENT_HINTS.get(ticket.service, ())
    best_candidate = candidates[0]
    best_score = -1

    for index, candidate in enumerate(candidates):
        score = max(0, 3 - index)  # position bonus

        if candidate in service_hints:
            score += 3

        normalized = candidate.replace("-", " ")
        if normalized in text:
            score += 4

        for keyword in COMPONENT_KEYWORDS.get(candidate, ()):
            if keyword in text:
                score += 3

        # Ad-hoc tie-breakers
        if candidate == "ios-app" and "login" in text:
            score += 1
        if candidate == "payment-service" and "gateway" in text:
            score += 1
        if candidate == "database" and "slow" in text:
            score += 1

        if score > best_score:
            best_score = score
            best_candidate = candidate

    return best_candidate


# ---------------------------------------------------------------------------
# Severity inference
# ---------------------------------------------------------------------------

def infer_severity(ticket, component: str) -> str:
    """
    Return the inferred severity level (``sev0``–``sev3``) for *ticket*.

    Rules are applied in priority order; the first match wins.
    """
    text = ticket_text(ticket)
    synthetic_high_signal = "signal quality is high" in text
    synthetic_low_signal = "signal quality is low" in text

    # sev0 — data / security / money corruption
    if any(k in text for k in ["security", "unauthorized", "double charge", "data loss", "corrupt"]):
        return "sev0"

    # sev0 — critical monitoring alert on enterprise
    if any(k in text for k in ["500 internal server error", "null pointer exception", "multiple monitoring alerts"]):
        if ticket.reporter_type == "monitoring" and ticket.customer_tier == "enterprise":
            return "sev0"

    # sev1 — high-signal synthetic flag on enterprise monitoring
    if synthetic_high_signal and ticket.reporter_type == "monitoring" and ticket.customer_tier == "enterprise":
        return "sev1"

    # sev1 — service down / timeout / outage
    if any(k in text for k in ["timeout", "timing out", "503", "outage", "down"]):
        return "sev1"

    # sev1 — high-signal on paid tiers
    if synthetic_high_signal and ticket.customer_tier in {"pro", "enterprise"}:
        return "sev1"

    # sev1/sev2 — incorrect tax (tier-dependent)
    if "incorrect tax" in text or ("tax" in text and "wrong" in text):
        return "sev1" if ticket.customer_tier in {"pro", "enterprise"} else "sev2"

    # sev2/sev3 — low-signal synthetic flag
    if synthetic_low_signal:
        return "sev3" if ticket.customer_tier == "free" else "sev2"

    # sev2 — clearly broken functionality
    if any(k in text for k in ["crash", "not responding", "not working", "broken image", "wrong values"]):
        return "sev2"

    # sev2 (or sev3 for free DB slowness) — degraded performance
    if any(k in text for k in ["latency", "slow", "degraded", "error", "failed"]):
        if component == "database" and ticket.customer_tier == "free":
            return "sev3"
        return "sev2"

    return "sev3"


# ---------------------------------------------------------------------------
# Priority inference
# ---------------------------------------------------------------------------

_SEV_TO_PRI: dict[str, str] = {
    "sev0": "p0",
    "sev1": "p1",
    "sev2": "p2",
    "sev3": "p3",
}


def severity_to_priority(severity: str) -> str:
    """Return the default priority matching *severity*."""
    return _SEV_TO_PRI.get(severity, "p2")


def infer_priority(ticket, severity: str) -> str:
    """
    Return the inferred priority for *ticket* given its *severity*.

    Upgrades sev2 → p1 when the ticket is enterprise, monitoring-sourced,
    or touches high-value areas (payments, CDN, images, shopping).
    """
    text = ticket_text(ticket)
    if severity == "sev2" and (
        ticket.customer_tier == "enterprise"
        or ticket.reporter_type == "monitoring"
        or any(k in text for k in ["payment", "checkout", "tax", "cdn", "image", "shopping"])
    ):
        return "p1"
    return severity_to_priority(severity)


# ---------------------------------------------------------------------------
# Info-request heuristics
# ---------------------------------------------------------------------------

def needs_more_info(ticket) -> bool:
    """Return ``True`` if the ticket is missing evidence needed for triage."""
    text = ticket_text(ticket)

    # Duplicates are handled separately; skip info request
    if ticket.suspected_duplicate_ids:
        return False

    if "signal quality is low" in text:
        return True

    if not ticket.repro_steps_present and not ticket.logs_present:
        return True

    # Non-monitoring reporters need repro steps
    if not ticket.repro_steps_present and ticket.reporter_type != "monitoring":
        return True

    # User/QA reporters should supply logs
    if not ticket.logs_present and ticket.reporter_type in {"user", "qa"}:
        return True

    return False


def suggested_info_type(ticket) -> str:
    """Return which type of information to request for *ticket*."""
    if not ticket.repro_steps_present and not ticket.logs_present:
        return "both"
    if not ticket.repro_steps_present:
        return "repro_steps"
    if not ticket.logs_present:
        return "logs"
    return "both"


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

__all__ = [
    "COMPONENT_TEAM_MAP",
    "COMPONENT_KEYWORDS",
    "SERVICE_COMPONENT_HINTS",
    "ticket_text",
    "infer_component",
    "infer_severity",
    "infer_priority",
    "severity_to_priority",
    "needs_more_info",
    "suggested_info_type",
]