mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat(core): complete P1 hardening phase - ContextGate, Localization Purge, SoM Safety Clamp

Browse Source
This commit is contained in:
Marc Mintel
2026-05-04 14:51:54 +02:00
parent 5c9ea0e5e2
commit 79e5784b93
9 changed files with 281 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
GramAddict/core/bot_flow.py
View File

@@ -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

16
GramAddict/core/goap.py
View File

@@ -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

78
GramAddict/core/perception/context_gate.py Normal file
View File

@@ -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

64
GramAddict/core/perception/intent_resolver.py
View File

@@ -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)
# ──────────────────────────────────────────────

26
GramAddict/core/perception/screen_identity.py
View File

@@ -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

3
GramAddict/core/q_nav_graph.py
View File

@@ -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)."""

24
GramAddict/core/situational_awareness.py
View File

@@ -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 "

65
tests/core/test_context_gate.py Normal file
View File

@@ -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": "<node resource-id='profile_edit_button' />",
}
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": "<node resource-id='button_follow' />",
}
# 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": "<node resource-id='row_feed_button_like' />",
}
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": "<node resource-id='direct_text_input' />",
}
assert gate.is_allowed("like", screen) is False
assert gate.is_allowed("comment", screen) is False

45
tests/core/test_intent_resolver_hardening.py Normal file
View File

@@ -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