utils/error_cache.py

"""
utils/error_cache.py
--------------------
Error classification cache for AutoDevAgent.

Tracks error fingerprints (cache_keys) across debug iterations within
a single pipeline run. When the same error appears in consecutive
iterations, the debug agent has already tried one fix that did not work
— continuing with the same strategy is pointless.

When a repeated error is detected, the DebugAgent escalates to a
fundamentally different fix strategy rather than patching the same line.

Design:
  - Keyed by the normalised error fingerprint (cache_key from
    ErrorClassification, e.g. "NameError:x_not_defined").
  - Records the count of how many times each key has been seen.
  - "Repeated" means seen more than once — i.e. the same error
    appeared in at least two consecutive iterations.
  - Reset between pipeline runs so state does not bleed across tasks.

Usage:
    from utils.error_cache import ErrorCache

    cache = ErrorCache()
    cache.record("NameError:x_not_defined")
    cache.is_repeated("NameError:x_not_defined")  # False (seen once)
    cache.record("NameError:x_not_defined")
    cache.is_repeated("NameError:x_not_defined")  # True (seen twice)
    cache.reset()
"""

import logging
from collections import defaultdict

logger = logging.getLogger(__name__)


class ErrorCache:
    """
    Tracks how many times each error fingerprint has been seen.

    One instance is created per module (in debug_agent.py) and shared
    across all DebugAgent calls within a session. Call reset() at the
    start of each new pipeline run to clear stale state.

    Attributes:
        _counts: Dict mapping cache_key → number of times seen.
    """

    def __init__(self) -> None:
        """Initialise an empty cache."""
        self._counts: dict[str, int] = defaultdict(int)

    def record(self, cache_key: str) -> None:
        """
        Record one occurrence of an error fingerprint.

        Args:
            cache_key: Normalised error fingerprint from ErrorClassification.
                       e.g. "NameError:x_not_defined"
        """
        self._counts[cache_key] += 1
        logger.debug(
            "ErrorCache recorded '%s' (count: %d)",
            cache_key,
            self._counts[cache_key],
        )

    def is_repeated(self, cache_key: str) -> bool:
        """
        Return True if this error has been seen more than once.

        A count of 1 means it appeared this iteration for the first time.
        A count of 2+ means a previous fix attempt did not resolve it.

        Args:
            cache_key: Normalised error fingerprint to check.

        Returns:
            True if seen more than once, False otherwise.
        """
        return self._counts[cache_key] > 1

    def count(self, cache_key: str) -> int:
        """
        Return how many times an error fingerprint has been seen.

        Args:
            cache_key: Normalised error fingerprint to look up.

        Returns:
            Integer count, 0 if never seen.
        """
        return self._counts[cache_key]

    def reset(self) -> None:
        """
        Clear all recorded error counts.

        Call this at the start of each new pipeline run to prevent
        error state from one task bleeding into the next.
        """
        cleared = len(self._counts)
        self._counts.clear()
        logger.debug("ErrorCache reset — cleared %d entries", cleared)

    def summary(self) -> dict[str, int]:
        """
        Return a snapshot of all recorded error counts.

        Useful for session history logging and debugging.

        Returns:
            Dict of cache_key → count for all recorded errors.
        """
        return dict(self._counts)