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

feat: structural hardening, grid fast-paths, and state-aware adaptive snap recovery

Browse Source
This commit is contained in:
Marc Mintel
2026-05-04 17:43:25 +02:00
parent 22216cbd2d
commit 3800254fe3
22 changed files with 262 additions and 761 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
GramAddict/core/account_switcher.py
View File

@@ -10,7 +10,10 @@ def verify_and_switch_account(device, nav_graph, target_username):
logger.info(f"🛂 [Identity Guard] Verifying if active account matches target: '{target_username}'")
# 1. Navigate to OwnProfile to reliably check identity
success = nav_graph.navigate_to("OwnProfile", zero_engine=None)
from GramAddict.core.goap import GoalExecutor
goap = GoalExecutor.get_instance(device, target_username)
success = goap.achieve("open profile")
if not success:
logger.error("❌ [Identity Guard] Failed to reach OwnProfile to verify account.")
return False
@@ -39,22 +42,31 @@ def verify_and_switch_account(device, nav_graph, target_username):
logger.warning(f"🔄 [Identity Guard] Account mismatch detected! Switching to '{target_username}'...")
# 3. Find the Profile Tab to long press using Telepathic Engine (Blank Start)
# 3. Find the Profile Tab to long press using deterministic structural markers
profile_tab = None
try:
from GramAddict.core.telepathic_engine import TelepathicEngine
# Priority 1: Structural ID
tab_view = device.find(resourceIdMatches=".*profile_tab.*")
if tab_view.exists():
bounds = tab_view.info.get("bounds")
if bounds:
left, top, right, bottom = bounds["left"], bounds["top"], bounds["right"], bounds["bottom"]
profile_tab = ((left + right) // 2, (top + bottom) // 2)
telepath = TelepathicEngine.get_instance()
# Priority 2: Geometric Fallback (Bottom Right)
if not profile_tab:
info = device.get_info()
width = info.get("displayWidth", 1080)
height = info.get("displayHeight", 2400)
# Profile tab is typically in the bottom right corner (last 20% of width, bottom 10% of height)
profile_tab = (int(width * 0.9), int(height * 0.95))
logger.info(f"📐 [Identity Guard] Using geometric fallback for profile tab: {profile_tab}")
# We ask the semantic engine to find the profile tab, ensuring 100% ID-agnostic behavior
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3, device=device)
if profile_tab_node:
profile_tab = (profile_tab_node["x"], profile_tab_node["y"])
except Exception as e:
logger.warning(f"Error resolving profile tab via telepathic engine: {e}")
logger.warning(f"Error resolving profile tab structurally: {e}")
if not profile_tab:
logger.error("❌ [Identity Guard] Cannot find profile_tab semantically to initiate account switch!")
logger.error("❌ [Identity Guard] Cannot find profile_tab to initiate account switch!")
return False
# Long press to open account selector

1
GramAddict/core/behaviors/__init__.py
View File

@@ -51,6 +51,7 @@ class BehaviorResult:
executed: bool = False # Did the behavior actually do something?
should_continue: bool = True # Should the feed loop continue to next post?
should_skip: bool = False # Should we skip to the next post immediately?
skip_type: str = "normal" # "normal" (humanized) or "fast" (ad evasion)
interactions: int = 0 # Number of interactions performed
metadata: Dict[str, Any] = field(default_factory=dict) # Plugin-specific data

15
GramAddict/core/behaviors/ad_guard.py
View File

@@ -1,8 +1,6 @@
import logging
from time import sleep
from GramAddict.core.behaviors import BehaviorContext, BehaviorPlugin, BehaviorResult
from GramAddict.core.physics.humanized_input import humanized_scroll
from GramAddict.core.utils import is_ad
logger = logging.getLogger(__name__)
@@ -56,19 +54,16 @@ class AdGuardPlugin(BehaviorPlugin):
if nav_graph:
nav_graph.navigate_to("HomeFeed", zero_engine)
self.consecutive_ads = 0
return BehaviorResult(executed=True, should_skip=True)
return BehaviorResult(executed=True, should_skip=True, skip_type="fast")
logger.info(f"🛡️ [AdGuard] Ad detected ({self.consecutive_ads}). Scrolling past it...")
humanized_scroll(ctx.device, is_skip=True)
sleep(1.0 * ctx.sleep_mod)
logger.info(f"🛡️ [AdGuard] Ad detected ({self.consecutive_ads}). Delegating skip to orchestrator...")
# Aggressive double skip for triple ad
if self.consecutive_ads >= 3:
logger.info("🛡️ [AdGuard] Aggressive skip for consecutive ads.")
humanized_scroll(ctx.device, is_skip=True)
sleep(1.0 * ctx.sleep_mod)
logger.info("🛡️ [AdGuard] Requesting aggressive double skip for consecutive ads.")
return BehaviorResult(executed=True, should_skip=True, skip_type="double_fast")
return BehaviorResult(executed=True, should_skip=True)
return BehaviorResult(executed=True, should_skip=True, skip_type="fast")
def reset_counter(self):
self.consecutive_ads = 0

2
GramAddict/core/behaviors/grid_like.py
View File

@@ -78,7 +78,7 @@ class GridLikePlugin(BehaviorPlugin):
if not nav_graph.do("tap first image post in profile grid"):
return BehaviorResult(executed=False, metadata={"reason": "grid_nav_failed"})
if not wait_for_post_loaded(ctx.device, timeout=5):
if not wait_for_post_loaded(ctx.device, timeout=5, nav_graph=nav_graph):
logger.warning(f"❌ [GridLike] Post failed to open from profile grid of @{ctx.username}.")
return BehaviorResult(executed=False, metadata={"reason": "post_load_failed"})

7
GramAddict/core/behaviors/post_interaction.py
View File

@@ -1,8 +1,6 @@
import logging
from time import sleep
from GramAddict.core.behaviors import BehaviorContext, BehaviorPlugin, BehaviorResult
from GramAddict.core.physics.humanized_input import humanized_scroll
logger = logging.getLogger(__name__)
@@ -43,9 +41,6 @@ class PostInteractionPlugin(BehaviorPlugin):
if telemetry:
telemetry.log_post_interaction(ctx.post_data, ctx.shared_state.get("session_outcomes", []))
humanized_scroll(ctx.device)
sleep(1.5 * ctx.sleep_mod)
return BehaviorResult(
executed=True, should_skip=True
executed=True, should_skip=True, skip_type="normal"
) # should_skip=True signals the feed loop to restart for the next post

12
GramAddict/core/behaviors/resonance_evaluator.py
View File

@@ -1,9 +1,7 @@
import logging
import random
from time import sleep
from GramAddict.core.behaviors import BehaviorContext, BehaviorPlugin, BehaviorResult
from GramAddict.core.physics.humanized_input import humanized_scroll
logger = logging.getLogger(__name__)
@@ -71,9 +69,9 @@ class ResonanceEvaluatorPlugin(BehaviorPlugin):
marker = vibe.get("ad_marker_text")
if marker and marker.strip():
from GramAddict.core.utils import learn_ad_marker
learn_ad_marker(marker, ctx.context_xml)
humanized_scroll(ctx.device)
return BehaviorResult(executed=True, should_skip=True)
return BehaviorResult(executed=True, should_skip=True, skip_type="fast")
# BUG 6 Fix: VLM returns {"should_like": true/false}, not "quality_score"
should_like = vibe.get("should_like", False)
@@ -97,10 +95,8 @@ class ResonanceEvaluatorPlugin(BehaviorPlugin):
{"username": ctx.username, "resonance": res_score, "action": "skip"}
)
# If we skip here, we MUST scroll to next post and terminate chain
humanized_scroll(ctx.device)
sleep(1.0 * ctx.sleep_mod)
return BehaviorResult(executed=True, should_skip=True)
# Delegate scrolling to the orchestrator
return BehaviorResult(executed=True, should_skip=True, skip_type="fast")
dopamine = ctx.cognitive_stack.get("dopamine")
if dopamine:

44
GramAddict/core/bot_flow.py
View File

@@ -249,6 +249,7 @@ def start_bot(**kwargs):
"crm": crm_db,
"dm_memory": dm_memory_db,
"writer": writer,
"system_recovery": {"consecutive_neutral": 0, "consecutive_same_user": 0, "last_username": None},
}
from GramAddict.core.behaviors import PluginRegistry
@@ -1046,19 +1047,60 @@ def _run_zero_latency_feed_loop(
# (for legacy interaction engine below until extracted)
should_continue_loop = True
skip_type = "normal"
for result in plugin_results:
if result.metadata.get("return_code") == "CONTEXT_LOST":
return "CONTEXT_LOST"
if result.executed and result.should_skip:
should_continue_loop = False
skip_type = getattr(result, "skip_type", "normal")
break
# ── System State Recovery: Loop Detection ──
sys_rec = cognitive_stack.get("system_recovery")
if sys_rec is not None:
current_user = ctx.username if ctx.username else getattr(ctx, "post_data", {}).get("username")
if current_user and current_user == sys_rec.get("last_username"):
sys_rec["consecutive_same_user"] += 1
if sys_rec["consecutive_same_user"] > 3:
logger.error(
f"🚨 [SystemStateRecovery] Stuck in loop on post by @{current_user}! Forcing HomeFeed restart."
)
nav_graph.navigate_to("HomeFeed", zero_engine)
sys_rec["consecutive_same_user"] = 0
continue
else:
sys_rec["consecutive_same_user"] = 0
sys_rec["last_username"] = current_user
if not should_continue_loop:
if skip_type == "double_fast":
_humanized_scroll(device, is_skip=True)
sleep(1.0 * sleep_mod)
_humanized_scroll(device, is_skip=True)
sleep(1.0 * sleep_mod)
elif skip_type == "fast":
_humanized_scroll(device, is_skip=True)
sleep(1.0 * sleep_mod)
else:
_humanized_scroll(device, is_skip=False)
sleep(1.5 * sleep_mod)
continue
res_score = ctx.shared_state.get("res_score", 0.5)
# ── Advance to next post ──
if sys_rec is not None:
if 0.4 <= res_score <= 0.6:
sys_rec["consecutive_neutral"] += 1
if sys_rec["consecutive_neutral"] > 3:
logger.error(
f"🚨 [SystemStateRecovery] Neutral resonance for {sys_rec['consecutive_neutral']} posts. Forcing HomeFeed restart."
)
nav_graph.navigate_to("HomeFeed", zero_engine)
sys_rec["consecutive_neutral"] = 0
continue
else:
sys_rec["consecutive_neutral"] = 0
# ── Advance to next post ──
_humanized_scroll(device, resonance_score=res_score)

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

@@ -353,6 +353,14 @@ class IntentResolver:
)
return story_nodes[0]
# --- Structural Grid Fast-Paths ---
if "first image post in profile grid" in intent_lower or "first post" in intent_lower:
for node in candidates:
desc = (node.content_desc or "").lower()
if "row 1, column 1" in desc:
logger.info(f"🎯 [Structural Fast-Path] Found first grid post: {node.resource_id} (desc: '{desc}')")
return node
# --- Navigation Tab Fast-Paths ---
# Deterministically identify bottom navigation tabs to prevent VLM confusion
tab_map = {
@@ -368,12 +376,22 @@ class IntentResolver:
}
for intent_key, resource_suffix in tab_map.items():
if intent_key in intent_lower:
# Priority 1: Structural lookup
for node in candidates:
rid = (node.resource_id or "").lower()
if rid.endswith(f":id/{resource_suffix}"):
logger.info(f"🎯 [Structural Fast-Path] Found {intent_key}: {rid}")
return node
# Priority 2: Fail Fast
# If a navigation tab is requested but not structurally present, it means the bottom
# bar is likely hidden (e.g. full-screen ad, post detail). We MUST NOT guess
# coordinates and we MUST NOT ask the VLM, as this leads to Play Store traps.
logger.warning(
f"🛡️ [Navigation Guard] Structural {intent_key} not found. Failing intent deterministically."
)
return None
# --- Semantic Match Guard ---
# If the intent explicitly quotes a target (e.g., "tap 'New Message'"),
# we strictly filter candidates to those whose text or content_desc contains the quote.

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

@@ -177,11 +177,14 @@ class ScreenIdentity:
# Priority 1: Content-creation overlays that block ALL navigation.
# These full-screen Instagram UIs have no navigation tabs and trap the bot.
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
if not is_normal_override:
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
browser_markers = ("ig_browser_text_title", "ig_browser_close_button", "ig_chrome_subsection")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
if any(marker in ids_str for marker in browser_markers):
logger.info("🛡️ [ScreenIdentity] In-App Browser detected → MODAL")
return ScreenType.MODAL
# Priority 2: Structural Heuristics (100% Deterministic)
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
@@ -207,6 +210,20 @@ class ScreenIdentity:
if any(marker in ids for marker in REELS_MARKERS):
return ScreenType.REELS_FEED
# Story view structural markers — present in full-screen story viewer.
# Stories hide the navigation tab bar, so selected_tab is always None.
# Must be checked BEFORE tab-based fallbacks and DM_THREAD to prevent UNKNOWN/DM classification due to message input.
STORY_MARKERS = (
"reel_viewer_media_layout",
"reel_viewer_header",
"reel_viewer_progress_bar",
"reel_viewer_root",
"story_viewer_container",
"reel_viewer_content_layout",
)
if any(marker in ids for marker in STORY_MARKERS):
return ScreenType.STORY_VIEW
# 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
@@ -220,23 +237,6 @@ class ScreenIdentity:
if "main_feed_action_bar" not in ids:
return ScreenType.POST_DETAIL
# Story view structural markers — present in full-screen story viewer.
# Stories hide the navigation tab bar, so selected_tab is always None.
# Must be checked BEFORE tab-based fallbacks to prevent UNKNOWN classification.
STORY_MARKERS = (
"reel_viewer_media_layout",
"reel_viewer_header",
"reel_viewer_progress_bar",
"reel_viewer_root",
"story_viewer_container",
"reel_viewer_content_layout",
)
if any(marker in ids for marker in STORY_MARKERS):
return ScreenType.STORY_VIEW
# 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":
return ScreenType.HOME_FEED
if selected_tab == "clips_tab":
@@ -397,8 +397,6 @@ class ScreenIdentity:
"""Extract meaningful context from the screen."""
context = {}
desc_text = " ".join(content_descs)
# Post/follower counts
for d in content_descs:
m = re.match(r"([\d,.]+K?M?)(\s*)(posts?|followers?|following)", d, re.IGNORECASE)

20
GramAddict/core/physics/timing.py
View File

@@ -53,6 +53,7 @@ def wait_for_post_loaded(device, timeout=5, nav_graph=None):
try:
xml_lower = xml.lower()
# 1. Trapped in a Story or Reel viewer? Press back.
if "reel_viewer_root" in xml_lower or "clips_viewer" in xml_lower:
logger.warning("🧗 [Adaptive Snap] Trapped in Story/Reel viewer. Pressing BACK.")
@@ -60,20 +61,33 @@ def wait_for_post_loaded(device, timeout=5, nav_graph=None):
sleep(1.5)
# Give it one more chance to load the feed
xml = device.dump_hierarchy()
xml_lower = xml.lower()
if any(marker in xml for marker in FEED_MARKERS):
logger.info("✅ Recovered to Feed.")
return True
# 2. Trapped in Profile?
if "profile_header" in xml_lower and "row_feed_photo_profile_name" not in xml_lower:
# Only press back if we did NOT intend to be on a profile!
expected_state = nav_graph.current_state if nav_graph else ""
if (
expected_state != "ProfileView"
and "profile_header" in xml_lower
and "row_feed_photo_profile_name" not in xml_lower
):
logger.warning("🧗 [Adaptive Snap] Trapped in Profile. Pressing BACK.")
device.press("back")
sleep(1.5)
xml = device.dump_hierarchy()
xml_lower = xml.lower()
# 3. Stuck on Grid? The tap didn't register. Do not wobble.
grid_markers = ["explore_tab", "explore_grid", "grid_card_layout_container", "profile_tabs_container"]
if any(m in xml_lower for m in grid_markers):
logger.warning("🧗 [Adaptive Snap] Detected bot is STILL on the Grid. Tap likely missed. Aborting snap.")
if any(m in xml_lower for m in grid_markers) or (
"profile_header" in xml_lower and "row_feed_photo_profile_name" not in xml_lower
):
logger.warning(
"🧗 [Adaptive Snap] Detected bot is STILL on the Grid/Profile. Tap likely missed. Aborting snap."
)
return False
# 4. Stuck between posts (Feed markers not fully visible)? Micro-wobble.

7
GramAddict/core/screen_topology.py
View File

@@ -66,14 +66,11 @@ class ScreenTopology:
"tap explore tab": ScreenType.EXPLORE_GRID,
"tap reels tab": ScreenType.REELS_FEED,
"tap profile tab": ScreenType.OWN_PROFILE,
"press back": ScreenType.HOME_FEED,
},
ScreenType.POST_DETAIL: {
"tap home tab": ScreenType.HOME_FEED,
"tap explore tab": ScreenType.EXPLORE_GRID,
"tap profile tab": ScreenType.OWN_PROFILE,
"tap reels tab": ScreenType.REELS_FEED,
"tap view all comments": ScreenType.COMMENTS,
"press back": ScreenType.EXPLORE_GRID, # Default fallback if unknown source
"press back": ScreenType.EXPLORE_GRID,
},
ScreenType.COMMENTS: {
"press back": ScreenType.POST_DETAIL,

48
GramAddict/core/situational_awareness.py
View File

@@ -422,30 +422,13 @@ class SituationalAwarenessEngine:
logger.warning(f"⚠️ [Smart Perceive] LLM Classification failed ({e}). Defaulting to FOREIGN_APP.")
return SituationType.OBSTACLE_FOREIGN_APP
# ── Modal/Obstacle Detection (Autonomous LLM + Memory) ──
# We explicitly query ScreenMemoryDB. If unknown, we ask the LLM.
# This replaces ALL brittle string/ID matching for modals.
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
compressed = self._compress_xml(xml_dump)
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "NORMAL":
return SituationType.NORMAL
# ── Structural Fast-Check: Content-Creation Overlays ──
# These full-screen overlays live INSIDE Instagram's package but block
# all normal navigation. They are invisible to the foreign-app detector
# and frequently fool the LLM into thinking they are "normal" browsing.
# Detecting them structurally is O(1) and requires ZERO LLM calls.
# This is checked AFTER Qdrant to ensure that if the LLM unlearned a false positive,
# we respect the learned NORMAL state and don't infinite-loop.
creation_flow_markers = (
"quick_capture", # Camera / story capture overlay
"gallery_cancel_button", # Story gallery "Back to Home" button
@@ -459,15 +442,12 @@ class SituationalAwarenessEngine:
re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE) for marker in creation_flow_markers
):
logger.info("🧠 [SAE Perceive] Content-creation overlay detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
# ── Structural Fast-Check: Instagram-Internal Modal Overlays ──
# Surveys, rating prompts, and interstitial modals live INSIDE Instagram's
# ── Structural Fast-Check: Instagram-Internal Modal Overlays & Browsers ──
# Surveys, rating prompts, interstitial modals, and web-views live INSIDE Instagram's
# package but block normal interaction. They share a common structural
# pattern: a container resource-id containing "survey", "interstitial",
# or "nux_" (new-user-experience), plus dismiss buttons ("Not Now").
# Detecting them structurally is O(1) and eliminates LLM hallucination risk.
# pattern. Detecting them structurally is O(1) and eliminates LLM hallucination risk.
instagram_modal_markers = (
"survey_overlay_container", # "How are you enjoying Instagram?" survey
"survey_title", # Survey title text view
@@ -476,15 +456,32 @@ class SituationalAwarenessEngine:
"nux_overlay", # New-user-experience onboarding modals
"rating_prompt", # App Store rating prompt
"feedback_dialog", # Feedback collection dialogs
"ig_browser_text_title", # In-App Browser Title
"ig_browser_close_button", # In-App Browser Close Button
"ig_chrome_subsection", # In-App Browser Chrome
)
if any(
re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE)
for marker in instagram_modal_markers
):
logger.info("🧠 [SAE Perceive] Instagram modal overlay detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
logger.info("🧠 [SAE Perceive] Instagram modal/browser overlay detected structurally → OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
# ── Modal/Obstacle Detection (Autonomous LLM + Memory) ──
# We explicitly query ScreenMemoryDB. If unknown, we ask the LLM.
# This replaces ALL brittle string/ID matching for modals.
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "NORMAL":
return SituationType.NORMAL
# Fallback heuristic: detect modals by dismiss-button text patterns.
# If we see "Not Now" or "Take Survey" as button text inside Instagram, it's a modal.
# Guard: match ONLY inside short text attributes (< 40 chars) to avoid caption false positives.
@@ -605,6 +602,7 @@ class SituationalAwarenessEngine:
"- If the Situation type is obstacle_system, you MUST look for 'Deny', 'Don't allow', or 'Cancel' and click it. \n"
" NEVER click 'Allow', 'OK', or 'Confirm' on system permissions.\n"
" If no negative action button exists, action must be 'back'\n"
"- If the screen shows an In-App Browser, WebView, website, or Advertisement, it is an obstacle. Action must be 'back' or click the close button.\n"
"- If there is NO obstacle and the screen is a normal Instagram view (false positive), action must be 'false_positive'\n"
"- If nothing else works, suggest 'app_start' to force-reopen Instagram\n"
"- NEVER click 'OK'/'Confirm'/'Accept' on surveys or prompts\n"