fix(memory): ScreenMemory threshold hardening + stale layout purge (P0-3)

- Raised similarity_threshold from 0.90 to 0.95 in get_screen_type() to eliminate
  POST_DETAIL false-positives caused by embedding saturation.
- Implemented purge_stale_screens() with 24h TTL to remove layout drift.
- Added collection size cap (200 points) to store_screen() with automatic wipe
  when saturated to clear overfitting bias.
- Triggered purge_stale_screens() on ScreenIdentity initialization.

TDD: 3 new tests for threshold defaults and purge method integrity.
This commit is contained in:
2026-05-04 14:39:55 +02:00
parent d81a5d4e81
commit a265dee3d0
3 changed files with 105 additions and 1 deletions

View File

@@ -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

View File

@@ -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: