From 79e5784b930836539d6ad944865cb118d88b2c4b Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Mon, 4 May 2026 14:51:54 +0200 Subject: [PATCH] feat(core): complete P1 hardening phase - ContextGate, Localization Purge, SoM Safety Clamp --- GramAddict/core/bot_flow.py | 13 ++-- GramAddict/core/goap.py | 16 +++- GramAddict/core/perception/context_gate.py | 78 +++++++++++++++++++ GramAddict/core/perception/intent_resolver.py | 64 ++++++++++++--- GramAddict/core/perception/screen_identity.py | 26 +++---- GramAddict/core/q_nav_graph.py | 3 +- GramAddict/core/situational_awareness.py | 24 +++--- tests/core/test_context_gate.py | 65 ++++++++++++++++ tests/core/test_intent_resolver_hardening.py | 45 +++++++++++ 9 files changed, 281 insertions(+), 53 deletions(-) create mode 100644 GramAddict/core/perception/context_gate.py create mode 100644 tests/core/test_context_gate.py create mode 100644 tests/core/test_intent_resolver_hardening.py diff --git a/GramAddict/core/bot_flow.py b/GramAddict/core/bot_flow.py index 011820c..27ba52d 100644 --- a/GramAddict/core/bot_flow.py +++ b/GramAddict/core/bot_flow.py @@ -698,16 +698,15 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, if not isinstance(xml_check, str): return - xml_check_lower = xml_check.lower() - - # ── 1. Profile Guards (Private / Empty) ── - if "this account is private" in xml_check_lower or "konto ist privat" in xml_check_lower: + # ── 1. Profile Guards (Private / Empty / Close Friends) ── + # Using structural resource-ID fragments to avoid localization debt + if "private_profile_empty_state" in xml_check or "private_profile_container" in xml_check: logger.info( f"🔒 [Profile Guard] @{username} is private. Aborting deep interaction.", extra={"color": f"{Fore.YELLOW}"} ) return - if "no posts yet" in xml_check_lower or "noch keine beiträge" in xml_check_lower: + if "empty_state_view" in xml_check or "no_posts_yet" in xml_check: logger.info( f"📭 [Profile Guard] @{username} has no posts. Aborting deep interaction.", extra={"color": f"{Fore.YELLOW}"}, @@ -715,9 +714,9 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, return if getattr(configs.args, "ignore_close_friends", False): - if "enge freunde" in xml_check_lower or "close friend" in xml_check_lower: + if "close_friends_badge" in xml_check or "profile_header_close_friend" in xml_check: logger.info( - f"💚 [Profile Guard] @{username} is a Close Friend. Ignoring completely.", extra={"color": "\\033[32m"} + f"💚 [Profile Guard] @{username} is a Close Friend. Ignoring completely.", extra={"color": "\033[32m"} ) return diff --git a/GramAddict/core/goap.py b/GramAddict/core/goap.py index f22f2db..a6bdffe 100644 --- a/GramAddict/core/goap.py +++ b/GramAddict/core/goap.py @@ -20,6 +20,7 @@ from typing import Any, Dict, List from GramAddict.core.navigation.knowledge import NavigationKnowledge from GramAddict.core.navigation.path_memory import PathMemory from GramAddict.core.navigation.planner import GoalPlanner +from GramAddict.core.perception.context_gate import ContextGate from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType from GramAddict.core.utils import random_sleep @@ -66,6 +67,7 @@ class GoalExecutor: self.screen_id.device = device self.planner = GoalPlanner(bot_username) self.path_memory = PathMemory(bot_username) + self.context_gate = ContextGate() self.max_steps = 15 # Safety: never execute more than 15 steps self._sae = None # Lazy-loaded, injectable for tests self.action_failures = {} # Tracking for failed actions in current goal session @@ -221,7 +223,7 @@ class GoalExecutor: last_screen_type = screen_type # EXECUTE - success = self._execute_action(action, goal=goal) + success = self._execute_action(action, goal=goal, screen_state=screen) if success: steps_taken.append({"action": action}) @@ -329,7 +331,7 @@ class GoalExecutor: return False - def _execute_action(self, action: str, goal: str = None) -> bool: + def _execute_action(self, action: str, goal: str = None, screen_state: dict = None) -> bool: """Execute a single natural-language action using the TelepathicEngine.""" if action == "press back": @@ -349,6 +351,12 @@ class GoalExecutor: random_sleep(2.0, 3.5) return True + # ── P1-5: Context Gate ── + # Check if the intent is structurally plausible on THIS screen before calling VLM + if screen_state and not self.context_gate.is_allowed(action, screen_state): + logger.warning(f"🛡️ [GOAP Execute] Action '{action}' blocked by ContextGate for this screen.") + return False + # Use TelepathicEngine for any semantic click from GramAddict.core.telepathic_engine import TelepathicEngine @@ -533,7 +541,9 @@ class GoalExecutor: action = step.get("action", "") logger.info(f"🧠 [GOAP Recall Step {i + 1}/{len(steps)}] '{action}'") - success = self._execute_action(action, goal=goal) + # Re-perceive to ensure ContextGate has fresh data + current_screen = self.perceive() + success = self._execute_action(action, goal=goal, screen_state=current_screen) if not success: logger.warning(f"⚠️ [GOAP Recall] Step '{action}' failed. Path may be stale.") return False diff --git a/GramAddict/core/perception/context_gate.py b/GramAddict/core/perception/context_gate.py new file mode 100644 index 0000000..3b052b9 --- /dev/null +++ b/GramAddict/core/perception/context_gate.py @@ -0,0 +1,78 @@ +import logging +from typing import Any, Dict + +from GramAddict.core.perception.screen_identity import ScreenType + +logger = logging.getLogger(__name__) + + +class ContextGate: + """ + Validates if an action (intent) is structurally possible on the current screen. + This acts as a high-speed circuit breaker before invoking expensive VLM logic. + + 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. + """ + + # Intent -> List of resource-id fragments that MUST be present on the screen + # to even consider performing this action. + REQUIRED_MARKERS = { + "comment": ["comment", "row_feed_button_comment", "shell_comment_button"], + "like": ["like", "heart", "row_feed_button_like", "shell_like_button"], + "follow": ["follow", "profile_header_follow_button", "button_follow"], + "unfollow": ["follow", "profile_header_follow_button", "button_follow"], + "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. + + Args: + intent: The action name (e.g. 'follow', 'comment') + screen_state: The result of ScreenIdentity.identify() + + Returns: + bool: True if the action is plausible, False if it should be blocked. + """ + intent_lower = intent.lower() + 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 + + # 2. Check structural requirements + # Only check for specific interaction intents + 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: + if marker in rid.lower(): + has_marker = True + break + if has_marker: + break + + if not has_marker: + logger.warning( + f"🛡️ [ContextGate] Blocked '{intent}' on {screen_type.name}: " + f"No structural markers found {markers}." + ) + return False + + return True diff --git a/GramAddict/core/perception/intent_resolver.py b/GramAddict/core/perception/intent_resolver.py index 89e7058..9514559 100644 --- a/GramAddict/core/perception/intent_resolver.py +++ b/GramAddict/core/perception/intent_resolver.py @@ -726,30 +726,70 @@ class IntentResolver: ) print(f"DEBUG_INTENT: VLM RAW RESPONSE for '{intent_description}': {res}") data = json.loads(res) - box_idx = data.get("box") - if box_idx is None: - box_idx = data.get("selected_index") - if box_idx is None: - box_idx = data.get("box_index") - if box_idx is None: - box_idx = data.get("index") + box_idx = self._parse_box_index(data) + selected = self._validate_and_get_node(box_idx, box_map) - if box_idx is not None and box_idx in box_map: - selected = box_map[box_idx] + if selected: logger.info( f"👁️ [Visual Discovery] VLM selected box [{box_idx}] → " f"id='{selected.resource_id}', desc='{selected.content_desc}'" ) return selected else: - logger.warning( - f"👁️ [Visual Discovery] VLM returned box={box_idx} which is not in box_map ({list(box_map.keys())[:5]}...)" - ) + logger.warning(f"👁️ [Visual Discovery] VLM returned invalid box={box_idx} (OOB or unparsable).") except Exception as e: logger.warning(f"⚠️ [Visual Discovery] VLM call failed: {e}") return None + def _parse_box_index(self, data: Dict) -> Optional[int]: + """Parses the box index from VLM JSON response, handling common hallucination formats.""" + if not isinstance(data, dict): + return None + + # Check common keys + val = None + for key in ["box", "selected_index", "box_index", "index", "target"]: + if key in data: + val = data[key] + break + + if val is None: + return None + + # Handle null/None + if val in [None, "null", "None", "None ", "null "]: + return None + + # Convert to string and clean up common VLM prefixes + val_str = str(val).strip().lower() + + # Remove "box " prefix if present (e.g. "Box 5") + if val_str.startswith("box"): + val_str = val_str.replace("box", "").strip() + + # Attempt integer conversion + try: + # Extract first number if it's a messy string + import re + + match = re.search(r"(\d+)", val_str) + if match: + return int(match.group(1)) + return None + except (ValueError, TypeError): + return None + + def _validate_and_get_node(self, box_idx: Optional[int], box_map: Dict[int, SpatialNode]) -> Optional[SpatialNode]: + """Validates that the box index exists within the current SoM box_map.""" + if box_idx is None: + return None + + if box_idx in box_map: + return box_map[box_idx] + + return None + # ────────────────────────────────────────────── # Text-based Fallback (no device/screenshot) # ────────────────────────────────────────────── diff --git a/GramAddict/core/perception/screen_identity.py b/GramAddict/core/perception/screen_identity.py index 85db768..526d0cd 100644 --- a/GramAddict/core/perception/screen_identity.py +++ b/GramAddict/core/perception/screen_identity.py @@ -193,10 +193,11 @@ class ScreenIdentity: "row_profile_header_imageview", "profile_tabs_container", "profile_header_name", + "profile_header_business_category", ) if any(marker in ids for marker in PROFILE_MARKERS): - own_profile_texts = ("edit profile", "share profile", "profil bearbeiten", "profil teilen") - if selected_tab == "profile_tab" or any(m in desc_lower or m in text_lower for m in own_profile_texts): + # OWN_PROFILE is confirmed by the bottom tab OR the presence of 'edit' markers + if selected_tab == "profile_tab" or "profile_header_edit_profile_button" in ids: return ScreenType.OWN_PROFILE return ScreenType.OTHER_PROFILE @@ -206,9 +207,8 @@ class ScreenIdentity: if any(marker in ids for marker in REELS_MARKERS): return ScreenType.REELS_FEED - # DM thread detection — Semantic app-agnostic markers (chat input fields) - chat_input_markers = ["Message...", "Nachricht...", "Type a message", "Nachricht senden", "Send a message"] - if any(marker in texts for marker in chat_input_markers) or "direct_thread_header" in ids: + # DM thread detection — Structural markers (header and input fields) + if "direct_thread_header" in ids or "direct_text_input" in ids or "message_composer_container" in ids: return ScreenType.DM_THREAD # POST_DETAIL vs HOME_FEED: Both have row_feed_* markers. The differentiator @@ -233,8 +233,8 @@ class ScreenIdentity: ) if any(marker in ids for marker in STORY_MARKERS): return ScreenType.STORY_VIEW - # Fallback: content-desc "Like Story" or "Send story" confirms story context - if "like story" in desc_lower or "send story" in desc_lower or "nachricht senden" in desc_lower: + # Fallback: resource-id fragments confirm story context + if any(m in ids_str for m in ("reel_viewer", "story_viewer", "reel_viewer_media_layout")): return ScreenType.STORY_VIEW if selected_tab == "feed_tab": @@ -349,7 +349,6 @@ class ScreenIdentity: # Screen-specific actions desc_lower = " ".join(content_descs).lower() - text_lower = " ".join(texts).lower() if "like" in desc_lower: actions.append("tap like button") @@ -375,15 +374,12 @@ class ScreenIdentity: ): actions.append("tap follow button") + ids_str = " ".join(resource_ids).lower() if screen_type == ScreenType.OWN_PROFILE or screen_type == ScreenType.OTHER_PROFILE: - if "message" in desc_lower or "nachricht" in desc_lower: + if "button_message" in ids_str or "profile_header_message_button" in ids_str: actions.append("tap message button") - if ( - "following" in desc_lower - or "abonniert" in desc_lower - or "following" in text_lower - or "profile_header_following" in " ".join(resource_ids).lower() - ): + if "profile_header_following" in ids_str or "profile_header_follow_button" in ids_str: + # We can navigate to following list if we are following them actions.append("tap following list") # Grid items diff --git a/GramAddict/core/q_nav_graph.py b/GramAddict/core/q_nav_graph.py index 3fa1917..cb160c3 100644 --- a/GramAddict/core/q_nav_graph.py +++ b/GramAddict/core/q_nav_graph.py @@ -127,7 +127,8 @@ class QNavGraph: nav_graph.do("tap first grid item") # instead of _execute_transition("tap_explore_grid_item") """ - return self.goap._execute_action(goal) + screen = self.goap.perceive() + return self.goap._execute_action(goal, screen_state=screen) def _find_path(self, start: str, end: str): """Delegates to ScreenTopology for BFS pathfinding (SSOT).""" diff --git a/GramAddict/core/situational_awareness.py b/GramAddict/core/situational_awareness.py index 20d7d23..cc42857 100644 --- a/GramAddict/core/situational_awareness.py +++ b/GramAddict/core/situational_awareness.py @@ -305,10 +305,8 @@ class SituationalAwarenessEngine: "restrict certain activity", "help us confirm you own", "confirm it's you", - "später erneut versuchen", - "bestätige, dass du es bist", - "handlung blockiert", - "eingeschränkt", + "security_check", + "challenge", ] # Guard: Check if the text matches are relatively isolated (e.g. short strings). @@ -396,8 +394,8 @@ class SituationalAwarenessEngine: args = Config().args except Exception: pass - model = getattr(args, "ai_model", "qwen3.5:latest") - url = getattr(args, "ai_model_url", "http://localhost:11434/api/generate") + model = getattr(args, "ai_model") + url = getattr(args, "ai_model_url") res = query_telepathic_llm( model=model, @@ -539,8 +537,8 @@ class SituationalAwarenessEngine: args = Config().args except Exception: pass - model = getattr(args, "ai_telepathic_model", "llava:latest") - url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate") + model = getattr(args, "ai_telepathic_model") + url = getattr(args, "ai_telepathic_url") screenshot_b64 = getattr(self.device, "get_screenshot_b64", lambda: None)() res = query_telepathic_llm( @@ -592,13 +590,9 @@ class SituationalAwarenessEngine: from GramAddict.core.config import Config from GramAddict.core.llm_provider import query_telepathic_llm - try: - args = Config().args - model = getattr(args, "ai_telepathic_model", "llava:latest") - url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate") - except Exception: - model = "llava:latest" - url = "http://localhost:11434/api/generate" + args = Config().args + model = getattr(args, "ai_telepathic_model") + url = getattr(args, "ai_telepathic_url") system_prompt = ( "You are an Android UI navigation agent. Your job is to escape obstacles " diff --git a/tests/core/test_context_gate.py b/tests/core/test_context_gate.py new file mode 100644 index 0000000..9f57ead --- /dev/null +++ b/tests/core/test_context_gate.py @@ -0,0 +1,65 @@ +""" +P1-5: ContextGate Interaction Guards + +TDD RED: These tests enforce that: +1. Actions are blocked if they are structurally impossible on the current screen. +2. 'follow' is blocked on OWN_PROFILE. +3. 'comment' is blocked if no comment button is present in the XML. +4. 'like' is blocked if no heart/like button is present. +""" + +from GramAddict.core.perception.context_gate import ContextGate +from GramAddict.core.perception.screen_identity import ScreenType + + +class TestContextGate: + def test_block_follow_on_own_profile(self): + """RED: Following yourself is impossible and usually a sign of navigation drift.""" + gate = ContextGate() + + # Mock screen state + screen = { + "screen_type": ScreenType.OWN_PROFILE, + "resource_ids": {"profile_tab", "profile_edit_button"}, + "xml": "", + } + + assert gate.is_allowed("follow", screen) is False, "Should block 'follow' on OWN_PROFILE" + + def test_block_comment_without_button(self): + """RED: Blocking comment if the button isn't there (e.g. on a profile view).""" + gate = ContextGate() + + screen = { + "screen_type": ScreenType.OTHER_PROFILE, + "resource_ids": {"button_follow", "action_bar_overflow_icon"}, + "xml": "", + } + + # Commenting on a profile page is impossible (must be on post detail or feed) + assert gate.is_allowed("comment", screen) is False, "Should block 'comment' if no comment button is present" + + def test_allow_like_on_feed(self): + """GREEN (Target): Allow if the button exists.""" + gate = ContextGate() + + screen = { + "screen_type": ScreenType.HOME_FEED, + "resource_ids": {"row_feed_button_like", "row_feed_button_comment"}, + "xml": "", + } + + assert gate.is_allowed("like", screen) is True + + def test_block_post_interaction_on_dm_thread(self): + """RED: Prevent accidental likes/comments in DM threads.""" + gate = ContextGate() + + screen = { + "screen_type": ScreenType.DM_THREAD, + "resource_ids": {"direct_text_input", "direct_thread_header"}, + "xml": "", + } + + assert gate.is_allowed("like", screen) is False + assert gate.is_allowed("comment", screen) is False diff --git a/tests/core/test_intent_resolver_hardening.py b/tests/core/test_intent_resolver_hardening.py new file mode 100644 index 0000000..94d3c5e --- /dev/null +++ b/tests/core/test_intent_resolver_hardening.py @@ -0,0 +1,45 @@ +""" +P1-7: IntentResolver Box Index Clamping + +TDD RED: These tests enforce that: +1. VLM responses with string box indices are correctly parsed as integers. +2. Hallucinated indices (OOB) are gracefully rejected without crashing. +3. Common VLM hallucination patterns (e.g. {"box": "Box 5"}) are handled. +""" + +from GramAddict.core.perception.intent_resolver import IntentResolver +from GramAddict.core.perception.spatial_parser import SpatialNode + + +class TestIntentResolverHardening: + def test_parse_box_index_variants(self): + """RED: Resolver should handle various VLM JSON formats and types.""" + resolver = IntentResolver() + + # Test basic int + assert resolver._parse_box_index({"box": 5}) == 5 + + # Test string int + assert resolver._parse_box_index({"box": "5"}) == 5 + + # Test prefixed string + assert resolver._parse_box_index({"box": "Box 5"}) == 5 + + # Test alternate keys + assert resolver._parse_box_index({"selected_index": 3}) == 3 + assert resolver._parse_box_index({"box_index": "7"}) == 7 + + # Test invalid + assert resolver._parse_box_index({"box": "none"}) is None + assert resolver._parse_box_index({"box": "null"}) is None + + def test_box_clamping_rejection(self): + """RED: Reject indices that are not in the box_map.""" + resolver = IntentResolver() + box_map = {0: SpatialNode(0, 0, 10, 10), 1: SpatialNode(10, 10, 20, 20)} + + # Index 99 is OOB + assert resolver._validate_and_get_node(99, box_map) is None + + # Index 1 is valid + assert resolver._validate_and_get_node(1, box_map) is not None