diff --git a/.gitignore b/.gitignore index f7a5905..f7edcea 100644 --- a/.gitignore +++ b/.gitignore @@ -47,6 +47,7 @@ traceback.log htmlcov/ .coverage coverage.xml +coverage_e2e.json .hypothesis/ # Local diagnostic traces diff --git a/GramAddict/core/perception/context_gate.py b/GramAddict/core/perception/context_gate.py index 3b052b9..0c92cfc 100644 --- a/GramAddict/core/perception/context_gate.py +++ b/GramAddict/core/perception/context_gate.py @@ -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: diff --git a/GramAddict/core/qdrant_memory.py b/GramAddict/core/qdrant_memory.py index 5f88cd7..abf4e63 100644 --- a/GramAddict/core/qdrant_memory.py +++ b/GramAddict/core/qdrant_memory.py @@ -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.") diff --git a/GramAddict/core/screen_topology.py b/GramAddict/core/screen_topology.py index 3c4ad0e..dd483ee 100644 --- a/GramAddict/core/screen_topology.py +++ b/GramAddict/core/screen_topology.py @@ -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, diff --git a/tests/tdd/test_context_gate_matrix.py b/tests/tdd/test_context_gate_matrix.py new file mode 100644 index 0000000..645d420 --- /dev/null +++ b/tests/tdd/test_context_gate_matrix.py @@ -0,0 +1,147 @@ +""" +TDD: ContextGate Action Compatibility Matrix + +Tests that the ContextGate blocks ALL structurally impossible actions, +not just the ones that happen to have matching resource-id markers. + +Red-Green-Blue: These tests MUST fail first, then we implement. +""" + +import pytest + +from GramAddict.core.perception.context_gate import ContextGate +from GramAddict.core.perception.screen_identity import ScreenType + + +@pytest.fixture +def gate(): + return ContextGate() + + +def _make_screen(screen_type: ScreenType, resource_ids=None): + """Helper to create a minimal screen_state dict.""" + return { + "screen_type": screen_type, + "resource_ids": resource_ids or set(), + "available_actions": [], + "context": {}, + } + + +class TestContextGateCategoricalBans: + """P2-2: Every screen/action combination that is structurally impossible + must be categorically banned โ€” not reliant on marker presence.""" + + def test_like_blocked_on_story_view(self, gate): + """Cannot like on STORY_VIEW โ€” there's no feed-style like button.""" + screen = _make_screen(ScreenType.STORY_VIEW) + assert gate.is_allowed("like", screen) is False + + def test_comment_blocked_on_story_view(self, gate): + """Cannot comment on STORY_VIEW โ€” reply is a different action.""" + screen = _make_screen(ScreenType.STORY_VIEW) + assert gate.is_allowed("comment", screen) is False + + def test_follow_blocked_on_story_view(self, gate): + """Cannot follow from STORY_VIEW โ€” no follow button exists.""" + screen = _make_screen(ScreenType.STORY_VIEW) + assert gate.is_allowed("follow", screen) is False + + def test_save_blocked_on_story_view(self, gate): + """Cannot save on STORY_VIEW.""" + screen = _make_screen(ScreenType.STORY_VIEW) + assert gate.is_allowed("save", screen) is False + + def test_like_blocked_on_follow_list(self, gate): + """Cannot like on FOLLOW_LIST โ€” it's a list of users, not posts.""" + screen = _make_screen(ScreenType.FOLLOW_LIST) + assert gate.is_allowed("like", screen) is False + + def test_comment_blocked_on_follow_list(self, gate): + """Cannot comment on FOLLOW_LIST.""" + screen = _make_screen(ScreenType.FOLLOW_LIST) + assert gate.is_allowed("comment", screen) is False + + def test_comment_blocked_on_explore_grid(self, gate): + """Cannot comment on EXPLORE_GRID โ€” it's a grid of thumbnails.""" + screen = _make_screen(ScreenType.EXPLORE_GRID) + assert gate.is_allowed("comment", screen) is False + + def test_save_blocked_on_dm_inbox(self, gate): + """Cannot save on DM_INBOX.""" + screen = _make_screen(ScreenType.DM_INBOX) + assert gate.is_allowed("save", screen) is False + + def test_save_blocked_on_dm_thread(self, gate): + """Cannot save on DM_THREAD.""" + screen = _make_screen(ScreenType.DM_THREAD) + assert gate.is_allowed("save", screen) is False + + def test_follow_blocked_on_home_feed(self, gate): + """Cannot follow from HOME_FEED โ€” there's no follow button on feed posts.""" + screen = _make_screen(ScreenType.HOME_FEED) + assert gate.is_allowed("follow", screen) is False + + def test_follow_blocked_on_explore_grid(self, gate): + """Cannot follow from EXPLORE_GRID.""" + screen = _make_screen(ScreenType.EXPLORE_GRID) + assert gate.is_allowed("follow", screen) is False + + def test_repost_blocked_on_follow_list(self, gate): + """Cannot repost from FOLLOW_LIST.""" + screen = _make_screen(ScreenType.FOLLOW_LIST) + assert gate.is_allowed("repost", screen) is False + + +class TestContextGatePositiveCases: + """Ensure that valid actions are NOT blocked.""" + + def test_like_allowed_on_post_detail(self, gate): + """Like IS valid on POST_DETAIL with proper markers.""" + screen = _make_screen(ScreenType.POST_DETAIL, {"row_feed_button_like", "row_feed_button_comment"}) + assert gate.is_allowed("like", screen) is True + + def test_comment_allowed_on_post_detail(self, gate): + """Comment IS valid on POST_DETAIL with proper markers.""" + screen = _make_screen(ScreenType.POST_DETAIL, {"row_feed_button_comment", "row_feed_button_like"}) + assert gate.is_allowed("comment", screen) is True + + def test_follow_allowed_on_other_profile(self, gate): + """Follow IS valid on OTHER_PROFILE with proper markers.""" + screen = _make_screen(ScreenType.OTHER_PROFILE, {"profile_header_follow_button"}) + assert gate.is_allowed("follow", screen) is True + + def test_like_allowed_on_home_feed(self, gate): + """Like IS valid on HOME_FEED with proper markers.""" + screen = _make_screen(ScreenType.HOME_FEED, {"row_feed_button_like"}) + assert gate.is_allowed("like", screen) is True + + +class TestScreenTopologyBackTransitions: + """P2-3: HD Map must NOT claim to know where 'press back' goes. + Back is inherently non-deterministic.""" + + def test_press_back_has_no_expected_screen(self): + """press back must NEVER return a specific expected screen.""" + from GramAddict.core.screen_topology import ScreenTopology + + # For every screen that has a 'press back' transition, + # we verify it should NOT be in the topology + for screen_type, transitions in ScreenTopology.TRANSITIONS.items(): + if "press back" in transitions: + # This test SHOULD fail until we remove press back from TRANSITIONS + # for screens where it's non-deterministic + if screen_type in ( + ScreenType.DM_INBOX, + ScreenType.FOLLOW_LIST, + ScreenType.STORY_VIEW, + ScreenType.COMMENTS, + ): + # These are "leaf" screens with a deterministic parent โ€” OK to keep + continue + # NON-DETERMINISTIC back: OTHER_PROFILE, POST_DETAIL, SEARCH_RESULTS + assert False, ( + f"ScreenTopology claims 'press back' from {screen_type.name} goes to " + f"{transitions['press back'].name}, but this is non-deterministic! " + f"Remove this transition โ€” it causes GOAP to reject valid navigation states." + ) diff --git a/tests/tdd/test_memory_hygiene.py b/tests/tdd/test_memory_hygiene.py new file mode 100644 index 0000000..486a211 --- /dev/null +++ b/tests/tdd/test_memory_hygiene.py @@ -0,0 +1,98 @@ +""" +TDD: Memory Hygiene โ€” Selective Amnesia (NOT Nuclear Wipe) + +Tests that startup performs intelligent memory pruning instead of +wiping all learned knowledge on every run. + +Red-Green-Blue: These tests MUST fail first, then we implement. +""" + +import pytest + + +class TestMemoryHygiene: + """P0-2: Replace blank_start nuclear wipe with selective memory hygiene.""" + + def test_high_confidence_entries_survive_hygiene(self): + """High-confidence learned patterns must survive startup hygiene.""" + from GramAddict.core.qdrant_memory import ScreenMemoryDB + + db = ScreenMemoryDB() + if not db.is_connected: + pytest.skip("Qdrant not available") + + # Store a high-confidence entry + db.store_screen("sig_high_conf", "HOME_FEED", confidence=0.95) + + # Run hygiene โ€” this should NOT delete the high-confidence entry + from GramAddict.core.qdrant_memory import memory_hygiene + + memory_hygiene() + + # Verify it survived + result = db.get_screen_type("sig_high_conf", similarity_threshold=0.90) + assert result == "HOME_FEED", "High-confidence entry was destroyed by memory_hygiene!" + + def test_low_confidence_entries_are_pruned(self): + """Low-confidence (stale/poisoned) entries must be pruned during hygiene.""" + from GramAddict.core.qdrant_memory import ScreenMemoryDB + + db = ScreenMemoryDB() + if not db.is_connected: + pytest.skip("Qdrant not available") + + # Store a low-confidence entry (representing a bad VLM guess) + db.store_screen("sig_low_conf", "MODAL", confidence=0.2) + + # Run hygiene โ€” this SHOULD delete the low-confidence entry + from GramAddict.core.qdrant_memory import memory_hygiene + + memory_hygiene() + + # Verify it was pruned + result = db.get_screen_type("sig_low_conf", similarity_threshold=0.90) + assert result is None, "Low-confidence poisoned entry survived hygiene!" + + def test_memory_hygiene_function_exists(self): + """The memory_hygiene function must be importable.""" + from GramAddict.core.qdrant_memory import memory_hygiene + + assert callable(memory_hygiene) + + def test_blank_start_is_not_default_config(self): + """blank_start must NOT be true in the default config. + It should only be available as a --blank-start CLI flag for explicit manual resets.""" + import yaml + + with open("config.yml", "r") as f: + config = yaml.safe_load(f) + + blank_start = config.get("blank_start", False) + assert blank_start is False or blank_start is None, ( + f"blank_start is {blank_start} in config.yml! " + "This wipes ALL learned knowledge on every run, negating the entire learning system. " + "Remove it or set to false. Use --blank-start CLI flag for explicit resets only." + ) + + +class TestScreenMemoryLRU: + """P2-4: Replace nuclear 200-entry wipe with LRU eviction.""" + + def test_screen_memory_does_not_nuke_at_200(self): + """ScreenMemoryDB must NOT wipe the entire collection at 200 entries.""" + from GramAddict.core.qdrant_memory import ScreenMemoryDB + + db = ScreenMemoryDB() + if not db.is_connected: + pytest.skip("Qdrant not available") + + # Store 201 entries + for i in range(201): + db.store_screen(f"sig_lru_{i}", "HOME_FEED", confidence=0.9) + + # Verify that the first high-confidence entry still exists + # (it should NOT have been nuked) + result = db.get_screen_type("sig_lru_0", similarity_threshold=0.90) + assert result is not None, ( + "ScreenMemoryDB nuked all 200+ entries! " "Must use LRU eviction instead of nuclear wipe." + )