diff --git a/headroom/memory/traffic_learner.py b/headroom/memory/traffic_learner.py index bcecefef0..abd200ff3 100644 --- a/headroom/memory/traffic_learner.py +++ b/headroom/memory/traffic_learner.py @@ -26,6 +26,7 @@ import os import re import sqlite3 import time +from collections import OrderedDict from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum @@ -450,6 +451,7 @@ class TrafficLearner: max_history: int = 20, dedup_window: int = 100, min_evidence: int = 5, + max_pending_patterns: int = 2048, ) -> None: """Initialize the traffic learner. @@ -468,12 +470,19 @@ class TrafficLearner: self.agent_type = agent_type self._max_history = max_history self._min_evidence = min_evidence + self._max_pending_patterns = max_pending_patterns # Recent tool call history for error→recovery matching self._tool_history: list[dict[str, Any]] = [] - # Pattern accumulator: hash → (pattern, count) - self._pattern_counts: dict[str, tuple[ExtractedPattern, int]] = {} + # Pattern accumulator: hash → (pattern, count). LRU-ordered and capped: + # a pattern that is seen once but never reaches ``min_evidence`` would + # otherwise linger here forever, so this dict grew unbounded over a + # long-lived proxy's traffic (the sibling ``_saved_hashes`` is trimmed + # to ``dedup_window`` for the same reason; this one was missed). Evicting + # the least-recently-corroborated pending pattern is safe: if it recurs + # it simply restarts accumulation. + self._pattern_counts: OrderedDict[str, tuple[ExtractedPattern, int]] = OrderedDict() # Dedup: hashes of patterns already saved to DB self._saved_hashes: set[str] = set() @@ -1250,7 +1259,13 @@ class TrafficLearner: existing, count = self._pattern_counts[h] count += 1 self._pattern_counts[h] = (existing, count) + # Mark as most-recently-corroborated so it survives LRU eviction. + self._pattern_counts.move_to_end(h) else: + # Bound the pending accumulator so one-off patterns can't grow it + # without limit; drop the least-recently-corroborated pending entry. + if len(self._pattern_counts) >= self._max_pending_patterns: + self._pattern_counts.popitem(last=False) self._pattern_counts[h] = (pattern, 1) return # First sighting — wait for more evidence diff --git a/tests/test_memory/test_traffic_learner.py b/tests/test_memory/test_traffic_learner.py index eee21450b..fae5bc8fe 100644 --- a/tests/test_memory/test_traffic_learner.py +++ b/tests/test_memory/test_traffic_learner.py @@ -362,6 +362,56 @@ class TestTrafficLearner: stats = learner.get_stats() assert stats["patterns_extracted"] >= 3 + def test_pending_accumulator_is_bounded(self): + """One-off patterns that never reach ``min_evidence`` must not grow the + pending ``_pattern_counts`` accumulator without bound — the sibling + ``_saved_hashes`` is already trimmed to ``dedup_window`` and this one was + missed, so a long-lived proxy leaked memory across varied traffic. It is + now LRU-capped at ``max_pending_patterns``. + + Sync test (drives the async accumulate via ``asyncio.run``) so it runs + without the pytest-asyncio plugin. + """ + import asyncio + + learner = TrafficLearner(backend=None, min_evidence=5, max_pending_patterns=8) + + async def feed_one_offs() -> None: + for i in range(500): + await learner._accumulate( + ExtractedPattern( + category=PatternCategory.PREFERENCE, + content=f"one-off pattern number {i}", + importance=0.5, + ) + ) + + asyncio.run(feed_one_offs()) + assert len(learner._pattern_counts) <= 8 # capped, not 500 + + def test_pending_accumulator_lru_still_promotes_corroborated_pattern(self): + """Capping the accumulator must not break promotion: a pattern + corroborated to ``min_evidence`` without interruption is still removed + from pending and recorded in ``_saved_hashes``.""" + import asyncio + + learner = TrafficLearner(backend=None, min_evidence=3, max_pending_patterns=100) + pattern = ExtractedPattern( + category=PatternCategory.PREFERENCE, + content="corroborated preference", + importance=0.5, + ) + + async def corroborate() -> None: + await learner._accumulate(pattern) # count 1 + await learner._accumulate(pattern) # count 2 + assert pattern.content_hash in learner._pattern_counts + await learner._accumulate(pattern) # count 3 == min_evidence -> promote + + asyncio.run(corroborate()) + assert pattern.content_hash not in learner._pattern_counts # removed on promotion + assert pattern.content_hash in learner._saved_hashes + @pytest.mark.asyncio async def test_dedup(self, learner: TrafficLearner): """Test that identical patterns are deduplicated."""