tracer-cloud--opensre/tools/system/fleet_monitoring/error_signals.py

"""Per-category error/retry rate detector for monitored agent streams."""

from __future__ import annotations

import re
import threading
import time
from collections import deque
from collections.abc import Callable, Iterable
from dataclasses import dataclass, field

_DEFAULT_WINDOW_SECONDS = 60.0


@dataclass(frozen=True, slots=True)
class ErrorCategory:
    name: str
    keywords: tuple[str, ...] = field(default=())
    patterns: tuple[re.Pattern[str], ...] = field(default=())


# Patterns require error-shape context (adjacent verb, status prefix, header
# form, etc.) so descriptive prose like "the OpenAI rate limit is 10k tokens"
# or "add tool error handling" does not fire. See #1497 acceptance criterion.
DEFAULT_CATEGORIES: tuple[ErrorCategory, ...] = (
    ErrorCategory(
        name="rate_limit",
        patterns=(
            re.compile(
                r"\brate[\s_-]?limit(?:ed|ing)?\b\s*"
                r"(?:exceeded|hit|reached|encountered|triggered|error|429)",
                re.IGNORECASE,
            ),
            re.compile(r"\bratelimit(?:ed|ing)\b", re.IGNORECASE),
            re.compile(
                r"\b429\b\s*(?::|-|—|too\s+many\s+requests|rate)",
                re.IGNORECASE,
            ),
        ),
    ),
    ErrorCategory(
        name="http_5xx",
        patterns=(
            re.compile(
                r"(?:"
                r"\bHTTP[/ ]\d(?:\.\d)?\s+5\d{2}\b"
                r"|\bstatus(?:\s*code)?[: ]+5\d{2}\b"
                r"|\b5\d{2}\s+(?:internal\s+server\s+error|bad\s+gateway|"
                r"service\s+unavailable|gateway\s+timeout|server\s+error)\b"
                r")",
                re.IGNORECASE,
            ),
        ),
    ),
    ErrorCategory(
        name="tool_failure",
        patterns=(
            re.compile(r"\btool[\s_-]?failure\b\s*[:\-—]\s*\S", re.IGNORECASE),
            re.compile(
                r"\btool\s+failed\b\s+(?:during|at|with|in|on|while|"
                r"after|when|to|for|because)\b",
                re.IGNORECASE,
            ),
            re.compile(r"\btool\s+exited\s+with\s+(?:code\s+)?\d+", re.IGNORECASE),
        ),
    ),
    ErrorCategory(
        name="traceback",
        patterns=(re.compile(r"Traceback\s*\(most\s+recent\s+call\s+last\)"),),
    ),
)


class ErrorSignals:
    """Sliding-window error/retry rate tracker for an agent stdout stream.

    Thread-safe: a ``threading.Lock`` guards the prune-and-mutate sections
    of ``observe()`` and ``rate_per_minute()``. Regex matching runs outside
    the lock so a heavy chunk does not block the renderer thread.
    """

    __slots__ = (
        "_categories",
        "_window_seconds",
        "_now",
        "_events",
        "_lock",
        "_keyword_patterns",
    )

    def __init__(
        self,
        *,
        categories: Iterable[ErrorCategory] | None = None,
        window_seconds: float = _DEFAULT_WINDOW_SECONDS,
        now: Callable[[], float] | None = None,
    ) -> None:
        if window_seconds <= 0:
            raise ValueError("window_seconds must be > 0")
        resolved = tuple(categories if categories is not None else DEFAULT_CATEGORIES)
        seen: set[str] = set()
        for cat in resolved:
            if cat.name in seen:
                raise ValueError(f"duplicate category name: {cat.name!r}")
            seen.add(cat.name)
        self._categories: tuple[ErrorCategory, ...] = resolved
        self._window_seconds = float(window_seconds)
        self._now = now if now is not None else time.monotonic
        self._events: dict[str, deque[float]] = {cat.name: deque() for cat in self._categories}
        self._lock = threading.Lock()
        # Pre-compile keywords with word boundaries so "error" does not match
        # "errored" / "errorless" / "noerror". Keeps custom-category keyword
        # use safe against adversarial substring inflation in agent stdout.
        self._keyword_patterns: dict[str, tuple[re.Pattern[str], ...]] = {
            cat.name: tuple(
                re.compile(r"\b" + re.escape(kw) + r"\b", re.IGNORECASE)
                for kw in cat.keywords
                if kw
            )
            for cat in self._categories
        }

    def observe(self, chunk: str) -> None:
        if not chunk:
            return
        timestamp = self._now()
        cutoff = timestamp - self._window_seconds

        # Run regex matching outside the lock so a heavy chunk does not block
        # the renderer thread.
        hits_by_category: list[tuple[str, int]] = []
        for category in self._categories:
            hits = self._count_hits(category, chunk)
            if hits:
                hits_by_category.append((category.name, hits))

        # Prune in observe() too so an idle dashboard does not grow unbounded.
        with self._lock:
            for bucket in self._events.values():
                while bucket and bucket[0] < cutoff:
                    bucket.popleft()
            for name, hits in hits_by_category:
                self._events[name].extend([timestamp] * hits)

    def rate_per_minute(self) -> dict[str, float]:
        cutoff = self._now() - self._window_seconds
        scale = 60.0 / self._window_seconds
        with self._lock:
            rates: dict[str, float] = {}
            for name, bucket in self._events.items():
                while bucket and bucket[0] < cutoff:
                    bucket.popleft()
                rates[name] = len(bucket) * scale
        return rates

    def _count_hits(self, category: ErrorCategory, chunk: str) -> int:
        total = 0
        # Keywords match with word boundaries via pre-compiled patterns.
        for kw_pattern in self._keyword_patterns[category.name]:
            total += sum(1 for _ in kw_pattern.finditer(chunk))
        # finditer is unambiguous regardless of capturing-group structure;
        # findall would return group contents instead of full matches if a
        # future pattern adds a non-(?:...) group.
        for pattern in category.patterns:
            total += sum(1 for _ in pattern.finditer(chunk))
        return total