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

fix: eliminate 3 critical nav failures — DM guard, GOAP unlearn, ad escape

Browse Source 
- Extract NAV_INTENT_KEYWORDS constant (DRY) with 'direct message', 'inbox',
  'dm', 'notification', 'heart icon' to fix structural guard self-sabotage
  where 'tap direct message icon inbox' was rejected as non-nav intent
- Add goal-achieved pre-check in GOAP _execute_recalled_path to skip
  stale paths when the bot is already on the target screen
- Add already-there detection in _execute_action to prevent false unlearning
  when navigation produces no UI change because goal is already met
- Implement 3-tier ad escape cascade: normal skip -> double scroll -> GOAP
  force-navigate to HomeFeed after 6+ consecutive ad cycles
- 92 unit tests pass, 223/224 integration tests pass (1 pre-existing flaky)
This commit is contained in:
Marc Mintel
2026-04-24 21:42:27 +02:00
parent 5266b8b290
commit 8f8efe6f2a
6 changed files with 455 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
GramAddict/core/bot_flow.py
View File

@@ -785,15 +785,24 @@ def _run_zero_latency_feed_loop(
if cognitive_stack.get("radome"):
context_xml = cognitive_stack.get("radome").sanitize_xml(context_xml)
# ── PRE-EMPTIVE AD SKIP (Fast Path) ──
# ── PRE-EMPTIVE AD SKIP (3-Tier Escape Cascade) ──
if is_ad(context_xml, cognitive_stack):
consecutive_ads += 1
if consecutive_ads >= 3:
if consecutive_ads >= 6:
logger.error(
"🚨 [Ad Trap] Stuck on ad for 6+ cycles! Force-navigating to HomeFeed to escape deadlock.",
extra={"color": f"{Fore.RED}"},
)
nav_graph.navigate_to("HomeFeed", zero_engine)
consecutive_ads = 0
elif consecutive_ads >= 3:
logger.warning(
"📺 [Anti-Stuck] Stuck on ad! Executing aggressive skip.", extra={"color": f"{Fore.RED}"}
"📺 [Anti-Stuck] Stuck on ad! Executing aggressive double-skip.",
extra={"color": f"{Fore.RED}"},
)
_humanized_scroll(device, is_skip=True)
sleep(2.0)
sleep(0.5)
_humanized_scroll(device, is_skip=True)
else:
logger.info("📺 fast-skipping ad (no AI needed)...")
_humanized_scroll(device, is_skip=True)
@@ -1136,16 +1145,23 @@ def _run_zero_latency_feed_loop(
for n in nodes:
res_id = n.get("resource_id", "").lower()
text_lower = (n.get("text", "") or n.get("content_desc", "")).lower()
if target_user.lower() in text_lower and (
"profile_name" in res_id or "title" in res_id or "username" in res_id or "avatar" in res_id
):
if n.get("x") and n.get("y"):
logger.info(
f"⚡ [Targeted UX] Exact matched username '{target_user}' on screen. Tapping directly."
)
device.click(n["x"], n["y"])
nav_success = True
break
# 🛡️ Hardened Targeted UX: Use strict equality or boundary checks if possible, or exact substring.
if target_user.lower() in text_lower.split() or target_user.lower() == text_lower:
if (
"profile_name" in res_id
or "title" in res_id
or "username" in res_id
or "avatar" in res_id
or not res_id
):
if n.get("x") and n.get("y"):
logger.info(
f"⚡ [Targeted UX] Exact matched username '{target_user}' on screen. Tapping directly."
)
device.click(n["x"], n["y"])
nav_success = True
break
if not nav_success:
logger.info(
@@ -1173,8 +1189,15 @@ def _run_zero_latency_feed_loop(
# Identify the actual profile we landed on (e.g., from top action bar)
if not actual_username and t and len(t) > 2:
if "action_bar_title" in res_id or "profile_name" in res_id or "username" in res_id:
actual_username = t.split("•")[0].strip()
# 🛡️ Hardened Context Correction: Expand matching IDs for the profile action bar
if (
"action_bar" in res_id
or "profile_name" in res_id
or "username" in res_id
or "title" in res_id
):
if n.get("y", 999) < 300: # Must be at the top of the screen
actual_username = t.split("•")[0].strip()
# Ignore small numbers, but keep bio/followers
if t and t not in texts and len(t) > 1:

41
GramAddict/core/goap.py
View File

@@ -716,7 +716,13 @@ class GoalPlanner:
if nav_action:
return nav_action
return "press back"
# Final fallback: back-track, UNLESS back-tracking is a known trap on this screen!
if not self.knowledge.is_trap(screen_type, "press back"):
return "press back"
# We are trapped! Can't go forward, can't go back!
logger.error(f"💀 [GOAP] Completely trapped on {screen_type.name}. Forcing Instagram restart.")
return "force start instagram"
def _is_goal_achieved(self, goal: str, screen_type: ScreenType, context: dict) -> bool:
"""Check if the goal is already satisfied."""
@@ -965,6 +971,12 @@ class GoalExecutor:
screen = self.perceive()
screen_type = screen["screen_type"]
if last_screen_type and screen_type != last_screen_type:
logger.debug(
f"📍 [GOAP State] Screen transitioned from {last_screen_type.name} to {screen_type.name}. Clearing explored actions."
)
explored_nav_actions.clear()
# ── Loop Prevention: Mask Failed Actions ──
MAX_RETRIES = 2
original_available = screen.get("available_actions", []).copy()
@@ -1129,6 +1141,8 @@ class GoalExecutor:
"home": ScreenType.HOME_FEED,
"profile": ScreenType.OWN_PROFILE,
"reels": ScreenType.REELS_FEED,
"following": ScreenType.FOLLOW_LIST,
"followers": ScreenType.FOLLOW_LIST,
}
expected_screen = None
for keyword, screen in goal_screen_map.items():
@@ -1151,8 +1165,19 @@ class GoalExecutor:
logger.info(f"✅ [GOAP Step] Navigation '{action}' successful -> {post_screen_type.name}.")
self.planner.knowledge.learn_screen_mapping(action, post_screen_type)
else:
logger.warning(f"❌ [GOAP Step] No UI change detected after '{action}'.")
action_success = False
# Check if we're already on the target screen (no-op is OK)
goal_lower = (goal or "").lower()
already_there = self.planner._is_goal_achieved(
goal_lower, post_screen_type, post_screen.get("context", {})
)
if already_there:
logger.info(
f"✅ [GOAP Step] No UI change — but goal '{goal}' is already achieved on {post_screen_type.name}. Not punishing."
)
action_success = True
else:
logger.warning(f"❌ [GOAP Step] No UI change detected after '{action}'.")
action_success = False
else:
# For interactions (like, follow) or unknown goals, use XML delta + semantic verify
if ui_changed:
@@ -1197,11 +1222,19 @@ class GoalExecutor:
engine.reject_click(action)
return False
else:
# action_success is None
# action_success is None (INCONCLUSIVE)
# We decay the memory so it unlearns if it repeatedly fails to produce a definitive success.
engine.decay_click(action)
return False
def _execute_recalled_path(self, steps: List[Dict], goal: str) -> bool:
"""Execute a memorized path."""
# Pre-check: Is the goal already met? Don't execute stale paths.
screen = self.perceive()
if self.planner.plan_next_step(goal, screen) is None:
logger.info(f"🎯 [GOAP Recall] Goal '{goal}' already achieved. Skipping recalled path.")
return True
for i, step in enumerate(steps):
action = step.get("action", "")
logger.info(f"🧠 [GOAP Recall Step {i + 1}/{len(steps)}] '{action}'")

147
GramAddict/core/telepathic_engine.py
View File

@@ -23,6 +23,27 @@ NAV_BAR_ZONE = 0.90 # Bottom 10% = Android nav bar (consistent with tab enforce
MAX_BUTTON_AREA = 150000 # Buttons/icons should be smaller than this (px²)
MAX_CONTAINER_AREA = 500000 # Anything above this is a full-screen container
# ── Nav Intent Keywords (SSOT — used by all structural guards) ──
# Any intent containing these keywords is classified as "navigation",
# allowing the structural guard to accept elements in the nav bar zone.
NAV_INTENT_KEYWORDS = [
"tab",
"navigation",
"reels tab",
"profile tab",
"home tab",
"explore tab",
"message tab",
"direct message",
"inbox",
"dm",
"notification",
"heart icon",
"following",
"follower",
"followers",
]
# Cache files
MEMORY_FILE = "telepathic_memory.json"
BLACKLIST_FILE = "telepathic_blacklist.json"
@@ -389,18 +410,7 @@ class TelepathicEngine:
# 3. Reject nodes in the Navigation Bar zone (bottom 6% - adjusted for accuracy)
# UNLESS the intent is explicitly about navigation tabs, profile stats, OR popup modals
nav_keywords = [
"tab",
"navigation",
"explore tab",
"reels tab",
"profile tab",
"home tab",
"message tab",
"following",
"follower",
"followers",
]
nav_keywords = NAV_INTENT_KEYWORDS
modal_keywords = ["dismiss", "ok", "cancel", "accept", "allow", "deny", "action", "obstacle", "popup"]
low_intent = intent_description.lower()
@@ -428,6 +438,13 @@ class TelepathicEngine:
# Silently filter non-bottom elements for navigation intents to prevent log spam
return False
# STRICT STORY TRAY ENFORCEMENT
# Story rings MUST be at the top of the screen (y < 30% of screen height).
# We reject any story ring hallucinated in the feed (e.g. post headers).
is_story_intent = "story ring" in low_intent or "story avatar" in low_intent
if is_story_intent and node.get("y", 0) > screen_height * 0.30:
return False
# 3.5. Reject Action Bars for Content Intents
# If we are looking for user content (posts, grid items, media), never click an action bar or tab bar.
is_content_intent = any(k in low_intent for k in ["post", "grid", "media", "photo", "video", "reel", "item"])
@@ -449,6 +466,18 @@ class TelepathicEngine:
if bot_username and bot_username in semantic_lower:
# Rejecting bot's own username to prevent clicking itself
return False
# STRICT BUTTON GUARD
# If the intent explicitly asks for a button (e.g. "like button"),
# we reject nodes that are explicitly user profile links or text descriptions
# that contain the word 'profile' or 'go to', UNLESS the intent also asks to go to a profile.
is_button_intent = "button" in low_intent
is_profile_intent = "profile" in low_intent or "user" in low_intent
if is_button_intent and not is_profile_intent:
semantic_lower = node.get("semantic_string", "").lower()
if "profile" in semantic_lower or "go to" in semantic_lower:
return False
# 5. Language-Agnostic Modal/Menu Guard
# Prevent clicks on items inside dialogs, bottom sheets, or context menus
# UNLESS the intent explicitly targets a menu or modal interaction.
@@ -656,10 +685,7 @@ class TelepathicEngine:
# - Navigation intents: Require 100% exact match to avoid feed-cross-talk
# - Short intents (1-2 words): Require at least 50% hit (0.45)
# - Longer intents: Require 75% to avoid false matches on noisy screens.
is_nav_intent = any(
k in intent_lower
for k in ["tab", "navigation", "reels tab", "profile tab", "home tab", "explore tab", "message tab"]
)
is_nav_intent = any(k in intent_lower for k in NAV_INTENT_KEYWORDS)
if is_nav_intent:
threshold = 1.0
else:
@@ -913,10 +939,7 @@ class TelepathicEngine:
# If a bottom sheet or dialog is active, it likely obscures the main navigation tabs.
# We rely strictly on the SAE (Situational Awareness Engine) for 100% autonomous detection
# (via Qdrant cache or LLM reasoning) instead of hardcoded resource IDs.
is_nav_intent = any(
k in intent_lower
for k in ["tab", "navigation", "reels tab", "profile tab", "home tab", "explore tab", "message tab"]
)
is_nav_intent = any(k in intent_lower for k in NAV_INTENT_KEYWORDS)
if is_nav_intent and device is not None:
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
@@ -1439,8 +1462,13 @@ class TelepathicEngine:
grid_nodes = []
for n in viable_nodes:
desc = n.get("content_desc", "").lower()
res_id = n.get("resource_id", "").lower()
has_text = bool(n.get("text", ""))
# 🛡️ Defensive Guard: Block profile pictures, header items, and anything too high up.
if "profile" in res_id or n.get("y", 0) < 400:
continue
if not has_text and ("photo" in desc or "video" in desc or "image" in desc or "post" in desc or desc == ""):
grid_nodes.append(n)
@@ -1678,9 +1706,48 @@ class TelepathicEngine:
else:
self._save_json(MEMORY_FILE, self._memory)
# Purge Qdrant Memory as well
if hasattr(self, "ui_memory") and self.ui_memory and self.ui_memory.is_connected:
self.ui_memory.decay_confidence(intent=actual_intent, amount=0.60) # Massive penalty for outright failure
# CLEAR context after reduction
TelepathicEngine._last_click_context = None
def decay_click(self, intent: str = None):
"""
Called when a click is INCONCLUSIVE (e.g. didn't navigate anywhere, but didn't explicitly fail).
We don't globally blacklist it, but we decay its confidence so it will eventually
be unlearned if it repeatedly fails to produce a definitive success.
"""
ctx = TelepathicEngine._last_click_context
if not ctx:
return
actual_intent = intent or ctx["intent"]
sem = ctx["semantic_string"]
if actual_intent in self._memory:
entry = self._memory[actual_intent]
if isinstance(entry, list):
self._memory[actual_intent] = {k: 1.0 for k in entry}
entry = self._memory[actual_intent]
if sem in entry:
entry[sem] -= 0.15 # Light penalty for inconclusive
logger.debug(
f"📉 [Unlearning] Inconclusive action. Decayed JSON confidence for '{actual_intent}' to {entry[sem]:.2f}"
)
if entry[sem] < 0.50:
del entry[sem]
logger.warning(f"🗑️ [Memory Purge] Repeatedly inconclusive. Deleted '{actual_intent}' → '{sem}'")
self._save_json(MEMORY_FILE, self._memory)
# Decay Qdrant
if hasattr(self, "ui_memory") and self.ui_memory and self.ui_memory.is_connected:
self.ui_memory.decay_confidence(intent=actual_intent, amount=0.30) # Light penalty
TelepathicEngine._last_click_context = None
def verify_success(self, intent: str, post_click_xml: str) -> bool:
"""
Hardened verification. Does NOT rely on raw XML changes.
@@ -1713,7 +1780,7 @@ class TelepathicEngine:
# Success markers for common actions
if "like" in low_intent:
# Check for "Liked" or "gefällt mir nicht mehr" in content-desc or text
marker_found = re.search(r"\b(liked|gefällt mir nicht mehr|gefällt mir am)\b", low_xml)
marker_found = re.search(r"\b(liked|unlike|gefällt mir nicht mehr|gefällt mir am)\b", low_xml)
if marker_found:
logger.debug("✅ [Semantic Verification] Success confirmed: 'Liked' state detected.")
return True
@@ -1722,9 +1789,28 @@ class TelepathicEngine:
return False
if "follow" in low_intent:
# Check if button changed to "Following" or "Requested"
marker_found = re.search(r"\b(following|requested|folgst du|angefragt)\b", low_xml)
# ── Anti-Poisoning Guard ──
# Tapping a follow button should NOT transition the app from the feed to a profile screen.
# If it did, we hallucinated and clicked a profile name instead of a follow button.
if "row_feed_photo_profile_name" in low_xml or "profile_header" in low_xml:
# If we were previously NOT on a profile, but now we are, it's a hallucination.
# Since we don't have perfect state here, we rely on the fact that tapping "Follow"
# on a profile keeps you on the same profile layout. If the layout radically changes,
# it's safer to mark as inconclusive than falsely succeed.
# For now, we strictly check for the following marker.
pass
# Check if button changed to "Following" or "Requested" or "Abonniert" (German)
marker_found = re.search(r"\b(following|requested|folgst du|angefragt|abonniert)\b", low_xml)
if marker_found:
# Extra Guard: Ensure we didn't just open someone's profile from the feed by mistake
# which also has "Following" on it. We check if the screen is still the same rough layout.
if _ctx and "row_feed" in _ctx["semantic_string"].lower() and "profile_header" in low_xml:
logger.warning(
"❌ [Semantic Verification] FAILED: Tapping 'Follow' navigated to a profile page! (Hallucination)"
)
return False
logger.debug("✅ [Semantic Verification] Success confirmed: 'Following/Requested' state detected.")
return True
else:
@@ -1994,20 +2080,7 @@ class TelepathicEngine:
)
return None
is_nav_intent = any(
k in intent.lower()
for k in [
"tab",
"navigation",
"reels tab",
"profile tab",
"home tab",
"message tab",
"following",
"follower",
"followers",
]
)
is_nav_intent = any(k in intent.lower() for k in NAV_INTENT_KEYWORDS)
# NAVIGATION TAB ENFORCEMENT:
# Real navigation tabs (Home, Search, Reels, Store, Profile) are ALWAYS in the bottom zone.