feat(core): P0 radical evolution — kill blank_start, complete ContextGate matrix, fix HD Map back transitions
P0-2: Kill blank_start: true in config — the nuclear option that wipes ALL learned knowledge on every run. Replaced with memory_hygiene() selective amnesia that prunes low-confidence entries while preserving high-confidence learned patterns. P0-3: Purge all root-level garbage scripts (scratch.py, test_run.py, profile_dump.xml, resp_dump.json, pytest_output.log, test_output.log, coverage.xml, coverage_e2e.json). Updated .gitignore to prevent reaccumulation. P2-2: Replace piecemeal BANNED_SCREENS/REQUIRED_MARKERS with complete Action Compatibility Matrix (VALID_SCREENS whitelist). Every interaction intent now has an explicit set of valid screens. Covers like, comment, follow, unfollow, save, repost across all 14 screen types. P2-3: Remove non-deterministic 'press back' transitions from HD Map topology for OTHER_PROFILE, POST_DETAIL, and SEARCH_RESULTS. Add tab transitions to POST_DETAIL. P2-4: Replace ScreenMemoryDB nuclear 200-entry wipe with LRU eviction. store_screen() now accepts confidence parameter. TDD: 22 new tests (all GREEN), 240 unit/tdd/integration passed, 0 regressions. E2E: 288 passed (+2 fixed), 6 pre-existing LIE DETECTED failures.
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, Set
|
||||
|
||||
from GramAddict.core.perception.screen_identity import ScreenType
|
||||
|
||||
@@ -13,10 +13,25 @@ class ContextGate:
|
||||
|
||||
Zero-Trust: If the required structural markers aren't in the XML, the action
|
||||
is blocked, even if the LLM/VLM thinks it's possible.
|
||||
|
||||
Architecture: Uses a whitelist-based Action Compatibility Matrix.
|
||||
Only screens that structurally support an interaction allow it.
|
||||
"""
|
||||
|
||||
# ── Action Compatibility Matrix (Whitelist) ──
|
||||
# Maps each interaction intent to the set of screens where it's structurally valid.
|
||||
# If an intent is NOT in this map, it's a navigation/system intent and passes through.
|
||||
VALID_SCREENS: Dict[str, Set[ScreenType]] = {
|
||||
"like": {ScreenType.HOME_FEED, ScreenType.POST_DETAIL},
|
||||
"comment": {ScreenType.HOME_FEED, ScreenType.POST_DETAIL, ScreenType.COMMENTS},
|
||||
"follow": {ScreenType.OTHER_PROFILE, ScreenType.FOLLOW_LIST},
|
||||
"unfollow": {ScreenType.OTHER_PROFILE, ScreenType.FOLLOW_LIST},
|
||||
"save": {ScreenType.HOME_FEED, ScreenType.POST_DETAIL},
|
||||
"repost": {ScreenType.HOME_FEED, ScreenType.POST_DETAIL},
|
||||
}
|
||||
|
||||
# Intent -> List of resource-id fragments that MUST be present on the screen
|
||||
# to even consider performing this action.
|
||||
# to even consider performing this action (structural proof).
|
||||
REQUIRED_MARKERS = {
|
||||
"comment": ["comment", "row_feed_button_comment", "shell_comment_button"],
|
||||
"like": ["like", "heart", "row_feed_button_like", "shell_like_button"],
|
||||
@@ -25,14 +40,6 @@ class ContextGate:
|
||||
"save": ["save", "bookmark"],
|
||||
}
|
||||
|
||||
# Intent -> Screens where this action is CATEGORICALLY BANNED
|
||||
BANNED_SCREENS = {
|
||||
"follow": [ScreenType.OWN_PROFILE, ScreenType.DM_THREAD, ScreenType.DM_INBOX],
|
||||
"unfollow": [ScreenType.OWN_PROFILE, ScreenType.DM_THREAD, ScreenType.DM_INBOX],
|
||||
"comment": [ScreenType.OWN_PROFILE, ScreenType.DM_THREAD, ScreenType.DM_INBOX],
|
||||
"like": [ScreenType.DM_THREAD, ScreenType.DM_INBOX],
|
||||
}
|
||||
|
||||
def is_allowed(self, intent: str, screen_state: Dict[str, Any]) -> bool:
|
||||
"""
|
||||
Evaluates the context gate.
|
||||
@@ -48,17 +55,22 @@ class ContextGate:
|
||||
screen_type = screen_state.get("screen_type", ScreenType.UNKNOWN)
|
||||
resource_ids = screen_state.get("resource_ids", set())
|
||||
|
||||
# 1. Check categorical bans
|
||||
for blocked_intent, banned_types in self.BANNED_SCREENS.items():
|
||||
if blocked_intent in intent_lower and screen_type in banned_types:
|
||||
logger.warning(f"🛡️ [ContextGate] Blocked '{intent}' on {screen_type.name}: Categorical ban.")
|
||||
return False
|
||||
# 1. Action Compatibility Matrix (Whitelist Check)
|
||||
# If the intent matches a known interaction type, check if the screen allows it.
|
||||
for interaction_type, valid_screens in self.VALID_SCREENS.items():
|
||||
if interaction_type in intent_lower:
|
||||
if screen_type not in valid_screens:
|
||||
logger.warning(
|
||||
f"🛡️ [ContextGate] Blocked '{intent}' on {screen_type.name}: "
|
||||
f"Not in valid screens {[s.name for s in valid_screens]}."
|
||||
)
|
||||
return False
|
||||
break # Found the interaction type, don't check others
|
||||
|
||||
# 2. Check structural requirements
|
||||
# Only check for specific interaction intents
|
||||
# 2. Structural Marker Verification
|
||||
# Even on a valid screen, the required UI elements must be present.
|
||||
for required_intent, markers in self.REQUIRED_MARKERS.items():
|
||||
if required_intent in intent_lower:
|
||||
# Does ANY marker match any resource-id?
|
||||
has_marker = False
|
||||
for marker in markers:
|
||||
for rid in resource_ids:
|
||||
|
||||
@@ -765,7 +765,7 @@ class ScreenMemoryDB(QdrantBase):
|
||||
def __init__(self):
|
||||
super().__init__(collection_name="gramaddict_screen_types_v1")
|
||||
|
||||
def store_screen(self, xml_signature: str, screen_type: str):
|
||||
def store_screen(self, xml_signature: str, screen_type: str, confidence: float = 0.7):
|
||||
if not self.is_connected or not xml_signature:
|
||||
return
|
||||
|
||||
@@ -773,13 +773,12 @@ class ScreenMemoryDB(QdrantBase):
|
||||
if not vector:
|
||||
return
|
||||
|
||||
# ── P0-3: Collection Management ──
|
||||
# Prevent embedding saturation by limiting to 200 high-quality signatures
|
||||
# ── LRU Eviction: Replace nuclear wipe with intelligent pruning ──
|
||||
# Keeps high-confidence entries, evicts oldest low-confidence ones.
|
||||
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()
|
||||
self._evict_lru(count - 150) # Evict down to 150 entries
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -789,9 +788,10 @@ class ScreenMemoryDB(QdrantBase):
|
||||
payload={
|
||||
"signature": xml_signature[:500],
|
||||
"screen_type": screen_type,
|
||||
"confidence": confidence,
|
||||
"stored_at": time.time(),
|
||||
},
|
||||
log_success=f"🧠 [ScreenMemory] Learned new layout mapping: {screen_type}",
|
||||
log_success=f"🧠 [ScreenMemory] Learned new layout mapping: {screen_type} (confidence: {confidence:.2f})",
|
||||
)
|
||||
|
||||
def get_screen_type(self, xml_signature: str, similarity_threshold: float = 0.95) -> Optional[str]:
|
||||
@@ -820,6 +820,51 @@ class ScreenMemoryDB(QdrantBase):
|
||||
logger.debug(f"Screen memory error: {e}")
|
||||
return None
|
||||
|
||||
def _evict_lru(self, evict_count: int):
|
||||
"""Evict the oldest, lowest-confidence entries instead of nuclear wipe.
|
||||
|
||||
Strategy: Sort by confidence ASC, then by stored_at ASC (oldest first).
|
||||
High-confidence entries (>=0.8) are NEVER evicted regardless of age.
|
||||
"""
|
||||
if not self.is_connected or evict_count <= 0:
|
||||
return
|
||||
|
||||
try:
|
||||
points, _ = self.client.scroll(
|
||||
collection_name=self.collection_name,
|
||||
limit=300,
|
||||
with_payload=True,
|
||||
)
|
||||
|
||||
# Score each point: lower = more evictable
|
||||
candidates = []
|
||||
for pt in points:
|
||||
payload = pt.payload or {}
|
||||
confidence = payload.get("confidence", 0.5)
|
||||
stored_at = payload.get("stored_at", 0)
|
||||
|
||||
# High-confidence entries are immune to eviction
|
||||
if confidence >= 0.8:
|
||||
continue
|
||||
|
||||
candidates.append((pt, confidence, stored_at))
|
||||
|
||||
# Sort: lowest confidence first, then oldest first
|
||||
candidates.sort(key=lambda x: (x[1], x[2]))
|
||||
|
||||
evicted = 0
|
||||
for pt, conf, _ in candidates[:evict_count]:
|
||||
sig = pt.payload.get("signature", str(pt.id))
|
||||
self.delete_point(sig)
|
||||
evicted += 1
|
||||
|
||||
if evicted:
|
||||
logger.info(
|
||||
f"🧹 [ScreenMemory] LRU evicted {evicted} low-confidence entries (preserved high-confidence)."
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(f"LRU eviction error: {e}")
|
||||
|
||||
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:
|
||||
@@ -1350,3 +1395,44 @@ def wipe_all_ai_caches():
|
||||
logger.warning(f"⚠️ Failed to wipe {db_cls.__name__}: {e}")
|
||||
|
||||
logger.info(f"🗑️ [Blank Start] Wiped {wiped_count}/{len(global_dbs)} global AI caches.")
|
||||
|
||||
|
||||
def memory_hygiene(min_confidence: float = 0.3):
|
||||
"""
|
||||
Selective Amnesia — prunes low-confidence entries while preserving
|
||||
high-confidence learned patterns. This replaces the nuclear blank_start
|
||||
approach with intelligent memory management.
|
||||
|
||||
Entries with confidence >= min_confidence survive.
|
||||
Entries below the threshold are pruned as potentially poisoned.
|
||||
"""
|
||||
pruned_total = 0
|
||||
|
||||
# Only prune ScreenMemoryDB — it's the most susceptible to VLM hallucination poisoning
|
||||
try:
|
||||
db = ScreenMemoryDB()
|
||||
if not db.is_connected:
|
||||
logger.debug("[Memory Hygiene] Qdrant not available, skipping.")
|
||||
return
|
||||
|
||||
points, _ = db.client.scroll(
|
||||
collection_name=db.collection_name,
|
||||
limit=500,
|
||||
with_payload=True,
|
||||
)
|
||||
|
||||
for pt in points:
|
||||
payload = pt.payload or {}
|
||||
confidence = payload.get("confidence", 0.5) # Default 0.5 for legacy entries
|
||||
if confidence < min_confidence:
|
||||
sig = payload.get("signature", str(pt.id))
|
||||
db.delete_point(sig)
|
||||
pruned_total += 1
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"[Memory Hygiene] Screen memory prune error: {e}")
|
||||
|
||||
if pruned_total:
|
||||
logger.info(f"🧹 [Memory Hygiene] Pruned {pruned_total} low-confidence entries (threshold: {min_confidence}).")
|
||||
else:
|
||||
logger.info("🧹 [Memory Hygiene] All memories healthy. Nothing to prune.")
|
||||
|
||||
@@ -66,18 +66,23 @@ class ScreenTopology:
|
||||
"tap explore tab": ScreenType.EXPLORE_GRID,
|
||||
"tap reels tab": ScreenType.REELS_FEED,
|
||||
"tap profile tab": ScreenType.OWN_PROFILE,
|
||||
"press back": ScreenType.HOME_FEED,
|
||||
# NOTE: 'press back' intentionally omitted — destination is non-deterministic
|
||||
# (could be HOME_FEED, EXPLORE_GRID, POST_DETAIL, etc. depending on navigation history)
|
||||
},
|
||||
ScreenType.POST_DETAIL: {
|
||||
"tap view all comments": ScreenType.COMMENTS,
|
||||
"press back": ScreenType.EXPLORE_GRID,
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
"tap explore tab": ScreenType.EXPLORE_GRID,
|
||||
"tap profile tab": ScreenType.OWN_PROFILE,
|
||||
# NOTE: 'press back' intentionally omitted — destination is non-deterministic
|
||||
# (could be HOME_FEED, EXPLORE_GRID, OTHER_PROFILE, etc.)
|
||||
},
|
||||
ScreenType.COMMENTS: {
|
||||
"press back": ScreenType.POST_DETAIL,
|
||||
},
|
||||
ScreenType.SEARCH_RESULTS: {
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
"press back": ScreenType.EXPLORE_GRID,
|
||||
# NOTE: 'press back' intentionally omitted — destination is non-deterministic
|
||||
},
|
||||
ScreenType.UNKNOWN: {
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
|
||||
Reference in New Issue
Block a user