diff --git a/GramAddict/core/perception/screen_identity.py b/GramAddict/core/perception/screen_identity.py index 2073b69..85db768 100644 --- a/GramAddict/core/perception/screen_identity.py +++ b/GramAddict/core/perception/screen_identity.py @@ -40,6 +40,8 @@ class ScreenIdentity: from GramAddict.core.qdrant_memory import ScreenMemoryDB self.screen_memory = ScreenMemoryDB() + if self.screen_memory: + self.screen_memory.purge_stale_screens() except ImportError: self.screen_memory = None diff --git a/GramAddict/core/qdrant_memory.py b/GramAddict/core/qdrant_memory.py index cc31eaf..4a10d0e 100644 --- a/GramAddict/core/qdrant_memory.py +++ b/GramAddict/core/qdrant_memory.py @@ -772,6 +772,16 @@ class ScreenMemoryDB(QdrantBase): if not vector: return + # โ”€โ”€ P0-3: Collection Management โ”€โ”€ + # Prevent embedding saturation by limiting to 200 high-quality signatures + try: + count = self.client.count(collection_name=self.collection_name).count + if count >= 200: + logger.warning("๐Ÿšจ [ScreenMemory] Collection saturated (>= 200). Wiping to clear overfitting bias.") + self.wipe_collection() + except Exception: + pass + self.upsert_point( seed_string=xml_signature, vector=vector, @@ -783,7 +793,7 @@ class ScreenMemoryDB(QdrantBase): log_success=f"๐Ÿง  [ScreenMemory] Learned new layout mapping: {screen_type}", ) - def get_screen_type(self, xml_signature: str, similarity_threshold: float = 0.90) -> Optional[str]: + def get_screen_type(self, xml_signature: str, similarity_threshold: float = 0.95) -> Optional[str]: if not self.is_connected or not xml_signature: return None @@ -809,6 +819,30 @@ class ScreenMemoryDB(QdrantBase): logger.debug(f"Screen memory error: {e}") return None + def purge_stale_screens(self, max_age_hours: float = 24): + """Removes entries older than max_age_hours to prevent layout drift poisoning.""" + if not self.is_connected: + return + try: + points, _ = self.client.scroll( + collection_name=self.collection_name, + limit=200, + with_payload=True, + ) + now = time.time() + purged = 0 + for pt in points: + payload = pt.payload or {} + stored_at = payload.get("stored_at", 0) + age_hours = (now - stored_at) / 3600 if stored_at > 0 else 0 + if age_hours > max_age_hours: + self.delete_point(payload.get("signature", str(pt.id))) + purged += 1 + if purged: + logger.info(f"๐Ÿงน [ScreenMemory] Purged {purged} stale layout mappings (> {max_age_hours}h).") + except Exception as e: + logger.debug(f"Screen purge error: {e}") + def purge_screen(self, xml_signature: str): """Unlearns a screen classification to recover from hallucinations.""" if not self.is_connected or not xml_signature: diff --git a/tests/core/test_screen_memory_hardening.py b/tests/core/test_screen_memory_hardening.py new file mode 100644 index 0000000..8af916d --- /dev/null +++ b/tests/core/test_screen_memory_hardening.py @@ -0,0 +1,68 @@ +""" +P0-3: ScreenMemory Cache Poisoning Guard + +TDD RED: These tests enforce that: +1. The similarity threshold for screen type cache hits is raised to 0.95 + to prevent false-positive POST_DETAIL matches. +2. A purge_stale_screens() method exists and removes entries older than + a configurable TTL. +3. The collection has a max size cap to prevent embedding saturation. +""" + +import time + +from GramAddict.core.qdrant_memory import ScreenMemoryDB + + +# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +# P0-3a: Raised Similarity Threshold +# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + + +class TestScreenMemoryThreshold: + """Session log: 34/60 screens classified as POST_DETAIL at 0.92-0.97 similarity.""" + + def test_default_threshold_is_095(self): + """ + RED: The default similarity_threshold in get_screen_type is 0.90, + which is too low and causes false-positive POST_DETAIL cache hits. + GREEN: Raise default to 0.95. + """ + import inspect + + sig = inspect.signature(ScreenMemoryDB.get_screen_type) + default = sig.parameters["similarity_threshold"].default + + assert default == 0.95, ( + f"Expected default similarity_threshold=0.95, got {default}. " + f"Threshold 0.90 causes POST_DETAIL over-matching." + ) + + +# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +# P0-3b: Staleness Purge +# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + + +class TestScreenMemoryStalePurge: + """Screen layouts change with IG updates; stale cache entries are poison.""" + + def test_purge_stale_screens_method_exists(self): + """ + RED: ScreenMemoryDB has no purge_stale_screens() method. + GREEN: Add a method that removes entries older than a configurable max_age_hours. + """ + db = ScreenMemoryDB() + assert hasattr(db, "purge_stale_screens"), ( + "ScreenMemoryDB must have a purge_stale_screens() method " + "to prevent cache poisoning from stale screen layouts." + ) + + def test_purge_stale_screens_has_max_age_parameter(self): + """The purge method must accept a max_age_hours parameter.""" + import inspect + + sig = inspect.signature(ScreenMemoryDB.purge_stale_screens) + assert "max_age_hours" in sig.parameters, ( + "purge_stale_screens must accept max_age_hours parameter" + )