Spaces:
Sleeping
Sleeping
File size: 3,710 Bytes
8edee29 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """
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)
|