From 52c553827f4e3c95a5480be65e434a6c1e01a3f5 Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Tue, 28 Apr 2026 10:34:44 +0200 Subject: [PATCH] fix(core): add structural sanity guards to prevent post-related VLM hallucinations on search and profile screens --- GramAddict/core/perception/screen_identity.py | 20 ++++----- GramAddict/core/telepathic_engine.py | 15 +++++++ tests/core/test_unfollow_engine.py | 5 +-- tests/e2e/test_brain_live.py | 33 +++++++++------ tests/e2e/test_goap_loop_prevention.py | 3 +- tests/test_planner_hierarchy.py | 41 +++++++++---------- tests/unit/behaviors/test_comment_plugin.py | 26 ++++++------ tests/unit/test_structural_guard.py | 38 +++++++++++++++++ 8 files changed, 118 insertions(+), 63 deletions(-) diff --git a/GramAddict/core/perception/screen_identity.py b/GramAddict/core/perception/screen_identity.py index 2812788..c74d675 100644 --- a/GramAddict/core/perception/screen_identity.py +++ b/GramAddict/core/perception/screen_identity.py @@ -164,16 +164,7 @@ class ScreenIdentity: logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL") return ScreenType.MODAL - # Priority 1: Check Qdrant Semantic Cache - if signature and self.screen_memory and self.screen_memory.is_connected: - cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92) - if cached_type_str: - try: - return ScreenType[cached_type_str] - except KeyError: - pass - - # Priority 2: Structural Heuristics (Instant, for core tabs) + # Priority 1: Structural Heuristics (100% Deterministic) if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids: return ScreenType.FOLLOW_LIST @@ -192,6 +183,15 @@ class ScreenIdentity: if "direct_thread_header" in ids or "row_thread_composer_edittext" in ids: return ScreenType.DM_THREAD + # Priority 2: Check Qdrant Semantic Cache (Fuzzy/VLM derived) + if signature and self.screen_memory and self.screen_memory.is_connected: + cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92) + if cached_type_str: + try: + return ScreenType[cached_type_str] + except KeyError: + pass + if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab: return ScreenType.POST_DETAIL diff --git a/GramAddict/core/telepathic_engine.py b/GramAddict/core/telepathic_engine.py index b476c8a..95eb96a 100644 --- a/GramAddict/core/telepathic_engine.py +++ b/GramAddict/core/telepathic_engine.py @@ -344,8 +344,23 @@ class TelepathicEngine: if "story" in semantic and y < screen_height * 0.2: # E.g. "Your Story" circle at the top return False + # Prevent tapping a search list item when looking for a post username + if "row search user container" in semantic.replace("_", " "): + return False return True + # 3.5 Media Content Guard + if "post media content" in intent: + # Prevent tapping a search keyword instead of a media post + if "row search keyword title" in semantic.replace("_", " "): + return False + + # 3.6 Post Author Username Header Guard + if "post author username header" in intent: + # Prevent tapping the follow button when looking for the username + if "follow button" in semantic.replace("_", " "): + return False + # 4. Profile Picture/Story Ring Guard if "story ring" in intent or "avatar" in intent: current_user = self._get_current_username() diff --git a/tests/core/test_unfollow_engine.py b/tests/core/test_unfollow_engine.py index 114b851..d75e05b 100644 --- a/tests/core/test_unfollow_engine.py +++ b/tests/core/test_unfollow_engine.py @@ -7,11 +7,10 @@ false-positive mocks. """ import os -from unittest.mock import MagicMock, call +from unittest.mock import MagicMock from GramAddict.core.unfollow_engine import _run_zero_latency_unfollow_loop - FIX_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures") @@ -71,7 +70,7 @@ def test_unfollow_engine_extracts_users_and_calls_back_on_high_resonance(): # x = (247+537)//2 = 392. y = (1014+1061)//2 = 1037. # It calls _humanized_click(device, x, y) which ultimately does device.click(x, y). # BUT _humanized_click uses gaussian distribution so exact coordinates are fuzzy. - + # The critical assertion: we MUST have pressed back to return to the list. assert device.back.call_count >= 1, "Engine failed to press back after inspecting profile!" diff --git a/tests/e2e/test_brain_live.py b/tests/e2e/test_brain_live.py index e2c8c38..93da266 100644 --- a/tests/e2e/test_brain_live.py +++ b/tests/e2e/test_brain_live.py @@ -1,34 +1,41 @@ -import pytest -from GramAddict.core.navigation.brain import ask_brain_for_action -from GramAddict.core.perception.screen_identity import ScreenType import logging +import pytest + +from GramAddict.core.navigation.brain import ask_brain_for_action + logger = logging.getLogger(__name__) + @pytest.mark.live_llm def test_brain_recommends_scroll_when_trapped(): """ - Test that the real, live LLM Brain correctly deduces that it should + Test that the real, live LLM Brain correctly deduces that it should scroll down when the target element is missing and it's trapped. """ goal = "open following list" screen = "OWN_PROFILE" - available_actions = ["tap profile tab", "tap share button", "press back", "tap reels tab", "tap messages tab", "scroll down", "scroll up"] + available_actions = [ + "tap profile tab", + "tap share button", + "press back", + "tap reels tab", + "tap messages tab", + "scroll down", + "scroll up", + ] explored_nav_actions = {"tap following list"} - + # We query the actual LLM as configured in the environment (e.g. qwen3.5:latest) # This prevents regressions where the LLM is misconfigured or returns empty strings. brain_action = ask_brain_for_action( - goal=goal, - screen_type=screen, - available_actions=available_actions, - explored_actions=explored_nav_actions + goal=goal, screen_type=screen, available_actions=available_actions, explored_actions=explored_nav_actions ) - + logger.info(f"Brain action returned: '{brain_action}'") - + if brain_action is None or brain_action == "": pytest.skip("Brain LLM returned None or empty string. Ollama timeout or hallucination.") - + if brain_action != "scroll down": pytest.skip(f"VLM chose '{brain_action}' instead of 'scroll down'. Small local models can be flaky.") diff --git a/tests/e2e/test_goap_loop_prevention.py b/tests/e2e/test_goap_loop_prevention.py index 41311c8..e4efb86 100644 --- a/tests/e2e/test_goap_loop_prevention.py +++ b/tests/e2e/test_goap_loop_prevention.py @@ -12,9 +12,10 @@ These tests prove: Requires: Real XML fixture at tests/fixtures/user_profile_dump.xml """ -import pytest from unittest.mock import patch +import pytest + from GramAddict.core.navigation.planner import GoalPlanner from GramAddict.core.perception.screen_identity import ScreenType from GramAddict.core.screen_topology import ScreenTopology diff --git a/tests/test_planner_hierarchy.py b/tests/test_planner_hierarchy.py index f533e5d..436dd21 100644 --- a/tests/test_planner_hierarchy.py +++ b/tests/test_planner_hierarchy.py @@ -1,12 +1,16 @@ -import pytest from unittest.mock import patch + +import pytest + from GramAddict.core.navigation.planner import GoalPlanner from GramAddict.core.perception.screen_identity import ScreenType + @pytest.fixture def planner(): return GoalPlanner("test_user") + @patch("GramAddict.core.navigation.brain.query_llm") @patch("GramAddict.core.screen_topology.ScreenTopology.find_route") def test_brain_is_primary_strategy(mock_find_route, mock_query, planner): @@ -16,23 +20,20 @@ def test_brain_is_primary_strategy(mock_find_route, mock_query, planner): """ # 1. Setup State goal = "open some screen" - screen = { - "screen_type": ScreenType.HOME_FEED, - "available_actions": ["action A", "action B"], - "context": {} - } - + screen = {"screen_type": ScreenType.HOME_FEED, "available_actions": ["action A", "action B"], "context": {}} + # 2. Setup Mocks - mock_query.return_value = "action A" # Brain picks A - mock_find_route.return_value = [("action B", ScreenType.EXPLORE_GRID)] # HD Map would pick B - + mock_query.return_value = "action A" # Brain picks A + mock_find_route.return_value = [("action B", ScreenType.EXPLORE_GRID)] # HD Map would pick B + # 3. Execute Planner action = planner.plan_next_step(goal, screen) - + # 4. Assertions assert action == "action A", "Planner did not use the Brain's action!" mock_query.assert_called_once() - mock_find_route.assert_not_called() # Crucial: HD Map must be skipped entirely! + mock_find_route.assert_not_called() # Crucial: HD Map must be skipped entirely! + @patch("GramAddict.core.navigation.brain.query_llm") @patch("GramAddict.core.screen_topology.ScreenTopology.find_route") @@ -43,20 +44,16 @@ def test_brain_fallback_to_hd_map(mock_goal_target, mock_find_route, mock_query, """ # 1. Setup State goal = "open explore screen" - screen = { - "screen_type": ScreenType.HOME_FEED, - "available_actions": ["action A", "action B"], - "context": {} - } - + screen = {"screen_type": ScreenType.HOME_FEED, "available_actions": ["action A", "action B"], "context": {}} + # 2. Setup Mocks - mock_query.return_value = None # Brain fails or is confused + mock_query.return_value = None # Brain fails or is confused mock_goal_target.return_value = ScreenType.EXPLORE_GRID - mock_find_route.return_value = [("action B", ScreenType.EXPLORE_GRID)] # HD Map picks B - + mock_find_route.return_value = [("action B", ScreenType.EXPLORE_GRID)] # HD Map picks B + # 3. Execute Planner action = planner.plan_next_step(goal, screen) - + # 4. Assertions assert action == "action B", "Planner did not fallback to HD Map when Brain failed!" mock_query.assert_called_once() diff --git a/tests/unit/behaviors/test_comment_plugin.py b/tests/unit/behaviors/test_comment_plugin.py index 6a267d0..385f6b6 100644 --- a/tests/unit/behaviors/test_comment_plugin.py +++ b/tests/unit/behaviors/test_comment_plugin.py @@ -9,10 +9,8 @@ Tests CommentPlugin against real XML fixtures to ensure: import os from unittest.mock import MagicMock -from GramAddict.core.behaviors import BehaviorContext from GramAddict.core.behaviors.comment import CommentPlugin - FIX_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "fixtures") @@ -46,10 +44,10 @@ def test_comment_plugin_can_activate_rejects_grids(): ctx.configs.args = MagicMock(comment_percentage=100) ctx.configs.get_plugin_config.return_value = {} ctx.shared_state = {} - + ctx.context_xml = _get_fixture("explore_feed_dump.xml") assert plugin.can_activate(ctx) is False, "CommentPlugin falsely activated on Explore Grid!" - + ctx.context_xml = _get_fixture("user_profile_dump.xml") assert plugin.can_activate(ctx) is False, "CommentPlugin falsely activated on Profile Grid!" @@ -60,42 +58,42 @@ def test_comment_plugin_fails_safely_without_writer(): must abort safely and press BACK to exit the comment sheet. """ plugin = CommentPlugin() - + ctx = MagicMock() ctx.configs.get_plugin_config.return_value = {} ctx.cognitive_stack = {} # No writer! - + nav_graph = MagicMock() nav_graph.do.return_value = True # Successfully opened comment sheet ctx.cognitive_stack["nav_graph"] = nav_graph - + result = plugin.execute(ctx) - + assert result.executed is False, "CommentPlugin must not execute without a writer!" ctx.device.press.assert_called_once_with("back") def test_comment_plugin_dry_run_exits_safely(): """ - Test: If dry_run is true, the plugin generates the text but presses BACK + Test: If dry_run is true, the plugin generates the text but presses BACK to cancel posting. """ plugin = CommentPlugin() - + writer = MagicMock() writer.generate_comment.return_value = "Awesome!" - + ctx = MagicMock() ctx.configs.get_plugin_config.return_value = {} ctx.cognitive_stack = {"writer": writer} ctx.configs.args = MagicMock(dry_run_comments=True) - + nav_graph = MagicMock() nav_graph.do.return_value = True # Successfully opened comment sheet ctx.cognitive_stack["nav_graph"] = nav_graph - + result = plugin.execute(ctx) - + assert result.executed is True, "Dry run is considered a successful execution." assert result.interactions == 0, "Dry run must yield 0 interactions." assert result.metadata["text"] == "Awesome!" diff --git a/tests/unit/test_structural_guard.py b/tests/unit/test_structural_guard.py index d9a0c3f..e408469 100644 --- a/tests/unit/test_structural_guard.py +++ b/tests/unit/test_structural_guard.py @@ -110,3 +110,41 @@ def test_structural_reels_first_grid_item_y_coords(): assert ( is_valid_nav is False ), "Structural Guard failed to reject a hallucinated navigation tab in the middle of the screen." + +def test_structural_guard_rejects_search_keyword_for_media_content(): + engine = TelepathicEngine() + + node = { + "semantic_string": "text: 'i\\'m', id context: 'row search keyword title'", + "class_name": "android.widget.TextView", + "y": 500 + } + + is_valid = engine._structural_sanity_check(node, "post media content", 2400) + assert is_valid is False, "Structural Guard failed to reject 'row_search_keyword_title' for 'post media content'." + + +def test_structural_guard_rejects_search_user_for_post_username(): + engine = TelepathicEngine() + + node = { + "semantic_string": "desc: 'Followed by pratiek_the_entrepreneur + 19 more', id context: 'row search user container'", + "class_name": "android.widget.LinearLayout", + "y": 800 + } + + is_valid = engine._structural_sanity_check(node, "tap post username", 2400) + assert is_valid is False, "Structural Guard failed to reject 'row_search_user_container' for 'tap post username'." + + +def test_structural_guard_rejects_follow_button_for_author_username_header(): + engine = TelepathicEngine() + + node = { + "semantic_string": "text: 'Following', desc: 'Following Mariischen', id context: 'profile header follow button'", + "class_name": "android.widget.Button", + "y": 600 + } + + is_valid = engine._structural_sanity_check(node, "post author username header", 2400) + assert is_valid is False, "Structural Guard failed to reject follow button for 'post author username header'."