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