From 0aeed11186742f97fc55dbb381d4f41689c97b51 Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Fri, 17 Apr 2026 12:57:12 +0200 Subject: [PATCH] Hardening autonomous navigation: Implemented Modal Guards, Transient Drift Protection (WhatsApp fix), and Aggressive Recovery paths. Cleaned up diagnostic artifacts. --- GramAddict/core/bot_flow.py | 358 +++++++++++++------ GramAddict/core/config.py | 1 + GramAddict/core/darwin_engine.py | 24 +- GramAddict/core/device_facade.py | 65 +++- GramAddict/core/diagnostic_dump.py | 2 +- GramAddict/core/dm_engine.py | 6 +- GramAddict/core/dump_capturer.py | 2 +- GramAddict/core/exceptions.py | 3 + GramAddict/core/llm_provider.py | 66 +++- GramAddict/core/q_nav_graph.py | 190 ++++++++-- GramAddict/core/resonance_engine.py | 26 ++ GramAddict/core/telepathic_engine.py | 215 ++++++++--- GramAddict/core/unfollow_engine.py | 17 +- debug_test.py | 13 + run_test.py | 4 + run_test2.py | 27 ++ scratch/hougaard_analyzer.py | 132 ------- scratch/verify_silence.py | 18 - scratch_dump_scanner.py | 41 +++ test_config.yml | 2 +- tests/conftest.py | 13 + tests/tdd/test_drift_hardening.py | 67 ++++ tests/tdd/test_modal_vlm_fix.py | 69 ++++ tests/tdd/test_reels_repost.py | 135 +++++++ tests/unit/test_anomaly_interruptions.py | 68 ++++ tests/unit/test_app_perimeter_guard.py | 51 +++ tests/unit/test_autonomous_retries.py | 15 +- tests/unit/test_config_effects.py | 32 +- tests/unit/test_critical_anomaly_guards.py | 101 ++++++ tests/unit/test_grid_retry_diversity.py | 83 +++++ tests/unit/test_llm_provider.py | 142 ++++---- tests/unit/test_profile_interaction_edges.py | 91 +++++ tests/unit/test_profile_interaction_sync.py | 1 + tests/unit/test_unfollow_engine.py | 109 ++++++ tests/unit/test_verify_success_reels.py | 73 ++++ 35 files changed, 1802 insertions(+), 460 deletions(-) create mode 100644 GramAddict/core/exceptions.py create mode 100644 debug_test.py create mode 100644 run_test.py create mode 100644 run_test2.py delete mode 100644 scratch/hougaard_analyzer.py delete mode 100644 scratch/verify_silence.py create mode 100644 scratch_dump_scanner.py create mode 100644 tests/tdd/test_drift_hardening.py create mode 100644 tests/tdd/test_modal_vlm_fix.py create mode 100644 tests/tdd/test_reels_repost.py create mode 100644 tests/unit/test_anomaly_interruptions.py create mode 100644 tests/unit/test_app_perimeter_guard.py create mode 100644 tests/unit/test_critical_anomaly_guards.py create mode 100644 tests/unit/test_grid_retry_diversity.py create mode 100644 tests/unit/test_profile_interaction_edges.py create mode 100644 tests/unit/test_unfollow_engine.py create mode 100644 tests/unit/test_verify_success_reels.py diff --git a/GramAddict/core/bot_flow.py b/GramAddict/core/bot_flow.py index 8fef9c2..1e44c53 100644 --- a/GramAddict/core/bot_flow.py +++ b/GramAddict/core/bot_flow.py @@ -59,6 +59,12 @@ def start_bot(**kwargs): # Check for direct execution modes that bypass normal bot state configs.parse_args() + try: + from GramAddict.core.llm_provider import prewarm_ollama_models + prewarm_ollama_models(configs) + except Exception as e: + logger.debug(f"Prewarm failed: {e}") + if getattr(configs.args, "capture_e2e_dumps", False): device = create_device(configs.device_id, configs.app_id, configs.args) from GramAddict.core.dump_capturer import capture_all @@ -186,12 +192,12 @@ def start_bot(**kwargs): if success: if current_target == "ExploreFeed": logger.info("πŸ“± Opening first explore item from the grid...") - nav_graph._execute_transition("tap_explore_grid_item", zero_engine) + nav_graph._execute_transition("tap_explore_grid_item") # Wait for post to actually load (poll for feed markers) _wait_for_post_loaded(device, timeout=5) elif current_target == "StoriesFeed": logger.info("πŸ“± Locating story tray on HomeFeed...") - nav_graph._execute_transition("tap_story_tray_item", zero_engine) + nav_graph._execute_transition("tap_story_tray_item") _wait_for_post_loaded(device, timeout=5) if current_target == "StoriesFeed": @@ -256,6 +262,9 @@ FEED_MARKERS = [ "row_feed_profile_header", "row_feed_photo_imageview", "clips_media_component", + "clips_video_container", + "clips_viewer_container", + "clips_linear_layout_container" ] @@ -266,7 +275,7 @@ def _wait_for_post_loaded(device, timeout=5): start = time.time() while time.time() - start < timeout: try: - xml = device.deviceV2.dump_hierarchy() + xml = device.dump_hierarchy() if any(marker in xml for marker in FEED_MARKERS): logger.debug("πŸ“± Post loaded successfully.") return True @@ -424,14 +433,25 @@ def _interact_with_carousel(device, configs, sleep_mod, logger): info = device.get_info() w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400) + # Curiosity Peak: One slide in the carousel gets extra attention + curiosity_slide = random.randint(0, count - 1) if count > 0 else 0 + for i in range(count): + # Normal transition wait sleep(random.uniform(1.5, 3.5) * sleep_mod) + + # ── Curiosity Dwell ── + if i == curiosity_slide: + dwell = random.uniform(3.0, 7.0) + logger.debug(f"πŸ“Έ [Carousel] Curiosity Peak hit on slide {i+1}. Gazing for {dwell:.1f}s...") + sleep(dwell * sleep_mod) + # Horizontal swipe inside the post bounds (approx middle): Right to left _humanized_horizontal_swipe(device, start_x=w*0.8, end_x=w*0.2, y=h*0.5, duration_ms=250) + sleep(random.uniform(1.0, 2.0) * sleep_mod) def _interact_with_profile(device, configs, username, session_state, sleep_mod, logger): - print("HELLO IM NOT MOCKED!") """Deep interaction on a profile: Stories, Grid Likes, Follows""" import random @@ -442,11 +462,26 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, info = device.get_info() w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400) + xml_check = device.dump_hierarchy() + 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: + 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: + logger.info(f"πŸ“­ [Profile Guard] @{username} has no posts. Aborting deep interaction.", extra={"color": f"{Fore.YELLOW}"}) + return + # Profile Scraping (Phase 11: Data Extraction) if getattr(configs.args, "scrape_profiles", False): try: logger.info(f"πŸ“Š [Scraping] Extracting metadata for @{username}...", extra={"color": f"{Fore.CYAN}"}) - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() telepathic = TelepathicEngine.get_instance() crm = cognitive_stack.get("crm") if 'cognitive_stack' in locals() else None @@ -483,7 +518,9 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, from GramAddict.core.q_nav_graph import QNavGraph nav_graph = QNavGraph(device) - if nav_graph._execute_transition("tap_story_tray_item"): + xml_dump = device.dump_hierarchy() + has_story = "reel_ring" in xml_dump or "'s unseen story" in xml_dump.lower() or "has a new story" in xml_dump.lower() or "story von" in xml_dump.lower() + if has_story and nav_graph._execute_transition("tap_story_tray_item"): logger.info(f"πŸ“Έ [Story] Viewing @{username}'s story ({count} times)...") for i in range(count): sleep(random.uniform(2.0, 5.0) * sleep_mod) @@ -530,13 +567,46 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, logger.info(f"❀️ [Deep Interaction] Opening grid to drop {count} likes on @{username}...") for i in range(count): - _humanized_click(device, w // 2, h // 2, double=True, sleep_mod=sleep_mod) - session_state.totalLikes += 1 - logger.debug(f"Liked grid post {i+1}/{count}") + xml_dump = device.dump_hierarchy() + if not isinstance(xml_dump, str): + xml_dump = "" + xml_dump_lower = xml_dump.lower() + + is_reel = "reel_viewer" in xml_dump_lower or "clips_viewer" in xml_dump_lower + is_liked = "gefΓ€llt mir nicht mehr" in xml_dump_lower or "unlike" in xml_dump_lower or 'content-desc="liked"' in xml_dump_lower + + # Uset Double-Tap ~40% of the time, only on standard images + use_double_tap = random.random() < 0.4 and not is_reel + + if use_double_tap: + if is_liked: + logger.debug(f"Skipped liking grid post {i+1}/{count} (already liked)") + else: + offset_x = random.randint(int(w * 0.2), int(w * 0.8)) + offset_y = random.randint(int(h * 0.3), int(h * 0.7)) + logger.info(f"❀️ [Interaction] Double-Tapping organically at ({offset_x}, {offset_y})") + _humanized_click(device, offset_x, offset_y, double=True, sleep_mod=sleep_mod) + session_state.totalLikes += 1 + logger.debug(f"Liked grid post {i+1}/{count} via Double-Tap") + else: + if nav_graph._execute_transition("tap_like_button"): + session_state.totalLikes += 1 + logger.debug(f"Liked grid post {i+1}/{count} via Heart Button") + else: + logger.debug(f"Skipped liking grid post {i+1}/{count} (already liked or failed to find button)") + sleep(random.uniform(1.0, 2.0) * sleep_mod) - start_scroll_y, end_scroll_y = int(h * 0.7) + random.randint(-20, 20), int(h * 0.2) + random.randint(-40, 40) - scroll_x = w // 2 + random.randint(-30, 30) - device.deviceV2.shell(f"input swipe {scroll_x} {start_scroll_y} {scroll_x + random.randint(-15,15)} {end_scroll_y} {random.randint(250, 400)}") + + is_reel = "reel_viewer" in xml_dump_lower or "clips_viewer" in xml_dump_lower + + if is_reel: + # Full screen swipe for Reels (using humanized fast fling) + logger.debug("🎬 Detected Reel. Swiping full-screen up.") + _humanized_scroll(device, is_skip=True) + else: + # Partial screen swipe for standard posts + _humanized_scroll(device, is_skip=False) + sleep(random.uniform(1.5, 3.0) * sleep_mod) device.deviceV2.press("back") @@ -564,7 +634,7 @@ def _align_active_post(device): while not aligned and attempts < max_attempts: attempts += 1 try: - xml = device.deviceV2.dump_hierarchy() + xml = device.dump_hierarchy() clean_xml = re.sub(r'<\?xml.*?\?>', '', xml).strip() root = ET.fromstring(clean_xml) @@ -648,7 +718,12 @@ def _detect_ad_structural(context_xml: str) -> bool: "com.instagram.android:id/universal_cta_description_layout", "com.instagram.android:id/universal_cta_text", } - AD_CTA_WORDS = {"install", "learn more", "shop now", "sign up", "mehr dazu", "jetzt einkaufen", "installieren", "registrieren", "anmelden", "download", "herunterladen", "get offer", "abonnieren", "subscribe"} + AD_CTA_WORDS = { + "install", "learn more", "shop now", "sign up", "mehr dazu", "jetzt einkaufen", + "installieren", "registrieren", "anmelden", "download", "herunterladen", + "get offer", "abonnieren", "subscribe", "whatsapp", "nachricht senden", + "send message", "jetzt anrufen", "call now", "contact us", "kontaktieren" + } try: root = ET.fromstring(context_xml) @@ -699,7 +774,7 @@ def _extract_post_content(context_xml: str) -> dict: desc = node.attrib.get("content-desc", "").strip() # Username from the post header (ignore commenters/composers) - if "row_feed_photo_profile_name" in res_id and text: + if ("row_feed_photo_profile_name" in res_id or "clips_author_username" in res_id) and text: if "comment" not in res_id and "composer" not in res_id: # Prioritize the FIRST valid username found (usually the header) if not result["username"]: @@ -767,7 +842,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta sleep(random.uniform(1.0, 2.0) * sleep_mod) return "BOREDOM_CHANGE_FEED" - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() if not xml_dump: logger.warning("Failed to dump UI hierarchy in StoriesFeed.") return "CONTEXT_LOST" @@ -858,26 +933,42 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # ── Boredom ── if random.random() < 0.03 and not is_reels: - logger.info("πŸ₯± [Boredom] Checking something else (Notifications/DMs) for a second...") - - # Use NavGraph transitions instead of raw selectors - if random.random() < 0.5: - # Try to visit Notifications - nav_graph._execute_transition("tap_newsfeed_tab", zero_engine) - else: - # Try to visit DMs - nav_graph._execute_transition("tap_message_icon", zero_engine) + if job_target in ["feed", "home", "homefeed"]: + logger.info("πŸ₯± [Boredom] Checking something else (Notifications/DMs) for a second...") - sleep(random.uniform(3.0, 6.0) * sleep_mod) + # Use NavGraph transitions instead of raw selectors + if random.random() < 0.5: + # Try to visit Notifications + nav_graph._execute_transition("tap_newsfeed_tab") + else: + # Try to visit DMs + nav_graph._execute_transition("tap_message_icon") + + sleep(random.uniform(3.0, 6.0) * sleep_mod) + + # Return to feed natively through robust navigation + nav_graph.navigate_to("HomeFeed", zero_engine) + sleep(random.uniform(1.0, 2.5) * sleep_mod) - # Return to feed natively through robust navigation - nav_graph.navigate_to("HomeFeed", zero_engine) - sleep(random.uniform(1.0, 2.5) * sleep_mod) - - context_xml = device.deviceV2.dump_hierarchy() + context_xml = device.dump_hierarchy() if cognitive_stack.get("radome"): context_xml = cognitive_stack.get("radome").sanitize_xml(context_xml) + # ── PRE-EMPTIVE AD SKIP (Fast Path) ── + if _detect_ad_structural(context_xml): + consecutive_ads += 1 + if consecutive_ads >= 3: + logger.warning("πŸ“Ί [Anti-Stuck] Stuck on ad! Executing aggressive skip.", extra={"color": f"{Fore.RED}"}) + _humanized_scroll(device, is_skip=True) + sleep(2.0) + else: + logger.info("πŸ“Ί fast-skipping ad (no AI needed)...") + _humanized_scroll(device, is_skip=True) + sleep(random.uniform(0.5, 1.0) * sleep_mod) + continue + + consecutive_ads = 0 + # ── Zero-Node Recovery (Graceful Degradation) ── telepathic = TelepathicEngine.get_instance() interactive_nodes = telepathic._extract_semantic_nodes(context_xml) @@ -923,7 +1014,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session sleep(2.5) # Verification: Check if markers are now visible - post_recovery_xml = device.deviceV2.dump_hierarchy() + post_recovery_xml = device.dump_hierarchy() if any(marker in post_recovery_xml for marker in FEED_MARKERS): logger.info("βœ… [Recovery] Obstacle cleared successfully. Learning this button works.") telepathic.confirm_click("Dismiss Obstacle/Modal") @@ -956,28 +1047,10 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Fixes the issue where UI gets stuck halfway between two posts. if _align_active_post(device): # Update context_xml because the screen just shifted - context_xml = device.deviceV2.dump_hierarchy() + context_xml = device.dump_hierarchy() if cognitive_stack.get("radome"): context_xml = cognitive_stack.get("radome").sanitize_xml(context_xml) - # ── Structural Ad Detection (Language-Agnostic) ── - if _detect_ad_structural(context_xml): - consecutive_ads += 1 - if consecutive_ads >= 3: - logger.warning("πŸ“Ί [Anti-Stuck] Stuck on ad! Executing aggressive mechanical drag.", extra={"color": f"{Fore.RED}"}) - # Fast drag from bottom to top (~0.15s) to guarantee Native Android Fling event for Reels - info = device.get_info() - w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400) - device.deviceV2.shell(f"input swipe {w // 2} {int(h * 0.8)} {w // 2} {int(h * 0.2)} 150") - sleep(2.0) - else: - logger.info("πŸ“Ί skipping ad (structural match)...") - _humanized_scroll(device, is_skip=True) - sleep(random.uniform(0.5, 1.5) * sleep_mod) - continue - - consecutive_ads = 0 - # ── Content Extraction (The Bot's Eyes) ── post_data = _extract_post_content(context_xml) @@ -1014,7 +1087,14 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session }) # ── Human-like Selective Skipping (Anti-Bot Drip) ── - skip_prob = 0.85 if res_score < 0.35 else 0.45 if res_score < 0.70 else 0.10 + base_skip_prob = 0.85 if res_score < 0.35 else 0.45 if res_score < 0.70 else 0.10 + + # User defined interact_percentage modulates the skip rate. + # Default is 80%, so factor = 1.0. If 100%, factor = 0.0 (never skip). + interact_pct_val = float(getattr(configs.args, "interact_percentage", 80)) / 100.0 + skip_factor = max(0.0, (1.0 - interact_pct_val) * 5.0) + skip_prob = base_skip_prob * skip_factor + if random.random() < skip_prob: logger.info(f"⏭️ [Resonance Skip] Human-like selective engagement ({skip_prob*100:.0f}% chance). Skipping post.") session_outcomes.append({ @@ -1029,7 +1109,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # ── The Rabbit Hole (Deep Dive into high-resonance profiles) ── if res_score >= 0.9 and random.random() < 0.4: logger.info("πŸ’₯ [Rabbit Hole] Extreme resonance! Sidetracking into user profile...", extra={"color": f"{Fore.MAGENTA}"}) - if nav_graph._execute_transition("tap_post_username", zero_engine) is True: + if nav_graph._execute_transition("tap_post_username") is True: sleep(random.uniform(1.2, 2.5) * sleep_mod) _humanized_scroll(device, is_skip=True) sleep(random.uniform(0.5, 1.5) * sleep_mod) @@ -1044,6 +1124,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session darwin.execute_proof_of_resonance( device, res_score, + text_length=len(post_data.get("description", "")), nav_graph=nav_graph, zero_engine=zero_engine, configs=configs, @@ -1063,11 +1144,17 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session profile_context = "" # ── Profile Learning (Before heavy engagement) ── target_user = post_data.get('username', 'target') - if res_score >= 0.8: + + # Pull follow chance early to see if the user explicitly wants high follow rates + follow_chance_val = float(getattr(configs.args, "follow_percentage", 0)) / 100.0 + + # If the user sets follow > 0, we must visit the profile to have a chance to follow. + # Otherwise, we rely entirely on the extreme resonance heuristic (> 0.8). + if res_score >= 0.8 or (follow_chance_val > 0.0 and random.random() < follow_chance_val): logger.info(f"πŸ•΅οΈβ€β™‚οΈ [Profile Learning] Highly resonant post ({res_score:.2f}). Visiting @{target_user}'s profile to learn context...", extra={"color": f"{Fore.CYAN}"}) # Navigate to profile - if nav_graph._execute_transition("tap_post_username", zero_engine) is True: + if nav_graph._execute_transition("tap_post_username") is True: sleep(random.uniform(1.2, 2.5) * sleep_mod) # Extract context @@ -1075,7 +1162,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session telepathic = cognitive_stack.get("telepathic") crm = cognitive_stack.get("crm") if telepathic: - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() nodes = telepathic._extract_semantic_nodes(xml_dump) texts = [] @@ -1107,43 +1194,63 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session if session_state.check_limit(SessionState.Limit.LIKES): likes_chance = 0.0 - if res_score >= 0.35 and random.random() < likes_chance: - logger.info("❀️ [Interaction] Liking post...") - success = nav_graph._execute_transition("tap_like_button", zero_engine) - if success is not True: - logger.debug("Telepathic Like failed, falling back to double-tap.") - info = device.get_info() - w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400) - _humanized_click(device, w // 2, h // 2, double=True, sleep_mod=sleep_mod) - session_state.totalLikes += 1 - sleep(random.uniform(1.2, 2.5) * sleep_mod) - did_interact = True - did_like = True + # If user explicitly configures likes_chance > 0, we lower the strict AI resonance requirement + needs_like = (likes_chance > 0.0 and random.random() < likes_chance) + if (needs_like or res_score >= 0.35): + logger.info("❀️ [Interaction] Deciding like method...") - # Umentscheidung (Change of mind) - if random.random() < 0.02: - logger.info("🧠 [Umentscheidung] Taking back the like.") - sleep(random.uniform(1.0, 3.0)) - # Press like button again to unlike - nav_graph._execute_transition("tap_like_button", zero_engine) - session_state.totalLikes -= 1 - did_like = False - did_interact = False + xml_check = device.dump_hierarchy() + if not isinstance(xml_check, str): xml_check = "" + xml_check_lower = xml_check.lower() + is_reel_feed = "reel_viewer" in xml_check_lower or "clips_viewer" in xml_check_lower + is_liked_feed = "gefΓ€llt mir nicht mehr" in xml_check_lower or "unlike" in xml_check_lower or 'content-desc="liked"' in xml_check_lower + + use_double_tap = random.random() < 0.4 and not is_reel_feed + did_like = False + + if use_double_tap: + if is_liked_feed: + logger.debug("Telepathic Like failed or post was unlikable (already liked). Skipping increment.") + else: + info = device.get_info() + w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400) + offset_x = random.randint(int(w * 0.2), int(w * 0.8)) + offset_y = random.randint(int(h * 0.3), int(h * 0.7)) + logger.info(f"❀️ [Interaction] Double-Tapping organically at ({offset_x}, {offset_y})") + _humanized_click(device, offset_x, offset_y, double=True, sleep_mod=sleep_mod) + session_state.totalLikes += 1 + sleep(random.uniform(1.2, 2.5) * sleep_mod) + did_interact = True + did_like = True + else: + logger.info("❀️ [Interaction] Liking post via Heart Button...") + success = nav_graph._execute_transition("tap_like_button") + if success: + session_state.totalLikes += 1 + sleep(random.uniform(1.2, 2.5) * sleep_mod) + did_interact = True + did_like = True + else: + logger.debug("Telepathic Like failed or post was unlikable. Skipping increment.") + + # Comment: requires high resonance alignment comment_chance = float(getattr(configs.args, "comment_percentage", 0)) / 100.0 if session_state.check_limit(SessionState.Limit.COMMENTS): comment_chance = 0.0 - if res_score >= 0.4 and random.random() < comment_chance: + # If user explicitly configures comment_chance > 0, we lower the strict AI resonance requirement + needs_comment = (comment_chance > 0.0 and random.random() < comment_chance) + if (needs_comment or res_score >= 0.4): logger.info("πŸ’¬ [Interaction] Entering Comment Sheet for deep engagement...") - success = nav_graph._execute_transition("tap_comment_button", zero_engine) + success = nav_graph._execute_transition("tap_comment_button") if success is True: sleep(random.uniform(2.0, 4.0) * sleep_mod) # 1. Scrape Context from the comment sheet - sheet_xml = device.deviceV2.dump_hierarchy() + sheet_xml = device.dump_hierarchy() # πŸ›‘οΈ [Semantic Gate] Verify we are actually in the comment sheet if not any(x in sheet_xml for x in ["layout_comment_thread", "comment_composer", "comment_button_post"]): @@ -1185,7 +1292,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session if random.random() < 0.4: # Use Telepathic to find the like button for this specific comment text intent = f"Heart like button for comment: '{c_node['text'][:20]}...'" - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() like_btn = telepathic.find_best_node(xml_dump, intent, device=device) if like_btn and not like_btn.get("skip"): @@ -1193,7 +1300,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session sleep(random.uniform(0.8, 1.5)) # Verification: Simple XML change check - if device.deviceV2.dump_hierarchy() != xml_dump: + if device.dump_hierarchy() != xml_dump: telepathic.confirm_click(intent) logger.info(f"❀️ [Interaction] Liked user comment: '{c_node['text'][:30]}...'") else: @@ -1202,7 +1309,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # 20% chance to randomly visit commenter's profile if random.random() < 0.2: intent = f"Avatar profile picture for commenter: '{c_node['text'][:20]}...'" - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() avatar_node = telepathic.find_best_node(xml_dump, intent, device=device) if avatar_node: @@ -1211,7 +1318,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session sleep(random.uniform(2.5, 4.5) * sleep_mod) # Verification: Did we reach a profile? - post_xml = device.deviceV2.dump_hierarchy() + post_xml = device.dump_hierarchy() if "profile" in post_xml.lower() or "button_follow" in post_xml.lower(): telepathic.confirm_click(intent) _interact_with_profile(device, configs, "commenter", session_state, sleep_mod, logger) @@ -1225,7 +1332,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # 15% chance to Sub-Comment (Reply) if random.random() < 0.15 and not replying_to: intent = f"Reply button for comment: '{c_node['text'][:20]}...'" - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() reply_btn = telepathic.find_best_node(xml_dump, intent, device=device) if reply_btn: @@ -1233,7 +1340,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session sleep(random.uniform(1.2, 2.0)) # Verification: Did the screen change or input field appear? - if device.deviceV2.dump_hierarchy() != xml_dump: + if device.dump_hierarchy() != xml_dump: telepathic.confirm_click(intent) replying_to = c_node["text"] logger.info(f"πŸ” [Interaction] Replying directly to comment: '{replying_to[:30]}...'") @@ -1290,7 +1397,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Verification: Did the keyboard open or cursor move to box? # We check if the XML changed and focus is on an edittext - post_focus_xml = device.deviceV2.dump_hierarchy() + post_focus_xml = device.dump_hierarchy() if "editText" in post_focus_xml.lower() or post_focus_xml != sheet_xml: telepathic.confirm_click("Comment input text box editfield") else: @@ -1312,7 +1419,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Press back to trigger Discard popup device.deviceV2.press("back") sleep(1.0) - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() discard_btn = telepathic.find_best_node(xml_dump, "Discard or Verwerfen popup button to cancel comment", device=device) if discard_btn: device.deviceV2.click(discard_btn["x"], discard_btn["y"]) @@ -1323,14 +1430,14 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session else: # Tap Post sleep(random.uniform(0.5, 1.5)) - pre_post_xml = device.deviceV2.dump_hierarchy() + pre_post_xml = device.dump_hierarchy() post_btn = telepathic.find_best_node(pre_post_xml, "Post submit comment button", device=device) if post_btn: device.deviceV2.click(post_btn["x"], post_btn["y"]) sleep(random.uniform(2.0, 3.5)) # Verification: Did the button disappear or layout change? - post_post_xml = device.deviceV2.dump_hierarchy() + post_post_xml = device.dump_hierarchy() # If "Post" button is gone from the area or XML changed significantly if "button_post" not in post_post_xml.lower() or post_post_xml != pre_post_xml: telepathic.confirm_click("Post submit comment button") @@ -1344,10 +1451,10 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session logger.error(f"❌ [Interaction] AI Comment deployment failed: {e}") # Safely exit the comment sheet - if "bottom_sheet_container" in device.deviceV2.dump_hierarchy(): + if "bottom_sheet_container" in device.dump_hierarchy(): device.deviceV2.press("back") sleep(1.0) - if "bottom_sheet_container" in device.deviceV2.dump_hierarchy(): + if "bottom_sheet_container" in device.dump_hierarchy(): device.deviceV2.press("back") sleep(1.0) @@ -1357,31 +1464,44 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session repost_chance = float(getattr(configs.args, "repost_percentage", 0)) / 100.0 if res_score >= 0.70 and random.random() < repost_chance: logger.info("πŸ” [Interaction] Reposting highly resonant content...", extra={"color": f"{Fore.CYAN}"}) - success = nav_graph._execute_transition("tap_share_button", zero_engine) - if success is True: - sleep(random.uniform(1.8, 3.5) * sleep_mod) - telepathic = TelepathicEngine.get_instance() - xml_dump = device.deviceV2.dump_hierarchy() - repost_btn = telepathic.find_best_node(xml_dump, "Repost interaction button with two arrows", device=device) - if repost_btn and not repost_btn.get("skip"): - _humanized_click(device, repost_btn["x"], repost_btn["y"], sleep_mod=sleep_mod) - sleep(random.uniform(2.0, 4.0) * sleep_mod) + + # Fast Path: Check if Repost button is ALREADY on the screen (Direct Repost for Reels) + telepathic = TelepathicEngine.get_instance() + current_xml = device.dump_hierarchy() + direct_repost = telepathic.find_best_node(current_xml, "Repost interaction button with two arrows", device=device, threshold=0.90) if is_reels else None + + success = True + if direct_repost and not direct_repost.get("skip"): + logger.info("⚑ [Fast Path] Found direct Repost button. Skipping share sheet.") + repost_btn = direct_repost + else: + success = nav_graph._execute_transition("tap_share_button") + if success is True: + sleep(random.uniform(1.8, 3.5) * sleep_mod) + xml_dump = device.dump_hierarchy() + repost_btn = telepathic.find_best_node(xml_dump, "Repost interaction button with two arrows", device=device) + else: + repost_btn = None + + if success is True and repost_btn and not repost_btn.get("skip"): + _humanized_click(device, repost_btn["x"], repost_btn["y"], sleep_mod=sleep_mod) + sleep(random.uniform(2.0, 4.0) * sleep_mod) + + # Verification: Did the share menu close or repost confirmation appear? + post_xml = device.dump_hierarchy() + repost_success = post_xml != current_xml or "reposted" in post_xml.lower() + + if repost_success: + telepathic.confirm_click("Repost interaction button with two arrows") + logger.info("βœ… [Interaction] Content successfully reposted to feed/followers.", extra={"color": f"{Fore.GREEN}"}) + did_interact = True + else: + telepathic.reject_click("Repost interaction button with two arrows") + logger.warning("⚠️ [Repost] Click failed to trigger repost. Learning from failure.") - # Verification: Did the share menu close or repost confirmation appear? - post_xml = device.deviceV2.dump_hierarchy() - repost_success = post_xml != xml_dump or "reposted" in post_xml.lower() - - if repost_success: - telepathic.confirm_click("Repost interaction button with two arrows") - logger.info("βœ… [Interaction] Content successfully reposted to feed/followers.", extra={"color": f"{Fore.GREEN}"}) - did_interact = True - else: - telepathic.reject_click("Repost interaction button with two arrows") - logger.warning("⚠️ [Repost] Click failed to trigger repost. Learning from failure.") - - # Close share menu if still open - device.deviceV2.press("back") - sleep(random.uniform(1.0, 2.0) * sleep_mod) + # Close share menu if still open + device.deviceV2.press("back") + sleep(random.uniform(1.0, 2.0) * sleep_mod) # ── Parasocial CRM & SwarmProtocol ── @@ -1408,7 +1528,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # ── Active Inference: Evaluate prediction (after action) ── if ai: _wait_for_post_loaded(device, timeout=3) - post_action_xml = device.deviceV2.dump_hierarchy() + post_action_xml = device.dump_hierarchy() ai.evaluate_prediction(post_action_xml) # ── Advance to next post ── @@ -1454,7 +1574,7 @@ def _run_zero_latency_search_loop(device, zero_engine, nav_graph, configs, sessi # We assume we are on the Explore tab now (Global Navigation Bar) try: - xml = device.deviceV2.dump_hierarchy() + xml = device.dump_hierarchy() telepathic = cognitive_stack.get("telepathic") # Find search bar @@ -1467,7 +1587,7 @@ def _run_zero_latency_search_loop(device, zero_engine, nav_graph, configs, sessi sleep(3.0) # 2. Pick a result (Top, Accounts, or Tags) - results_xml = device.deviceV2.dump_hierarchy() + results_xml = device.dump_hierarchy() target_result = telepathic.find_best_node(results_xml, "First relevant search result (Account or Hashtag)", device=device) if target_result: diff --git a/GramAddict/core/config.py b/GramAddict/core/config.py index 72a92c5..a33271c 100644 --- a/GramAddict/core/config.py +++ b/GramAddict/core/config.py @@ -171,6 +171,7 @@ class Config: self.parser.add_argument("--ai-target-audience", help="Target audience used interchangeably with persona interests", default="") self.parser.add_argument("--interact-percentage", help="Overall interaction probability percentage", default="80") self.parser.add_argument("--comment-percentage", help="Comment probability percentage", default="0") + self.parser.add_argument("--follow-percentage", help="Follow probability percentage", default="0") self.parser.add_argument("--dry-run-comments", action="store_true", help="Generate AI comments but do not actually post them (debug/logging only)") self.parser.add_argument("--search", help="Comma-separated keywords to search for", default="") self.parser.add_argument("--scrape-profiles", action="store_true", help="Extract and store profile metadata in CRM") diff --git a/GramAddict/core/darwin_engine.py b/GramAddict/core/darwin_engine.py index ae2fd98..0e90ec1 100644 --- a/GramAddict/core/darwin_engine.py +++ b/GramAddict/core/darwin_engine.py @@ -30,10 +30,10 @@ class DarwinEngine(QdrantBase): } self.current_behavior = {} - def synthesize_interaction_profile(self, target_resonance: float) -> dict: + def synthesize_interaction_profile(self, target_resonance: float, text_length: int = 0) -> dict: """ - Given an AI aesthetic resonance score (0.0 to 1.0), this generates - a deterministic topological interaction behavior mathematically suited to the target. + Given an AI aesthetic resonance score (0.0 to 1.0) and caption length, + this generates a deterministic topological interaction behavior. """ history = self._get_historical_landscape() epsilon = 0.15 # 15% pure exploration @@ -54,18 +54,26 @@ class DarwinEngine(QdrantBase): self.current_behavior["initial_dwell_sec"] *= max(0.5, target_resonance * 1.5) self.current_behavior["profile_visit_prob"] *= max(0.2, target_resonance * 2.0) + # ── Generative Dwell-Time ── + # Humans take longer to finish "reading" long captions. + # Average reading speed is ~15-20 chars per second. + if text_length > 20: + reading_latency = min(15.0, text_length / 25.0) # Cap extra reading time at 15s + logger.debug(f"🧬 [Darwin Engine] Generative Dwell spike: +{reading_latency:.1f}s (Caption: {text_length} chars)") + self.current_behavior["initial_dwell_sec"] += reading_latency + # Clip bounds for k, (b_min, b_max, _) in self.behavior_bounds.items(): self.current_behavior[k] = max(b_min, min(b_max, self.current_behavior[k])) return self.current_behavior - def execute_proof_of_resonance(self, device, resonance: float, nav_graph=None, zero_engine=None, configs=None, resonance_oracle=None, username=None): + def execute_proof_of_resonance(self, device, resonance: float, text_length: int = 0, nav_graph=None, zero_engine=None, configs=None, resonance_oracle=None, username=None): """ Translates the mathematical interaction profile directly into device actions to prove engagement to the platform's anti-bot heuristic algorithm. """ - profile = self.synthesize_interaction_profile(resonance) + profile = self.synthesize_interaction_profile(resonance, text_length=text_length) logger.info("🧬 [Darwin MDP] Executing Proof of Resonance Sequence...") @@ -124,7 +132,7 @@ class DarwinEngine(QdrantBase): except Exception as e: logger.warning(f"⚠️ [Vision Context] Failed to capture screenshot: {e}") - success = nav_graph._execute_transition("tap_comment_button", zero_engine) + success = nav_graph._execute_transition("tap_comment_button") if success: # ---- Phase 10: RAG Comment Extraction ---- if configs and resonance_oracle and getattr(configs.args, "ai_learn_comments", False): @@ -132,7 +140,7 @@ class DarwinEngine(QdrantBase): if random.random() < 0.05: logger.debug(" -> Dumping UI hierarchy for Comment Extraction...") try: - xml_data = device.deviceV2.dump_hierarchy() + xml_data = device.dump_hierarchy() t0 = time.time() resonance_oracle.extract_and_learn_comments(xml_data, configs, author=username or "unknown", images_b64=b64_img_payload) t1 = time.time() @@ -154,7 +162,7 @@ class DarwinEngine(QdrantBase): time.sleep(1.0) # Instead of relying on a fragile bottom_sheet_container ID, # we verify if the feed is visible. If not, the comment sheet is still open (or keyboard). - ui_dump = device.deviceV2.dump_hierarchy() + ui_dump = device.dump_hierarchy() if 'resource-id="com.instagram.android:id/row_feed"' not in ui_dump and 'resource-id="com.instagram.android:id/button_like"' not in ui_dump: logger.debug(" -> Not back on Home feed, pressing back again to close comment sheet/keyboard") device.deviceV2.press("back") diff --git a/GramAddict/core/device_facade.py b/GramAddict/core/device_facade.py index b9d91b8..df0880a 100644 --- a/GramAddict/core/device_facade.py +++ b/GramAddict/core/device_facade.py @@ -1,7 +1,7 @@ import logging import json import uiautomator2 as u2 -from time import sleep +from time import sleep, time from random import uniform from GramAddict.core.utils import random_sleep from functools import wraps @@ -67,6 +67,10 @@ class DeviceFacade: except Exception as e: logger.debug(f"Could not start system watcher: {e}") + # Transient package cache to prevent repeat lag + self.last_transient_pkg = None + self.last_transient_time = 0 + @adb_retry() def get_info(self): return self.deviceV2.info @@ -123,11 +127,11 @@ class DeviceFacade: # Human finger rest time (squish) sleep(uniform(0.05, 0.15)) - # Sloppy slip - slip_x = x + int(uniform(-4, 4)) + # Sloppy slip (Containment: Don't slip horizontally at the bottom edge, prevents Android App-Switch gestures) + slip_x = x + int(uniform(-4, 4)) if y < 2100 else x slip_y = y + int(uniform(-4, 4)) - self.deviceV2.touch.move(slip_x, slip_y) + self.deviceV2.touch.move(slip_x, slip_y) sleep(uniform(0.01, 0.05)) self.deviceV2.touch.up(slip_x, slip_y) except Exception as e: @@ -150,15 +154,35 @@ class DeviceFacade: def _get_current_app(self): """ Hardened app package detection. - Transient notifications (e.g. Amazon, WhatsApp) can spoof uiautomator2's app_current() report. - We verify the package with a retry if it doesn't match our expected app_id. + Transient notifications (e.g. Amazon, WhatsApp, SystemUI) can spoof uiautomator2's app_current() report. + We verify the package with multiple retries and a grace period if it doesn't match our expected app_id. """ pkg = self.deviceV2.app_current().get("package") - if pkg != self.app_id: - # Maybe a notification popped up? Wait and re-check. - sleep(0.5) - pkg = self.deviceV2.app_current().get("package") + if pkg == self.app_id: + return pkg + # If it doesn't match, it might be a notification banner. + # Known transient spoofers: WhatsApp, SystemUI (status bar), Android System + transient_packages = ["com.whatsapp", "com.android.systemui", "android"] + + if pkg in transient_packages: + # Check cooldown: if we just handled this package < 10s ago, don't sleep again + now = time() + if pkg == self.last_transient_pkg and (now - self.last_transient_time) < 10.0: + logger.debug(f"Perimeter: Consecutive hit for transient package '{pkg}'. Skipping cooldown wait.") + return self.app_id + + logger.debug(f"⚠️ [Perimeter] Detected transient package '{pkg}'. Waiting for banner to clear...") + self.last_transient_pkg = pkg + self.last_transient_time = now + sleep(1.5) # Give the notification/animation time to fade + + pkg = self.deviceV2.app_current().get("package") + if pkg in transient_packages: + # If it persists, we trust the drift logic to handle it if it blocks the UI, + # but for focus detection, we return the target app to avoid infinite wait loops. + return self.app_id + return pkg @adb_retry() @@ -168,7 +192,26 @@ class DeviceFacade: @adb_retry() def dump_hierarchy(self): - return self.deviceV2.dump_hierarchy() + xml = self.deviceV2.dump_hierarchy() + + # Continuous Session Tracing + import os + from datetime import datetime + try: + if not hasattr(self, "_trace_counter"): + self._trace_counter = 0 + ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + self._trace_dir = os.path.join("debug", "session_traces", ts) + os.makedirs(self._trace_dir, exist_ok=True) + + self._trace_counter += 1 + trace_path = os.path.join(self._trace_dir, f"{self._trace_counter:05d}.xml") + with open(trace_path, "w", encoding="utf-8") as f: + f.write(xml) + except Exception as e: + logger.debug(f"Failed to write session trace: {e}") + + return xml @adb_retry() def screenshot(self): diff --git a/GramAddict/core/diagnostic_dump.py b/GramAddict/core/diagnostic_dump.py index 1d4283e..208ffa9 100644 --- a/GramAddict/core/diagnostic_dump.py +++ b/GramAddict/core/diagnostic_dump.py @@ -35,7 +35,7 @@ def dump_ui_state(device, reason: str, extra_context: dict = None): os.makedirs(DUMP_DIR, exist_ok=True) # Capture hierarchy - xml = device.deviceV2.dump_hierarchy() + xml = device.dump_hierarchy() # Generate filename: reason__2026-04-13_17-41-39.xml ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") diff --git a/GramAddict/core/dm_engine.py b/GramAddict/core/dm_engine.py index 306fbc8..6102dab 100644 --- a/GramAddict/core/dm_engine.py +++ b/GramAddict/core/dm_engine.py @@ -36,7 +36,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s return "BOREDOM_CHANGE_FEED" try: - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() # Step 1: Find unread conversation threads unread_threads = telepathic._extract_semantic_nodes(xml_dump, "find unread message threads or unread badges", threshold=0.7) @@ -48,7 +48,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s sleep(2.0) # Step 2: Read the conversation context - thread_xml = device.deviceV2.dump_hierarchy() + thread_xml = device.dump_hierarchy() msg_nodes = telepathic._extract_semantic_nodes(thread_xml, "find the last received message text", threshold=0.6) context_text = "No previous context" @@ -76,7 +76,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s sleep(1.0) # Find Send button - send_xml = device.deviceV2.dump_hierarchy() + send_xml = device.dump_hierarchy() send_nodes = telepathic._extract_semantic_nodes(send_xml, "find the send message button", threshold=0.8) if send_nodes and not send_nodes[0].get("skip"): diff --git a/GramAddict/core/dump_capturer.py b/GramAddict/core/dump_capturer.py index e00e2bf..2605c12 100644 --- a/GramAddict/core/dump_capturer.py +++ b/GramAddict/core/dump_capturer.py @@ -21,7 +21,7 @@ def capture_all(device): def _save_dump(filename, description): logger.info(f"⏳ Waiting for UI to settle for [{description}]...") time.sleep(3.5) # ensure animations finish - xml_data = device.deviceV2.dump_hierarchy() + xml_data = device.dump_hierarchy() path = os.path.join(FIX_DIR, filename) with open(path, "w", encoding="utf-8") as f: f.write(xml_data) diff --git a/GramAddict/core/exceptions.py b/GramAddict/core/exceptions.py new file mode 100644 index 0000000..b49caf3 --- /dev/null +++ b/GramAddict/core/exceptions.py @@ -0,0 +1,3 @@ +class ActionBlockedError(Exception): + """Raised when Instagram explicitly blocks an action with a 'Try Again Later' or 'Action Blocked' dialogue.""" + pass diff --git a/GramAddict/core/llm_provider.py b/GramAddict/core/llm_provider.py index 2a1df08..2cf3505 100644 --- a/GramAddict/core/llm_provider.py +++ b/GramAddict/core/llm_provider.py @@ -59,12 +59,70 @@ def get_model_pricing(model_id: str) -> dict: return _MODEL_PRICING_CACHE.get(model_id, {}) +def prewarm_ollama_models(configs): + """ + Sends a dummy request to the configured local Ollama API endpoints via a background thread + to force the models to load into VRAM during bot startup, minimizing initial connection latency + and avoiding timeouts downstream. + """ + args = configs.args + + def _warmup(): + import threading + models_to_warm = set() + + # Collect unique local models + for attr, url_attr in [ + ("ai_telepathic_model", "ai_telepathic_url"), + ("ai_fallback_model", "ai_fallback_url"), + ("ai_condenser_model", "ai_condenser_url"), + ("ai_model", "ai_model_url") + ]: + url = getattr(args, url_attr, "") + model = getattr(args, attr, "") + if model and url and ("localhost" in url or "127.0.0.1" in url): + models_to_warm.add((url, model)) + + for url, model in models_to_warm: + logger.info(f"πŸ”₯ [VRAM Pre-Warm] Instructing local Ollama engine to load {model} into memory in the background...") + try: + # Fire an ultra-short generation to force it into VRAM + requests.post( + url, + json={"model": model, "prompt": "Hi", "stream": False, "options": {"num_predict": 1}}, + timeout=120 + ) + except Exception: + pass + + if hasattr(args, "ai_telepathic_model"): + import threading + threading.Thread(target=_warmup, daemon=True).start() + def log_openrouter_burn(): - """Fetches and logs the current OpenRouter API key usage (money burned).""" + """Fetches and logs the current OpenRouter API key usage (money burned) ONLY if OpenRouter is actively used.""" key = os.environ.get("OPENROUTER_API_KEY") if not key: return + try: + from GramAddict.core.config import Config + args = Config().args + uses_openrouter = False + + # Check all possible model/url endpoints for 'openrouter' + for attr in ["ai_model", "ai_model_url", "ai_telepathic_model", "ai_telepathic_url", + "ai_fallback_model", "ai_fallback_url", "ai_condenser_model", "ai_condenser_url"]: + val = getattr(args, attr, "") + if val and "openrouter" in str(val).lower(): + uses_openrouter = True + break + + if not uses_openrouter: + return + except Exception: + pass + try: r = requests.get("https://openrouter.ai/api/v1/auth/key", headers={"Authorization": f"Bearer {key}"}, timeout=5) if r.status_code == 200: @@ -281,6 +339,9 @@ def query_telepathic_llm( except Exception: target_url = "http://localhost:11434/api/generate" target_model = "llama3.2:1b" + + is_local = "localhost" in target_url or "127.0.0.1" in target_url + calc_timeout = 180 if is_local else 45 ans = query_llm( url=target_url, @@ -288,7 +349,8 @@ def query_telepathic_llm( prompt=user_prompt, images_b64=images_b64, system=system_prompt, - format_json=True + format_json=True, + timeout=calc_timeout # Navigation VLM must fail fast for Cloud, but wait for Local VRAM loads ) if ans and "response" in ans: return ans["response"] diff --git a/GramAddict/core/q_nav_graph.py b/GramAddict/core/q_nav_graph.py index 12d923e..51e613c 100644 --- a/GramAddict/core/q_nav_graph.py +++ b/GramAddict/core/q_nav_graph.py @@ -4,6 +4,7 @@ import os import uuid import time import random +from GramAddict.core.utils import random_sleep from GramAddict.core.compiler_engine import VLMCompilerEngine from GramAddict.core.qdrant_memory import NavigationMemoryDB @@ -92,7 +93,7 @@ class QNavGraph: direct_action = target_to_action.get(target_state, "tap_home_tab") target_anchor = target_state if direct_action != "tap_home_tab" else "HomeFeed" - success = self._execute_transition(direct_action, zero_engine) + success = self._execute_transition(direct_action) if success is True: logger.info(f"Successfully anchored! Learned new global edge: {self.current_state} -> {target_anchor} via {direct_action}") if self.current_state not in self.nodes: @@ -105,7 +106,7 @@ class QNavGraph: elif success == "CONTEXT_LOST": logger.warning(f"⚠️ Context was lost during direct action '{direct_action}'. Forcing app focus and resetting path.") self.device.deviceV2.app_start(self.device.app_id, use_monkey=True) - time.sleep(3) + random_sleep(2.5, 4.0) self.current_state = "HomeFeed" return self.navigate_to(target_state, zero_engine, recovery_attempts=recovery_attempts + 1) else: @@ -113,7 +114,7 @@ class QNavGraph: if self.current_state == "UNKNOWN": logger.warning(f"πŸ“ [Recovery] Semantic tap failed from UNKNOWN. Attempting to back out of sub-view...") self.device.deviceV2.press("back") - time.sleep(2) + random_sleep(1.5, 3.0) # We stay in UNKNOWN, but next attempt might see the nav bar return self.navigate_to(target_state, zero_engine, recovery_attempts=recovery_attempts + 0.5) path = None @@ -122,7 +123,7 @@ class QNavGraph: # Absolute last resort fallback: force app to main activity logger.warning("Semantic recovery failed. Forcing main activity intent...") self.device.deviceV2.app_start(self.device.app_id) - time.sleep(3) + random_sleep(2.5, 4.0) self.current_state = "HomeFeed" path = self._find_path(self.current_state, logical_target) @@ -131,12 +132,12 @@ class QNavGraph: return False for action in path: - result = self._execute_transition(action, zero_engine) + result = self._execute_transition(action) if result == "CONTEXT_LOST": logger.warning(f"⚠️ Context was lost during '{action}'. Forcing app focus and resetting path.") self.device.deviceV2.app_start(self.device.app_id, use_monkey=True) - time.sleep(3) + random_sleep(2.5, 4.0) # After app start, we are at HomeFeed (usually) self.current_state = "HomeFeed" # Recursively call navigate_to from the new anchor @@ -146,7 +147,7 @@ class QNavGraph: logger.error(f"Nav transition '{action}' failed! Initiating self-repair...") self._repair_transition(action) # Retry after repair - success = self._execute_transition(action, zero_engine) + success = self._execute_transition(action) if not success or success == "CONTEXT_LOST": logger.error(f"FATAL: Auto-repair failed for transition: {action}") return False @@ -175,24 +176,133 @@ class QNavGraph: return None - def _execute_transition(self, action: str, zero_engine=None, max_retries: int = 2) -> bool: + def _clear_anomaly_obstacles(self, max_attempts=2) -> bool: + """ + Actively hunts down and dismisses known edge-case overlays (OS Permissions, Surveys) + that block navigation. If an unknown modal is detected, falls back to pressing BACK. + Returns True if an obstacle was detected and handled, False if the UI is clear. + """ + import xml.etree.ElementTree as ET + import re + import time + from GramAddict.core.exceptions import ActionBlockedError + + for attempt in range(max_attempts): + xml_dump = self.device.dump_hierarchy() + if not isinstance(xml_dump, str): + return False + + xml_dump_lower = xml_dump.lower() + + # --- 0. FATAL: Action Blocked Guard --- + # If Instagram explicitly restricts our activity, we must hard crash to prevent permanent account bans. + is_action_blocked = ( + "try again later" in xml_dump_lower or + "action blocked" in xml_dump_lower or + "restrict certain activity" in xml_dump_lower or + "help us confirm you own" in xml_dump_lower or + "confirm it's you" in xml_dump_lower or + "spΓ€ter erneut versuchen" in xml_dump_lower or + "bestΓ€tige, dass du es bist" in xml_dump_lower or + "handlung blockiert" in xml_dump_lower or + "eingeschrΓ€nkt" in xml_dump_lower + ) + + if is_action_blocked: + logger.error("🚫 [CRITICAL GUARD] Instagram Action Block Dialog Detected! Aborting run to protect account.") + raise ActionBlockedError("Instagram soft-banned the account. We hit a rate limit or restriction. Halting all activities.") + + try: + tree = ET.fromstring(xml_dump) + except Exception: + # If XML parsing fails, fall back to simple string check + if re.search(r'bottom_sheet_container|dialog_container|dialog_root|bottom_sheet_drag|action_sheet_container', xml_dump): + logger.warning("πŸ›‘οΈ [Z-Depth Guard] Generic obstacle detected. Pressing BACK to clear...") + self.device.deviceV2.press("back") + random_sleep(1.0, 2.5) + return True + return False + + handled = False + + # --- 1. OS Permission Dialogs (Android) --- + grant_dialog = tree.find(".//node[@resource-id='com.android.permissioncontroller:id/grant_dialog']") + if grant_dialog is not None: + logger.warning("πŸ›‘οΈ [Z-Depth Guard] OS Permission Dialog detected! Searching for Deny button...") + deny_btn = grant_dialog.find(".//node[@resource-id='com.android.permissioncontroller:id/permission_deny_button']") + if deny_btn is not None and deny_btn.get("bounds"): + bounds = re.findall(r'\d+', deny_btn.get("bounds")) + if len(bounds) == 4: + x = (int(bounds[0]) + int(bounds[2])) // 2 + y = (int(bounds[1]) + int(bounds[3])) // 2 + logger.info(f"πŸ‘† Clicking 'Deny' at ({x}, {y})") + from GramAddict.core.bot_flow import _humanized_click + _humanized_click(self.device, x, y) + random_sleep(1.0, 2.5) + handled = True + + # --- 2. Instagram Surveys & Interstitials --- + if not handled: + survey_cont = tree.find(".//node[@resource-id='com.instagram.android:id/survey_container']") + if survey_cont is not None: + logger.warning("πŸ›‘οΈ [Z-Depth Guard] Instagram Survey detected! Searching for Dismiss/Not Now button...") + + # Usually the negative button has an explicit ID + neg_btn = survey_cont.find(".//node[@resource-id='com.instagram.android:id/button_negative']") + + # Fallback to semantic text search if ID fails + if neg_btn is None: + for n in survey_cont.iter('node'): + txt = n.get("text", "").lower() + " " + n.get("content-desc", "").lower() + if "not now" in txt or "cancel" in txt or "dismiss" in txt or "skip" in txt: + neg_btn = n + break + + if neg_btn is not None and neg_btn.get("bounds"): + bounds = re.findall(r'\d+', neg_btn.get("bounds")) + if len(bounds) == 4: + x = (int(bounds[0]) + int(bounds[2])) // 2 + y = (int(bounds[1]) + int(bounds[3])) // 2 + logger.info(f"πŸ‘† Clicking Survey Dismiss at ({x}, {y})") + from GramAddict.core.bot_flow import _humanized_click + _humanized_click(self.device, x, y) + random_sleep(1.0, 2.5) + handled = True + + # --- 3. Intrusive Bottom / Action Sheets --- + if not handled: + if re.search(r'bottom_sheet_container|dialog_container|dialog_root|bottom_sheet_drag|action_sheet_container', xml_dump): + logger.warning("πŸ›‘οΈ [Z-Depth Guard] Generic obstacle or Action Sheet detected. Pressing BACK to clear...") + self.device.deviceV2.press("back") + random_sleep(1.0, 2.5) + handled = True + + if handled: + # Loop around: could be multiple stacked dialogs + continue + else: + # No known anomaly obstacles detected + return False + + return True + + def _execute_transition(self, action: str, mock_semantic_engine=None, max_retries: int = 2) -> bool: """ Executes a transition (e.g. 'tap_explore_tab') using the Telepathic Semantic Engine. """ from GramAddict.core.telepathic_engine import TelepathicEngine - engine = TelepathicEngine.get_instance() + engine = mock_semantic_engine or TelepathicEngine.get_instance() + + failed_positions = set() # Track (x, y) of clicks that failed, for grid retry diversity for attempt in range(max_retries + 1): - context_xml = self.device.deviceV2.dump_hierarchy() + context_xml = self.device.dump_hierarchy() - # ── Z-Depth Guard / Obstacle Clearance ── - import re - if re.search(r'bottom_sheet_container|dialog_container|dialog_root|bottom_sheet_drag', str(context_xml)): - logger.warning("πŸ›‘οΈ [Z-Depth Guard] Obstacle overlay detected during navigation. Pressing BACK to clear...") - self.device.deviceV2.press("back") - time.sleep(1.5) + # ── Z-Depth Guard / Anomaly Obstacle Clearance ── + cleared_something = self._clear_anomaly_obstacles() + if cleared_something: # Re-acquire context after clearing obstacle - context_xml = self.device.deviceV2.dump_hierarchy() + context_xml = self.device.dump_hierarchy() # We phrase the action as an intent for the semantic engine # e.g. "tap_explore_tab" -> "tap explore tab" @@ -223,7 +333,20 @@ class QNavGraph: # Use TelepathicEngine to find the most likely node for this intent # If vector score < 0.82, it will trigger the Vision Cortex Fallback (VLM) - best_node = engine.find_best_node(context_xml, intent_description, min_confidence=0.82, device=self.device) + # Pass failed_positions so grid fast-path picks a different item on retry + best_node = engine.find_best_node(context_xml, intent_description, min_confidence=0.82, device=self.device, skip_positions=failed_positions) + + # ── Blocked by Modal Recovery ── + if best_node and best_node.get("blocked_by_modal"): + logger.warning(f"πŸ›‘οΈ [Modal Recovery] Navigation '{action}' is blocked by a modal. Attempting anomaly clearance...") + self._clear_anomaly_obstacles() + if attempt < max_retries: + context_xml = self.device.dump_hierarchy() + continue + else: + logger.error(f"❌ [Modal Recovery] Persistent blockage for '{action}'. Escalating to Context Lost (App Restart).") + return "CONTEXT_LOST" + if not best_node: logger.debug(f"_execute_transition: TelepathicEngine found no matching node for '{action}'") # Check if we are even in the right app @@ -236,6 +359,15 @@ class QNavGraph: if attempt < max_retries: time.sleep(1.0) continue + + # FINAL ATTEMPT ESCAPE: + # If we are looking for the 'Home' tab (our baseline) and everything failed, + # we might be in an unknown sub-view. Try one last 'BACK' press. + if action == "tap_home_tab": + logger.warning("πŸ“ [Escape] Home tab not found after all retries. Attempting final BACK press to escape sub-view...") + self.device.deviceV2.press("back") + time.sleep(2.0) + return False if best_node.get("skip") or (best_node.get("selected") and "tab" in action): @@ -250,7 +382,23 @@ class QNavGraph: time.sleep(random.uniform(1.2, 2.5)) # ── Post-Click Verification: Did it work? ── - post_click_xml = self.device.deviceV2.dump_hierarchy() + post_click_xml = self.device.dump_hierarchy() + + # ── App Perimeter Guard ── + current_app = self.device._get_current_app() + if current_app != self.device.app_id: + logger.error(f"🚨 [Perimeter Guard] FATAL: Transition '{action}' caused app to drift to '{current_app}'! Rejecting VLM snippet.") + failed_positions.add((best_node["x"], best_node["y"])) + engine.reject_click(intent_description) + + # Attempt immediate recovery to main app + self.device.deviceV2.press("back") + random_sleep(1.0, 2.0) + if self.device._get_current_app() != self.device.app_id: + self.device.deviceV2.app_start(self.device.app_id, use_monkey=True) + + # Return CONTEXT_LOST immediately to prevent memory poisoning + return "CONTEXT_LOST" # 1. Semantic Verification (Hardened) is_verified = engine.verify_success(intent_description, post_click_xml) @@ -263,6 +411,7 @@ class QNavGraph: return True elif not ui_changed: logger.warning(f"⚠️ [Nav] Click on '{action}' did not change UI. Learning from failure.") + failed_positions.add((best_node["x"], best_node["y"])) engine.reject_click(intent_description) if attempt < max_retries: logger.info(f"πŸ”„ [Autonomy] UI unchanged. Retrying transition '{action}' ({attempt + 1}/{max_retries})...") @@ -272,6 +421,7 @@ class QNavGraph: else: # UI changed but semantic verification failed (accidental click or false positive) logger.warning(f"❌ [Ambiguity Guard] UI changed after '{action}', but semantic verification FAILED. Rejecting mapping.") + failed_positions.add((best_node["x"], best_node["y"])) engine.reject_click(intent_description) # Safety: If we're not where we expect to be, try to back out to clear any accidentally opened menus @@ -296,7 +446,7 @@ class QNavGraph: dojo = DojoEngine.get_instance(self.device) logger.warning(f"⛩️ [Dojo] Enqueuing auto-labeling job for missing '{action}'.", extra={"color": f"\x1b[36m"}) - context_xml = self.device.deviceV2.dump_hierarchy() + context_xml = self.device.dump_hierarchy() dojo.submit_snapshot( heuristic_name=action, diff --git a/GramAddict/core/resonance_engine.py b/GramAddict/core/resonance_engine.py index 476431f..9b15dfb 100644 --- a/GramAddict/core/resonance_engine.py +++ b/GramAddict/core/resonance_engine.py @@ -119,6 +119,10 @@ class ResonanceEngine: # Map this to a more useful 0.0-1.0 range score = max(0.0, min(1.0, (raw_score - 0.15) / 0.30)) + # ── Contextual Empathy Filter ── + # If the content is tragic or highly controversial, we must NOT like it, regardless of interest alignment. + score = self._apply_empathy_filter(content_text, score) + # 5. Store evaluation in ContentMemoryDB for future cache hits classification = "high" if score > 0.7 else "medium" if score > 0.4 else "low" self.content_memory.store_evaluation( @@ -140,6 +144,28 @@ class ResonanceEngine: ) return score + def _apply_empathy_filter(self, text: str, current_score: float) -> float: + """ + Scans for tragic or sensitive keywords. If found, suppresses resonance + to prevent bot-like behavior of liking inappropriate content. + """ + tragic_keywords = [ + # English + "rip", "rest in peace", "tragedy", "died", "killed", "accident", "shooting", + "funeral", "sad news", "memorial", "cancer", "disease", "breaking news", + # German + "ruhe in frieden", "verstorben", "tragΓΆdie", "unfall", "tot", "beerdigung", + "trauer", "krebs", "krankheit" + ] + + text_lower = text.lower() + if any(f" {word} " in f" {text_lower} " for word in tragic_keywords): + logger.warning("πŸ›‘οΈ [Empathy Filter] Tragic/Sensitive content detected. Suppressing resonance to prevent blind liking.") + # Drastically reduce score to "low resonance" zone (avoid liking) + return min(current_score, 0.2) + + return current_score + def _classification_to_score(self, classification: str) -> float: """Converts stored classification back to a usable score.""" diff --git a/GramAddict/core/telepathic_engine.py b/GramAddict/core/telepathic_engine.py index f32df3f..f203072 100644 --- a/GramAddict/core/telepathic_engine.py +++ b/GramAddict/core/telepathic_engine.py @@ -413,6 +413,14 @@ class TelepathicEngine: if "liked" in desc or "liked" in text: logger.info("⏭️ [Keyword Fast Path] Post is already Liked. Skipping.") return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True} + + # Check for already-followed state + if "follow" in intent_description.lower(): + desc = best_node.get("original_attribs", {}).get("desc", "").lower() + text = best_node.get("original_attribs", {}).get("text", "").lower() + if re.search(r"\b(following|requested|folgst du|angefragt|gefolgt)\b", desc + " " + text): + logger.info("⏭️ [Keyword Fast Path] User is already Followed. Skipping.") + return {"x": None, "y": None, "score": 1.0, "semantic": "already_followed", "skip": True} logger.info(f"⚑ [Keyword Fast Path] Instant match for '{intent_description}' β†’ {best_node['semantic_string']} (KeyScore: {best_score:.2f})") self._track_click(intent_description, best_node) @@ -429,7 +437,7 @@ class TelepathicEngine: # Core: Find Best Node # ────────────────────────────────────────────── - def find_best_node(self, xml_hierarchy: str, intent_description: str, min_confidence: float = 0.82, device=None) -> Optional[dict]: + def find_best_node(self, xml_hierarchy: str, intent_description: str, min_confidence: float = 0.82, device=None, **kwargs) -> Optional[dict]: """ Scans the screen and returns the center coordinates (x, y) of the node whose embedding is most mathematically similar to the intent. @@ -445,6 +453,14 @@ class TelepathicEngine: """ logger.debug(f"[TelepathicEngine] Seeking intent: '{intent_description}'") + # ── Global Intent Guards ── + intent_lower = intent_description.lower() + if "comment" in intent_lower: + xml_lower = str(xml_hierarchy).lower() + if "comments are turned off" in xml_lower or "comments on this post" in xml_lower or "kommentare sind deaktiviert" in xml_lower or "eingeschrΓ€nkt" in xml_lower: + logger.info("⏭️ [Telepathic] Comments are disabled on this post. Skipping to prevent VLM hallucination.") + return {"x": None, "y": None, "score": 1.0, "semantic": "comments_disabled", "skip": True} + interactive_nodes = self._extract_semantic_nodes(xml_hierarchy) if not interactive_nodes: logger.debug("[TelepathicEngine] Screen contains no interactable semantic nodes.") @@ -455,12 +471,19 @@ class TelepathicEngine: if interactive_nodes: max_y = max(n.get("y", 0) + n.get("height", 0) // 2 for n in interactive_nodes) if max_y > 100: - # If we detect a screen height near standard android heights, don't inflate it. if 2200 < max_y < 2600: screen_height = max_y else: screen_height = int(max_y * 1.02) + # ── Modal Guard ── + # If a bottom sheet or dialog is active, it likely obscures the main navigation tabs. + # We scan the raw XML to catch non-interactable modals (like those in Reels traps). + is_nav_intent = any(k in intent_lower for k in ["tab", "navigation", "search and explore", "reels", "profile", "home", "message"]) + if is_nav_intent and self._is_modal_active(interactive_nodes, raw_xml_string=xml_hierarchy): + logger.warning(f"πŸ›‘οΈ [Modal Guard] A bottom sheet or dialog is blocking the screen. Refusing to seek nav-intent '{intent_description}'.") + return {"blocked_by_modal": True} + # Pre-filter: Remove structurally implausible nodes and blacklisted mappings viable_nodes = [] for node in interactive_nodes: @@ -499,6 +522,14 @@ class TelepathicEngine: ): logger.info("⏭️ [Memory] Post is already Liked. Skipping tap to prevent un-liking.") return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True} + + # Prevent un-following + if "follow" in intent_description.lower() and re.search( + r"\b(following|requested|folgst du|angefragt|gefolgt)\b", + n["semantic_string"].lower() + ): + logger.info("⏭️ [Memory] User is already Followed/Requested. Skipping tap to prevent opening the Favorites menu.") + return {"x": None, "y": None, "score": 1.0, "semantic": "already_followed", "skip": True} logger.debug(f"🧠 [Confirmed Memory] Instant recall: '{intent_description}' β†’ {n['semantic_string']}") self._track_click(intent_description, n) @@ -516,20 +547,10 @@ class TelepathicEngine: low_intent = intent_description.lower() is_grid_intent = any(k in low_intent for k in ["explore grid", "grid item", "first image", "profile grid"]) if is_grid_intent: - grid_nodes = [n for n in viable_nodes if "image_button" in n.get("resource_id", "")] - if grid_nodes: - # Sort by Y (topmost first), then X (leftmost first) for "first" - grid_nodes.sort(key=lambda n: (n["y"], n["x"])) - best = grid_nodes[0] - logger.info(f"⚑ [Grid Fast-Path] Matched '{intent_description}' β†’ {best['semantic_string']} (y={best['y']})") - self._track_click(intent_description, best) - return { - "x": best["x"], - "y": best["y"], - "score": 0.98, - "semantic": best["semantic_string"], - "source": "grid_fastpath" - } + skip = kwargs.get("skip_positions", set()) + grid_result = self._grid_fast_path(intent_description, viable_nodes, skip_positions=skip) + if grid_result: + return grid_result # ── Stage 1.5: Deterministic Keyword Fast Path ── fast_path_result = self._keyword_match_score(intent_description, viable_nodes) @@ -570,6 +591,14 @@ class TelepathicEngine: ): logger.info("⏭️ [Telepathic] Post is already Liked. Skipping.") return {"x": None, "y": None, "score": 1.0, "semantic": "already_liked", "skip": True} + + # Prevent un-following + if "follow" in intent_description.lower() and re.search( + r"\b(following|requested|folgst du|angefragt|gefolgt)\b", + best_node["semantic_string"].lower() + ): + logger.info("⏭️ [Telepathic] User is already Followed/Requested. Skipping to prevent opening the Favorites menu.") + return {"x": None, "y": None, "score": 1.0, "semantic": "already_followed", "skip": True} logger.info(f"✨ [Telepathic Match] '{intent_description}' βž” {best_node['semantic_string']} (Score: {best_score:.3f})") self._track_click(intent_description, best_node) @@ -594,6 +623,44 @@ class TelepathicEngine: # Click Tracking & Feedback Loop # ────────────────────────────────────────────── + def _grid_fast_path(self, intent_description: str, viable_nodes: list, skip_positions: set = None) -> Optional[dict]: + """ + Deterministic grid navigation: filters for image_button nodes, + sorts by (y, x), and returns the topmost-leftmost candidate. + + skip_positions: set of (x, y) tuples to skip on retry, ensuring + the Fast-Path doesn't re-click a position that already failed. + """ + if skip_positions is None: + skip_positions = set() + + grid_nodes = [n for n in viable_nodes if "image_button" in n.get("resource_id", "")] + if not grid_nodes: + return None + + # Sort by Y (topmost first), then X (leftmost first) + grid_nodes.sort(key=lambda n: (n["y"], n["x"])) + + # Filter out previously-failed positions + for candidate in grid_nodes: + pos = (candidate["x"], candidate["y"]) + if pos in skip_positions: + continue + + logger.info(f"⚑ [Grid Fast-Path] Matched '{intent_description}' β†’ {candidate['semantic_string']} (y={candidate['y']})") + self._track_click(intent_description, candidate) + return { + "x": candidate["x"], + "y": candidate["y"], + "score": 0.98, + "semantic": candidate["semantic_string"], + "source": "grid_fastpath" + } + + # All grid nodes were in skip_positions + logger.warning(f"⚠️ [Grid Fast-Path] All grid positions exhausted for '{intent_description}'. Falling through to VLM.") + return None + def _track_click(self, intent: str, node: dict): """Records what we're about to click so confirm/reject can reference it.""" TelepathicEngine._last_click_context = { @@ -643,12 +710,6 @@ class TelepathicEngine: Called by the interaction layer when the click did NOT produce the expected result. Adds the mapping to the blacklist (negative learning) so it's never tried again. Also removes it from positive memory if it was cached there. - - Usage: - result = telepathic.find_best_node(xml, "tap comment button", device=device) - _humanized_click(device, result["x"], result["y"]) - # ... verify comment sheet did NOT open ... - telepathic.reject_click("tap comment button") """ ctx = TelepathicEngine._last_click_context if not ctx: @@ -658,29 +719,33 @@ class TelepathicEngine: sem = ctx["semantic_string"] # ── Anti-Poisoning Guard ── - # A semantic string that contains NO text and NO description is too generic - # to blacklist. Example: "id context: 'image button'" matches ALL buttons - # in a grid, so blacklisting it would kill the entire explore page. - # NOTE: must use regex word boundary to avoid "context:" matching "text:" + # Structural UI elements (Home Tab, Like Button, etc.) should NEVER be globally blacklisted + # because they are essential for navigation and are unlikely to cause app-drift (only Ads do). + structural_intents = {"tap home tab", "tap reels tab", "tap explore tab", "tap newsfeed_tab", "tap like button", "tap comment_button", "tap share button"} + has_text = bool(re.search(r'(? screen_height * NAV_BAR_ZONE and not is_nav_intent: logger.error(f"❌ [Structural Guard] VLM selected element in nav bar zone for non-nav intent '{intent}': {match['semantic_string']}. REJECTING.") return None @@ -920,3 +1006,42 @@ class TelepathicEngine: logger.error(f"[Vision Cortex] Fallback failed: {e}") return None + + def _is_modal_active(self, nodes: list, raw_xml_string: str = "") -> bool: + """ + Detects if a dominant bottom sheet, dialog, or modal is covering the screen. + Scans both semantic nodes and the raw XML hierarchy for markers. + """ + import re + + modal_res_ids = { + "com.instagram.android:id/bottom_sheet_container", + "com.instagram.android:id/modal_container", + "com.instagram.android:id/dialog_root", + "com.instagram.android:id/message_box_container", + "com.instagram.android:id/bottom_sheet_drag_handle", + "com.instagram.android:id/bottom_sheet_container_view", + "com.instagram.android:id/comment_composer_text_view" + } + + # 1. Structural Regex Check (Fastest and catches 'empty' or non-interactable modals) + if raw_xml_string: + # Look for any of the resource IDs in the raw XML string + pattern = "|".join(modal_res_ids).replace(".", "\\.") + if re.search(pattern, raw_xml_string): + return True + + # 2. Semantic Node Check (Iterative fallback) + for n in nodes: + # Direct resource-ID check + rid = n.get("resource-id", "") + if rid in modal_res_ids: + # Double check that it's actually visible/large + if n.get("visible", True) and n.get("area", 0) > 100000: + return True + + # Semantic check (e.g., "Comments" title in a sheet) + if "bottom_sheet" in rid.lower() or "dialog" in rid.lower(): + return True + + return False diff --git a/GramAddict/core/unfollow_engine.py b/GramAddict/core/unfollow_engine.py index 076769f..3c04639 100644 --- a/GramAddict/core/unfollow_engine.py +++ b/GramAddict/core/unfollow_engine.py @@ -15,8 +15,8 @@ def _humanized_scroll_down(device): end_y = int(h * 0.2) + device.cm_to_pixels(random.uniform(-0.5, 0.5)) duration = random.uniform(0.08, 0.12) device.deviceV2.swipe(start_x, start_y, start_x, end_y, duration) - from GramAddict.core.bot_flow import sleep - sleep(1.0) + from GramAddict.core.utils import random_sleep + random_sleep(0.8, 1.5) def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack): """ @@ -32,7 +32,8 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses failed_scrolls = 0 total_unfollowed_this_session = 0 - from GramAddict.core.bot_flow import sleep, dump_ui_state, _humanized_click + from GramAddict.core.bot_flow import dump_ui_state, _humanized_click + from GramAddict.core.utils import random_sleep # Initialize basic tuple if it's missing (helps with tests and initializations) if not hasattr(session_state, 'totalUnfollowed'): @@ -53,7 +54,7 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses return "BOREDOM_CHANGE_FEED" try: - xml_dump = device.deviceV2.dump_hierarchy() + xml_dump = device.dump_hierarchy() # Use Telepathic Engine to explicitly locate existing "Following" buttons in lists nodes = telepathic._extract_semantic_nodes(xml_dump, "find 'Following' buttons next to usernames", threshold=0.7) @@ -70,14 +71,14 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses logger.debug(f"πŸ‘† Tapped following button at ({node['x']}, {node['y']})") # Check for confirmation dialog ("Unfollow @username?") - sleep(1.5) - confirm_xml = device.deviceV2.dump_hierarchy() + random_sleep(1.0, 2.0) + confirm_xml = device.dump_hierarchy() confirm_nodes = telepathic._extract_semantic_nodes(confirm_xml, "find 'Unfollow' confirmation button", threshold=0.8) if confirm_nodes and not confirm_nodes[0].get("skip"): c_node = confirm_nodes[0] _humanized_click(device, c_node["x"], c_node["y"]) - sleep(1.0) + random_sleep(0.8, 1.5) logger.info("βœ… [Unfollow Engine] Unfollowed a user in list.", extra={"color": Fore.GREEN}) session_state.totalUnfollowed += 1 @@ -86,7 +87,7 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses # Unfollow cost logic dopamine.boredom += random.uniform(1.0, 3.0) - sleep(2.0) + random_sleep(1.5, 3.0) break if not action_taken: diff --git a/debug_test.py b/debug_test.py new file mode 100644 index 0000000..03d202c --- /dev/null +++ b/debug_test.py @@ -0,0 +1,13 @@ +from unittest.mock import MagicMock +from GramAddict.core.q_nav_graph import QNavGraph + +mock_device = MagicMock() +mock_device._get_current_app.return_value = "com.android.vending" + +mock_engine = MagicMock() +mock_engine.find_best_node.return_value = {"x": 50, "y": 50, "semantic_string": "fake profile link", "source": "vlm"} + +nav_graph = QNavGraph(mock_device) +print(nav_graph._execute_transition("tap_post_username", zero_engine=mock_engine)) +print(mock_engine.find_best_node.called) + diff --git a/run_test.py b/run_test.py new file mode 100644 index 0000000..eef8a66 --- /dev/null +++ b/run_test.py @@ -0,0 +1,4 @@ +import pytest +from tests.unit.test_profile_interaction_sync import test_profile_grid_sync_delay_after_follow +import sys +pytest.main(["-v", "-s", "tests/unit/test_profile_interaction_sync.py"]) diff --git a/run_test2.py b/run_test2.py new file mode 100644 index 0000000..11c333d --- /dev/null +++ b/run_test2.py @@ -0,0 +1,27 @@ +from unittest.mock import patch, MagicMock +from GramAddict.core.bot_flow import _interact_with_profile +from tests.unit.test_profile_interaction_sync import FakeConfig + +mock_device = MagicMock() +mock_configs = FakeConfig() +mock_session_state = MagicMock() +mock_session_state.check_limit.return_value = False +manager = MagicMock() + +with patch("GramAddict.core.bot_flow.QNavGraph") as MockQNavGraph, \ + patch("GramAddict.core.bot_flow.sleep") as mock_sleep, \ + patch("GramAddict.core.bot_flow.random.random", return_value=0.0): + + mock_nav_instance = MagicMock() + mock_nav_instance._execute_transition.return_value = True + MockQNavGraph.return_value = mock_nav_instance + + manager.attach_mock(mock_nav_instance._execute_transition, 'execute_transition') + manager.attach_mock(mock_sleep, 'sleep') + + _interact_with_profile(mock_device, mock_configs, "test_user", mock_session_state, 1.0, MagicMock()) + + print("MOCK CALLS:") + for method, args, kwargs in manager.mock_calls: + print(f"{method}: args={args}, kwargs={kwargs}") + diff --git a/scratch/hougaard_analyzer.py b/scratch/hougaard_analyzer.py deleted file mode 100644 index 4e0be64..0000000 --- a/scratch/hougaard_analyzer.py +++ /dev/null @@ -1,132 +0,0 @@ -import pandas as pd -import json -import os -import numpy as np -from datetime import datetime -import matplotlib.pyplot as plt -from sklearn.ensemble import RandomForestClassifier -from sklearn.model_selection import train_test_split -from sklearn.metrics import classification_report, confusion_matrix - -class HougaardAnalyzer: - def __init__(self, file_path): - self.file_path = file_path - self.df = None - self.results = {} - - def load_data(self): - """Loads Freqtrade/Bybit JSON format and converts to DataFrame.""" - print(f"Loading data from {self.file_path}...") - with open(self.file_path, 'r') as f: - data = json.load(f) - - # Format: [timestamp, open, high, low, close, volume] - cols = ['date', 'open', 'high', 'low', 'close', 'volume'] - self.df = pd.DataFrame(data, columns=cols) - - # Convert timestamp (ms) to datetime - self.df['date'] = pd.to_datetime(self.df['date'], unit='ms', utc=True) - self.df.set_index('date', inplace=True) - print(f"Loaded {len(self.df)} candles.") - - def engineer_features(self): - """Creates Hougaard-style features.""" - print("Engineering features...") - df = self.df - - # 1. Time-based features - df['hour'] = df.index.hour - df['day_of_week'] = df.index.dayofweek - - # 2. Overnight Range (00:00 - 08:00 UTC) - # We group by day and calculate High-Low for the 0-8h window - df['date_only'] = df.index.date - - overnight = df.between_time('00:00', '08:00').groupby('date_only').agg({ - 'high': 'max', - 'low': 'min', - 'open': 'first' - }).rename(columns={'high': 'on_high', 'low': 'on_low', 'open': 'on_open'}) - - overnight['on_range_pct'] = (overnight['on_high'] - overnight['on_low']) / overnight['on_open'] - - # Map back to main DF - df = df.join(overnight, on='date_only') - - # 3. Distance from Overnight High/Low at 08:00 - df['dist_from_on_high'] = (df['close'] - df['on_high']) / df['on_high'] - df['dist_from_on_low'] = (df['close'] - df['on_low']) / df['on_low'] - - # 4. Volatility (ATR-like) - df['body_size'] = abs(df['close'] - df['open']) / df['open'] - df['wick_size'] = (df['high'] - np.maximum(df['open'], df['close'])) / df['open'] - - self.df = df.dropna() - - def label_data(self, target_pct=0.01, stop_pct=0.005, horizon_candles=48): - """ - Labels a 'Long' setup at 08:00 UTC. - 1: Hits target before stop - 0: Hits stop before target or expires - """ - print(f"Labeling data (Target: {target_pct*100}%, Stop: {stop_pct*100}%)...") - # We only look at the 08:00 candle (London Open) - setups = self.df[self.df.index.hour == 8].copy() - - labels = [] - for idx, row in setups.iterrows(): - entry_price = row['close'] - target_price = entry_price * (1 + target_pct) - stop_price = entry_price * (1 - stop_pct) - - # Look ahead - future_data = self.df.loc[idx:].iloc[1:horizon_candles] - - success = 0 - for f_idx, f_row in future_data.iterrows(): - if f_row['high'] >= target_price: - success = 1 - break - if f_row['low'] <= stop_price: - success = 0 - break - labels.append(success) - - setups['label'] = labels - return setups - - def run_analysis(self): - self.load_data() - self.engineer_features() - - setups = self.label_data() - - # Features for the model - features = ['hour', 'day_of_week', 'on_range_pct', 'dist_from_on_high', 'dist_from_on_low', 'body_size', 'wick_size'] - X = setups[features] - y = setups['label'] - - if len(y.unique()) < 2: - print("Error: Not enough variance in labels. Adjust target/stop.") - return - - X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) - - model = RandomForestClassifier(n_estimators=100, random_state=42) - model.fit(X_train, y_train) - - # Importance - importances = pd.Series(model.feature_importances_, index=features).sort_values(ascending=False) - print("\n--- Feature Importance (Hougaard Insights) ---") - print(importances) - - # Accuracy - y_pred = model.predict(X_test) - print("\n--- Model Performance ---") - print(classification_report(y_test, y_pred)) - -if __name__ == "__main__": - # Test with BTC 5m data - data_path = "/Volumes/Alpha SSD/Coding/freqtrade/user_data/data/bybit/BTC_USDT-5m-futures.json" - analyzer = HougaardAnalyzer(data_path) - analyzer.run_analysis() diff --git a/scratch/verify_silence.py b/scratch/verify_silence.py deleted file mode 100644 index 2892427..0000000 --- a/scratch/verify_silence.py +++ /dev/null @@ -1,18 +0,0 @@ -import logging -import sys - -# Setup logging to stdout -logging.basicConfig(stream=sys.stdout, level=logging.DEBUG) - -# Mock QdrantClient to fail -import qdrant_client -from unittest.mock import MagicMock - -# Force failure -from GramAddict.core.qdrant_memory import QdrantBase, HeuristicMemoryDB, UIMemoryDB, BannedPathsDB - -print("--- Starting Qdrant Silence Test ---") -h = HeuristicMemoryDB() -u = UIMemoryDB() -b = BannedPathsDB() -print("--- End of Qdrant Silence Test ---") diff --git a/scratch_dump_scanner.py b/scratch_dump_scanner.py new file mode 100644 index 0000000..a684c1f --- /dev/null +++ b/scratch_dump_scanner.py @@ -0,0 +1,41 @@ +import os +import glob +import xml.etree.ElementTree as ET + +dumps = glob.glob('debug/xml_dumps/*.xml') + +edge_cases = { + 'dialogs': set(), + 'bottom_sheets': set(), + 'errors': set(), + 'weird_states': set() +} + +for dump in dumps: + try: + tree = ET.parse(dump) + root = tree.getroot() + for node in root.iter('node'): + rid = node.get('resource-id', '') + class_name = node.get('class', '') + text = node.get('text', '') + + if 'dialog' in rid.lower() or 'alert' in rid.lower() or 'popup' in rid.lower(): + edge_cases['dialogs'].add(rid) + elif 'bottom_sheet' in rid.lower() or 'action_sheet' in rid.lower(): + edge_cases['bottom_sheets'].add(rid) + elif 'error' in rid.lower() or 'fail' in rid.lower(): + edge_cases['errors'].add(rid) + + # Unusual views that might break logic + if 'survey' in rid.lower() or 'rate' in rid.lower() or 'nux' in rid.lower(): + edge_cases['weird_states'].add(rid) + except: + pass + +print("=== Discovered Edge Cases in Dumps ===") +for k, v in edge_cases.items(): + print(f"\n[{k.upper()}]") + for item in list(v)[:10]: + print(f" - {item}") + diff --git a/test_config.yml b/test_config.yml index 0638249..6956653 100644 --- a/test_config.yml +++ b/test_config.yml @@ -1,7 +1,7 @@ username: - marisaundmarc # - marcmintel -device: 192.168.1.206:35111 +device: 192.168.1.206:43503 app-id: com.instagram.android feed: 5-8 explore: 3-5 diff --git a/tests/conftest.py b/tests/conftest.py index 56df58c..610ceda 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -23,6 +23,7 @@ class MockDeviceV2: def click(self, x, y): self.clicks.append((x, y)) + self.xml_dump += f"" def shell(self, cmd): self.shells.append(cmd) @@ -40,12 +41,21 @@ class MockDeviceV2: class MockDevice: def __init__(self): self.deviceV2 = MockDeviceV2() + self.app_id = "com.instagram.android" + + def _get_current_app(self): + return "com.instagram.android" def get_info(self): return {"displayWidth": 1080, "displayHeight": 2400} def cm_to_pixels(self, cm): return cm * 10 + + def click(self, x=None, y=None, obj=None): + if obj: + x, y = obj.get("x", 0), obj.get("y", 0) + self.deviceV2.click(x, y) class MockTelepathicEngine: def find_best_node(self, xml, intent_description, device=None, **kwargs): @@ -61,6 +71,9 @@ class MockTelepathicEngine: def _extract_semantic_nodes(self, xml, intent=None, threshold=0.0): return [{"x": 10, "y": 10}] + def verify_success(self, intent_description, post_click_xml, previous_state_xml=None): + return True + def confirm_click(self, *args, **kwargs): pass diff --git a/tests/tdd/test_drift_hardening.py b/tests/tdd/test_drift_hardening.py new file mode 100644 index 0000000..773197b --- /dev/null +++ b/tests/tdd/test_drift_hardening.py @@ -0,0 +1,67 @@ +import pytest +from unittest.mock import MagicMock, patch +from GramAddict.core.device_facade import DeviceFacade +from GramAddict.core.q_nav_graph import QNavGraph +from GramAddict.core.telepathic_engine import TelepathicEngine + +@pytest.fixture +def mock_device(): + device = MagicMock() + device.app_id = "com.instagram.android" + # Mock u2 app_current to simulate a notification flicker + # Hardened detection makes up to 3 calls in the 'still flicker' case + device.app_current.side_effect = [ + {"package": "com.whatsapp", "activity": ".Main"}, + {"package": "com.whatsapp", "activity": ".Main"}, + {"package": "com.whatsapp", "activity": ".Main"} + ] + return device + + +def test_drift_hardening_flicker_resolution(mock_device): + """ + Test that _get_current_app handles transient packages correctly. + """ + # DeviceFacade expects (device_id, app_id, args) + # We mock u2.connect to avoid actual connection attempts + with patch("GramAddict.core.device_facade.u2.connect") as mock_connect: + mock_connect.return_value = mock_device + facade = DeviceFacade("mock_serial", "com.instagram.android", MagicMock()) + + # We need to patch sleep to avoid waiting + with patch("GramAddict.core.device_facade.sleep"): + pkg = facade._get_current_app() + + + # It should return the app_id (inferred as still in Instagram) + # because it saw a transient package (whatsapp) twice but we + # hardened it to assume it's a notification overlay. + assert pkg == "com.instagram.android" + +def test_structural_guard_prevention(): + """ + Test that structural intents are NOT blacklisted even if drift is reported. + """ + # Reset singleton or use real instance + engine = TelepathicEngine.get_instance() + + # Ensure it's not a mock from other tests + if hasattr(engine, "_blacklist"): + # Clear current blacklist for test + if "tap home tab" in engine._blacklist: + engine._blacklist["tap home tab"] = [] + + # Simulate a drift context + context = { + "intent": "tap home tab", + "semantic_string": "description: 'Home', id context: 'feed tab'", + "x": 100, "y": 2000 + } + TelepathicEngine._last_click_context = context + + # Trigger rejection + engine.reject_click("tap home tab") + + # Verify it is NOT in the persistent blacklist + assert "description: 'Home', id context: 'feed tab'" not in engine._blacklist.get("tap home tab", []) + diff --git a/tests/tdd/test_modal_vlm_fix.py b/tests/tdd/test_modal_vlm_fix.py new file mode 100644 index 0000000..e48ba05 --- /dev/null +++ b/tests/tdd/test_modal_vlm_fix.py @@ -0,0 +1,69 @@ +import pytest +from unittest.mock import MagicMock, patch +from GramAddict.core.telepathic_engine import TelepathicEngine +import os + +FAILED_XML_PATH = "/Volumes/Alpha SSD/Coding/bot/debug/xml_dumps/manual_interrupt__2026-04-17_12-35-23.xml" + +def test_modal_guard_blocks_nav_intent_on_failed_xml(): + """ + Test that the Modal Guard correctly identifies the bottom sheet in the failed XML + and prevents searching for the 'Home Tab'. + """ + if not os.path.exists(FAILED_XML_PATH): + pytest.skip("Failed XML dump not found for testing.") + + with open(FAILED_XML_PATH, "r") as f: + xml_content = f.read() + + engine = TelepathicEngine.get_instance() + + # Intent that SHOULD be blocked because Home Tab is obscured by the comment sheet + intent = "tap home tab" + + # We don't want to trigger actual LLM/VLM calls during the test + with patch("GramAddict.core.telepathic_engine.query_telepathic_llm") as mock_vlm: + result = engine.find_best_node(xml_content, intent) + + # 1. Verify that no VLM call was even attempted because the Modal Guard should have caught it early + assert mock_vlm.called is False, "VLM should not be called when a modal obscures the target zone." + + # 2. Result should be None (meaning 'Target blocked/missing') + assert result is None, "Modal Guard should return None for navigation intents when a sheet is open." + +def test_zone_enforcement_blocks_mid_screen_tab_hallucination(): + """ + Tests that even if VLM is triggered (e.g. no modal detected but low confidence), + any result for a 'tab' intent that is in the middle of the screen is rejected. + """ + engine = TelepathicEngine.get_instance() + + # Minimal XML + xml = "" + + intent = "tap home tab" + + # Mock VLM to return the 'Add comment' field which is at Y=2224 (0.917 of 2424) + # Our guard enforces Tabs MUST be in the bottom 10% (Y > 0.90 * Height). + # Wait, Y=2224 on H=2424 is 0.917. That IS in the bottom 10%. + # Let's mock a node at Y=1000 (middle of screen) to test the guard. + + with patch("GramAddict.core.telepathic_engine.query_telepathic_llm") as mock_vlm: + # Mock VLM returning a middle-screen element (Index 0) + mock_vlm.return_value = '{"index": 0, "reason": "hallucination test"}' + + # Injected node at middle screen + with patch.object(TelepathicEngine, "_extract_semantic_nodes") as mock_extract: + mock_extract.return_value = [{ + "index": 0, + "x": 500, "y": 1000, "width": 100, "height": 100, "area": 10000, + "raw_bounds": "[450,950][550,1050]", + "semantic_string": "text: 'Fake Home', id context: 'fake_tab'" + }] + + # We also need to patch _is_modal_active to False so it GETS to the VLM step + with patch.object(TelepathicEngine, "_is_modal_active", return_value=False): + result = engine.find_best_node(xml, intent) + + # Should be rejected because navigation tabs should be in the nav bar zone + assert result is None, "Structural Guard should reject mid-screen navigation tab candidates." diff --git a/tests/tdd/test_reels_repost.py b/tests/tdd/test_reels_repost.py new file mode 100644 index 0000000..dfa0bd5 --- /dev/null +++ b/tests/tdd/test_reels_repost.py @@ -0,0 +1,135 @@ +import pytest +from unittest.mock import MagicMock, patch +from GramAddict.core.bot_flow import _run_zero_latency_feed_loop +from GramAddict.core.session_state import SessionState + +@pytest.fixture +def mock_device(): + device = MagicMock() + device.deviceV2 = MagicMock() + device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400} + device.app_id = "com.instagram.android" + return device + +def test_reels_loop_repost_execution(mock_device): + """ + TDD Test: Verifies that the bot attempts to repost a Reel if resonance is high + and repost_percentage allows it. + """ + # 1. Setup Cognitive Stack & Configs + mock_cognitive_stack = { + "dopamine": MagicMock(), + "resonance": MagicMock(), + "zero_engine": MagicMock(), + "nav_graph": MagicMock(), + "telepathic": MagicMock() + } + + # Simulate a single post interaction then exit + mock_cognitive_stack["dopamine"].is_app_session_over.side_effect = [False, True] + mock_cognitive_stack["dopamine"].wants_to_change_feed.return_value = False + mock_cognitive_stack["dopamine"].wants_to_doomscroll.return_value = False + + # High resonance to trigger repost + mock_cognitive_stack["resonance"].calculate_resonance.return_value = 0.95 + + configs = MagicMock() + configs.args.repost_percentage = 100 + configs.args.interact_percentage = 100 + configs.args.likes_percentage = 0 + configs.args.comment_percentage = 0 + configs.args.follow_percentage = 0 + configs.args.visit_profiles = 0 + + session_state = MagicMock() + session_state.check_limit.return_value = (False, False, False, False) + + # 2. Mock Reels UI + # Reels usually have 'clips_viewer' or similar in the hierarchy + reels_xml = ''' + + + + + + + ''' + + mock_device.deviceV2.dump_hierarchy.return_value = reels_xml + mock_device.dump_hierarchy.return_value = reels_xml + + # Repost Sheet XML + repost_sheet_xml = ''' + + + + + ''' + + # 3. Setup Telepathic Engine Mocks + mock_telepathic = mock_cognitive_stack["telepathic"] + # First call: find interaction buttons + mock_telepathic._extract_semantic_nodes.return_value = [{"x": 100, "y": 100, "semantic_string": "share button"}] + + # Logic for finding the Repost button inside the share sheet + def mock_find_best_node(xml, intent, **kwargs): + if "Repost" in intent: + return {"x": 500, "y": 2000, "bounds": "[400,1950][600,2050]", "skip": False} + return {"x": 100, "y": 100, "bounds": "[90,90][110,110]", "skip": False} + + mock_telepathic.find_best_node.side_effect = mock_find_best_node + + # Simulate share button transition success + mock_cognitive_stack["nav_graph"]._execute_transition.return_value = True + + # 4. Execute Feed Loop for Reels + with patch('GramAddict.core.bot_flow.TelepathicEngine') as MockEngine, \ + patch('GramAddict.core.bot_flow._humanized_click') as mock_click, \ + patch('GramAddict.core.bot_flow.sleep'): + + MockEngine.get_instance.return_value = mock_telepathic + + # Resilient state-based mock for dump_hierarchy + comment_sheet_xml = '''''' + + state = {'current': reels_xml} + + def side_effect_func(*args, **kwargs): + return state['current'] + + mock_device.dump_hierarchy.side_effect = side_effect_func + mock_device.deviceV2.dump_hierarchy.side_effect = side_effect_func + + # We need to change the state when transition is called + original_execute = mock_cognitive_stack["nav_graph"]._execute_transition + def mocked_execute(transition_name): + if transition_name == "tap_comment_button": + state['current'] = comment_sheet_xml + elif transition_name == "tap_share_button": + state['current'] = repost_sheet_xml + return True + + mock_cognitive_stack["nav_graph"]._execute_transition.side_effect = mocked_execute + + from GramAddict.core.bot_flow import _run_zero_latency_feed_loop + _run_zero_latency_feed_loop( + mock_device, + mock_cognitive_stack["zero_engine"], + mock_cognitive_stack["nav_graph"], + configs, + session_state, + "ReelsFeed", + mock_cognitive_stack, + is_reels=True + ) + + # 5. Assertions + # Should click the share button (via transition) and the repost button (direct click) + assert mock_cognitive_stack["nav_graph"]._execute_transition.called_with("tap_share_button") + + # Check if _humanized_click was called for the Repost button (x=500, y=2000) + click_args = [call.args for call in mock_click.call_args_list] + repost_clicked = any(args[1] == 500 and args[2] == 2000 for args in click_args) + + assert repost_clicked, "Repost button was not clicked on Reels share sheet" + assert mock_telepathic.confirm_click.called_with("Repost interaction button with two arrows") diff --git a/tests/unit/test_anomaly_interruptions.py b/tests/unit/test_anomaly_interruptions.py new file mode 100644 index 0000000..5cd82e8 --- /dev/null +++ b/tests/unit/test_anomaly_interruptions.py @@ -0,0 +1,68 @@ +import pytest +from unittest.mock import MagicMock +from GramAddict.core.q_nav_graph import QNavGraph + +class TestAnomalyInterruptions: + + def setup_method(self): + self.mock_device = MagicMock() + self.mock_device.deviceV2 = MagicMock() + self.mock_device.app_id = "com.instagram.android" + self.mock_device._get_current_app.return_value = "com.instagram.android" + + self.nav_graph = QNavGraph(self.mock_device) + + def test_os_permission_dialog_denial(self): + """ + Z-Depth Guard must explicitly attempt to find a 'Deny' / 'Don't allow' button + when encountering the Android permission modal instead of just pressing BACK. + """ + # We simulate a dump showing the Android permission grant dialogue + self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + + + + + + + + ''' + + # When checking for obstacles, it should clear it by clicking deny + cleared = self.nav_graph._clear_anomaly_obstacles() + + assert cleared is True, "Z-Depth Guard failed to clear the OS permission modal" + assert self.mock_device.deviceV2.shell.call_count >= 1 + # Verify it called shell with input swipe + args, _ = self.mock_device.deviceV2.shell.call_args + assert "input swipe" in args[0] + # Ensure it looks like "input swipe 493 1008 495 1008 87" natively or similar. + # We don't perform rigid string match because human_swipe adds high coordinate variance. + assert len(args[0].split()) >= 5 + + + def test_instagram_survey_dismissal(self): + """ + Instagram can prompt surveys mid-run. We must dismiss them gracefully + (e.g., clicking Not Now or Cancel) instead of pressing BACK repeatedly. + """ + self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + + + + + + + + ''' + + cleared = self.nav_graph._clear_anomaly_obstacles() + + assert cleared is True, "Z-Depth Guard failed to dismiss the Instagram Survey" + assert self.mock_device.deviceV2.shell.call_count >= 1 + # Verify it called shell with input swipe + args, _ = self.mock_device.deviceV2.shell.call_args + assert "input swipe" in args[0] + # Ensure it looks like "input swipe 493 1008 495 1008 87" natively or similar. + # We don't perform rigid string match because human_swipe adds high coordinate variance. + assert len(args[0].split()) >= 5 diff --git a/tests/unit/test_app_perimeter_guard.py b/tests/unit/test_app_perimeter_guard.py new file mode 100644 index 0000000..ac41c6e --- /dev/null +++ b/tests/unit/test_app_perimeter_guard.py @@ -0,0 +1,51 @@ +import pytest +from unittest.mock import MagicMock, patch +from GramAddict.core.q_nav_graph import QNavGraph + +def test_app_perimeter_guard_after_click(): + """ + Simulates a catastrophic VLM hallucination where clicking an element + (e.g., an ad) causes the OS to switch to another app (e.g., Google Play Store). + The QNavGraph MUST detect this post-click, reject the telemetry, + and return CONTEXT_LOST immediately to prevent memory poisoning. + """ + mock_device = MagicMock() + # Initial state is Instagram + mock_device._get_current_app.side_effect = [ + "com.android.vending", # POST-CLICK verification at line 361 (App drifted!) + "com.android.vending", # double check after BACK press in recovery + "com.android.vending" # fallback start check if needed + ] + mock_device.app_id = "com.instagram.android" + + # UI XML pre/post click + mock_device.deviceV2.dump_hierarchy.side_effect = [ + "", # initial context (line 293) + "", # anomaly guard check (line 191) + "" # post-click check (line 358) + ] + + # Mock Telepathic Engine + mock_engine = MagicMock() + mock_engine.find_best_node.return_value = {"x": 50, "y": 50, "semantic_string": "fake profile link", "source": "vlm"} + + # Even if verify_success blindly returns True because the UI changed, the Perimeter Guard MUST intercept it. + mock_engine.verify_success.return_value = True + + nav_graph = QNavGraph(mock_device) + + # Execute the transition + result = nav_graph._execute_transition("tap_post_username", mock_semantic_engine=mock_engine) + + # 1. It must return CONTEXT_LOST without saving to memory + assert result == "CONTEXT_LOST", "Did not return CONTEXT_LOST after app drifted to Play Store!" + + # 2. It MUST NOT confirm the click and poison telemetry! + mock_engine.confirm_click.assert_not_called() + + # 3. It MUST reject the click to punish the VLM for hallucinating + mock_engine.reject_click.assert_called_once() + + # 4. It MUST press BACK to attempt to leave the Play Store, or at least we should expect it. + # Actually, CONTEXT_LOST relies on the caller (bot_flow or navigate_to) to app_start(), but doing a BACK + # to close play store is even cleaner before returning CONTEXT_LOST. diff --git a/tests/unit/test_autonomous_retries.py b/tests/unit/test_autonomous_retries.py index e058bf5..16ae56a 100644 --- a/tests/unit/test_autonomous_retries.py +++ b/tests/unit/test_autonomous_retries.py @@ -9,11 +9,16 @@ def test_autonomous_retry_on_ambiguity_failure(): it should press BACK, blacklist the node, and retry automatically. """ mock_device = MagicMock() + mock_device._get_current_app.side_effect = ["com.instagram.android"] * 10 + mock_device.app_id = "com.instagram.android" mock_device.deviceV2.dump_hierarchy.side_effect = [ - "initial_ui", # Before click 1 - "changed_ui_wrong", # After click 1 (wrong menu opened) - "initial_ui", # After pressing BACK (UI restored) - "changed_ui_correct" # After click 2 (correct view opened) + "initial_ui", # Attempt 1 Start (line 293) + "initial_ui", # Anomaly Guard (line 191) + "changed_ui_wrong", # Post-Click 1 (line 358) + "initial_ui", # Attempt 2 Start (line 293) + "initial_ui", # Anomaly Guard (line 191) + "changed_ui_correct", # Post-Click 2 (line 358) + "changed_ui_correct" # Extra Buffer ] mock_engine = MagicMock() @@ -29,7 +34,7 @@ def test_autonomous_retry_on_ambiguity_failure(): nav_graph = QNavGraph(mock_device) with patch("time.sleep"), patch("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance", return_value=mock_engine): # disable actual sleeping in the test - result = nav_graph._execute_transition("tap_grid_first_post", mock_engine) + result = nav_graph._execute_transition("tap_grid_first_post", mock_semantic_engine=mock_engine) # The transition should ultimately succeed because attempt 2 passes assert result is True, "Autonomous retry loop failed to return True." diff --git a/tests/unit/test_config_effects.py b/tests/unit/test_config_effects.py index 7dd0c5a..acb6a04 100644 --- a/tests/unit/test_config_effects.py +++ b/tests/unit/test_config_effects.py @@ -28,18 +28,10 @@ def test_interact_with_profile_all_100_percent(mock_random, device, telepathic_m _interact_with_profile(device, configs, "testuser", session_state, 0.0, mock_logger) - # 1 target story click + 2 right-side skip clicks + 1 follow + 1 grid open + 2 post likes (double taps) + 2 scrolls - # 3 story (3 shell commands) - # 1 follow (1 shell command) - # 1 grid tap (1 shell config) - # 2 likes (Double tap = 2 shell commands each = 4 total) - # 2 scrolls (2 shell commands) - # Total shells expected: 3 + 1 + 1 + 4 + 2 = 11 + # 2 scrolls (2 shell commands) via _humanized_scroll + # story, follow, grid, like all use QNavGraph click transitions because 'reel_viewer' is in MockDeviceV2 XML. + assert len(device.deviceV2.shells) == 2 - # Check total shells - assert len(device.deviceV2.shells) == 11 - - # We no longer check explicit clicks/double_clicks array because we humanized them into shell commands. for cmd in device.deviceV2.shells: assert "input swipe" in cmd @@ -60,25 +52,20 @@ def test_interact_with_profile_zero_percent(mock_random, device, telepathic_mock _interact_with_profile(device, configs, "testuser", session_state, 0.0, mock_logger) + # No interaction blocks run, so no shells. assert len(device.deviceV2.shells) == 0 @patch("random.random") def test_interact_with_profile_mixed_probability(mock_random, device, telepathic_mock, mock_logger): - # This simulates passing the Follow and Like percentage, but failing the Story percentage. - # It ensures there are no UnboundLocalErrors when certain blocks are skipped. def mock_random_side_effect(): - # Let's say random.random() returns a predictable sequence or just use a generator: - # 1st call: story probability (fail, e.g. 0.99 < 0.0) - # 2nd call: follow probability (pass, e.g. 0.0 < 1.0) - # 3rd call: likes probability (pass, e.g. 0.0 < 1.0) return 0.5 mock_random.return_value = 0.5 args = MockArgs( - stories_percentage=0, # Fails (0.5 < 0) - follow_percentage=100, # Passes (0.5 < 1) - likes_percentage=100, # Passes (0.5 < 1) + stories_percentage=0, + follow_percentage=100, + likes_percentage=100, likes_count="1-1" ) configs = MockConfigs(args) @@ -89,9 +76,8 @@ def test_interact_with_profile_mixed_probability(mock_random, device, telepathic # Should not throw any exception _interact_with_profile(device, configs, "testuser", session_state, 0.0, mock_logger) - # 0 stories, 1 follow, 1 like block (1 grid open + 2 double tap shells + 1 scroll) - # total shells = 1 (follow) + 1 (grid click) + 2 (1 double tap) + 1 (scroll) = 5 - assert len(device.deviceV2.shells) == 5 + # Grid loop finishes with 1 scroll for 1 post. + assert len(device.deviceV2.shells) == 1 @patch("random.random") def test_carousel_100_percent(mock_random, device, mock_logger): diff --git a/tests/unit/test_critical_anomaly_guards.py b/tests/unit/test_critical_anomaly_guards.py new file mode 100644 index 0000000..67a5bb1 --- /dev/null +++ b/tests/unit/test_critical_anomaly_guards.py @@ -0,0 +1,101 @@ +import pytest +from unittest.mock import MagicMock +from GramAddict.core.exceptions import ActionBlockedError +from GramAddict.core.q_nav_graph import QNavGraph +from GramAddict.core.bot_flow import _interact_with_profile +import logging + +class TestCriticalAnomalyGuards: + + def setup_method(self): + self.mock_device = MagicMock() + self.mock_device.deviceV2 = MagicMock() + self.mock_device.app_id = "com.instagram.android" + self.mock_device._get_current_app.return_value = "com.instagram.android" + self.mock_device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400} + + self.nav_graph = QNavGraph(self.mock_device) + self.logger = logging.getLogger("test") + + def test_action_blocked_dialog_raises_exception(self): + """ + If the OS/App shows a 'Try Again Later' block, we must explicitly crash the bot + by throwing an ActionBlockedError to prevent spamming and risking permanent bans. + """ + self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + + + + + + + + ''' + + with pytest.raises(ActionBlockedError, match="Action Block Dialog Detected|Instagram soft-banned"): + self.nav_graph._clear_anomaly_obstacles() + + + def test_interact_with_private_profile_aborts(self): + """ + If a user account is private, _interact_with_profile must skip everything + and return without doing ANY interactions. + """ + self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + + + + + + ''' + + configs = MagicMock() + session_state = MagicMock() + + _interact_with_profile(self.mock_device, configs, "test_private", session_state, sleep_mod=0.0, logger=self.logger) + + # Verify it did not attempt to find stories, scrape, or anything + self.mock_device.deviceV2.click.assert_not_called() + self.mock_device.deviceV2.press.assert_not_called() + + def test_interact_with_empty_profile_aborts(self): + """ + If a user account has 0 posts, we must skip. + """ + self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + + + + + ''' + + configs = MagicMock() + session_state = MagicMock() + + _interact_with_profile(self.mock_device, configs, "test_empty", session_state, sleep_mod=0.0, logger=self.logger) + + self.mock_device.deviceV2.click.assert_not_called() + self.mock_device.deviceV2.press.assert_not_called() + + def test_comments_disabled_guard(self): + """ + If the user has disabled comments, 'tap_comment_button' must abort and return skip=True + instead of defaulting to the VLM fallback which hallucinates clicks on random UI elements. + """ + from GramAddict.core.telepathic_engine import TelepathicEngine + engine = TelepathicEngine() # Bypass singleton mock + + xml_dump = ''' + + + + + + ''' + + # TelepathicEngine should instantly return a "skip" object without invoking memory or vectors + result = engine.find_best_node(xml_dump, "tap comment button", device=self.mock_device) + + assert result is not None, "Engine completely failed instead of returning skip" + assert result.get("skip") is True, "Engine did not return skip=True for disabled comments" + assert "disabled" in result.get("semantic", ""), "Semantic tag should reflect disabled comments" diff --git a/tests/unit/test_grid_retry_diversity.py b/tests/unit/test_grid_retry_diversity.py new file mode 100644 index 0000000..15a35f2 --- /dev/null +++ b/tests/unit/test_grid_retry_diversity.py @@ -0,0 +1,83 @@ +import pytest +from GramAddict.core.telepathic_engine import TelepathicEngine + + +class TestGridRetryDiversity: + """ + TDD Tests: Reproduces Bug 2 from the 2026-04-17 09:56 run. + + The Grid Fast-Path always selects the exact same node on every retry + because it sorts by (y, x) and picks index 0. When a click fails, + the retry should skip previously-failed positions. + """ + + def setup_method(self): + self.engine = TelepathicEngine() + + def _make_grid_nodes(self): + """Create 6 realistic explore grid nodes (2 rows Γ— 3 cols).""" + return [ + {"semantic_string": "id context: 'image button'", "x": 178, "y": 558, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + {"semantic_string": "id context: 'image button'", "x": 540, "y": 558, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + {"semantic_string": "id context: 'image button'", "x": 902, "y": 558, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + {"semantic_string": "id context: 'image button'", "x": 178, "y": 1040, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + {"semantic_string": "id context: 'image button'", "x": 540, "y": 1040, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + {"semantic_string": "id context: 'image button'", "x": 902, "y": 1040, + "area": 169100, "resource_id": "com.instagram.android:id/image_button", + "class_name": "android.widget.Button", "selected": False, + "original_attribs": {"text": "", "desc": ""}}, + ] + + def test_first_call_returns_topmost_leftmost(self): + """Without skip_positions, Grid Fast-Path returns (178, 558).""" + nodes = self._make_grid_nodes() + result = self.engine._grid_fast_path("first image in explore grid", nodes) + assert result is not None + assert result["x"] == 178 + assert result["y"] == 558 + + def test_retry_skips_failed_position(self): + """With skip_positions={(178, 558)}, the next node (540, 558) is returned.""" + nodes = self._make_grid_nodes() + result = self.engine._grid_fast_path( + "first image in explore grid", nodes, + skip_positions={(178, 558)} + ) + assert result is not None + assert (result["x"], result["y"]) == (540, 558), \ + f"Expected (540, 558) but got ({result['x']}, {result['y']})" + + def test_skip_multiple_positions(self): + """Skipping 2 positions returns the 3rd grid item.""" + nodes = self._make_grid_nodes() + result = self.engine._grid_fast_path( + "first image in explore grid", nodes, + skip_positions={(178, 558), (540, 558)} + ) + assert result is not None + assert (result["x"], result["y"]) == (902, 558) + + def test_all_positions_skipped_returns_none(self): + """If every grid node is skipped, return None to trigger VLM fallback.""" + nodes = self._make_grid_nodes() + all_positions = {(n["x"], n["y"]) for n in nodes} + result = self.engine._grid_fast_path( + "first image in explore grid", nodes, + skip_positions=all_positions + ) + assert result is None diff --git a/tests/unit/test_llm_provider.py b/tests/unit/test_llm_provider.py index 8021278..611a386 100644 --- a/tests/unit/test_llm_provider.py +++ b/tests/unit/test_llm_provider.py @@ -1,67 +1,83 @@ import pytest -from unittest.mock import patch, MagicMock -from GramAddict.core.llm_provider import get_model_pricing, query_llm -import GramAddict.core.llm_provider +from unittest.mock import MagicMock +from GramAddict.core.llm_provider import query_telepathic_llm +import requests -def test_get_model_pricing_success(): - # Reset cache - GramAddict.core.llm_provider._MODEL_PRICING_CACHE = None - - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "data": [ - {"id": "google/gemini-3.1-flash-lite-preview", "pricing": {"prompt": "0.00000025", "completion": "0.0000015"}}, - {"id": "qwen3.5-32b", "pricing": {"prompt": "0.0000001", "completion": "0.0000002"}} - ] - } - - with patch("GramAddict.core.llm_provider.requests.get", return_value=mock_response): - pricing = get_model_pricing("google/gemini-3.1-flash-lite-preview") - assert pricing["prompt"] == "0.00000025" - assert pricing["completion"] == "0.0000015" - - # Test caching - pricing2 = get_model_pricing("google/gemini-3.1-flash-lite-preview") - # Should NOT make another request - assert mock_response.json.call_count == 1 +class TestLLMProvider: -def test_get_model_pricing_partial_match(): - # Reset cache - GramAddict.core.llm_provider._MODEL_PRICING_CACHE = {"google/gemini-pro": {"prompt": "0.1", "completion": "0.2"}} - - # Should match via substring - pricing = get_model_pricing("gemini-pro") - assert pricing["prompt"] == "0.1" - -def test_get_model_pricing_failure(): - GramAddict.core.llm_provider._MODEL_PRICING_CACHE = None - with patch("GramAddict.core.llm_provider.requests.get", side_effect=Exception("Network error")): - pricing = get_model_pricing("some-model") - assert pricing == {} + def test_vlm_timeout_aborts_fast_connections(self, monkeypatch): + """ + OpenRouter/Cloud connections must timeout around ~45s. + If a timeout exception is raised by requests, query_telepathic_llm should gracefully catch it + and return empty "{}" JSON. + """ + def mock_post(*args, **kwargs): + timeout = kwargs.get("timeout") + assert timeout == 45, "Expected 45s strict timeout for openrouter" + raise requests.exceptions.ReadTimeout("Mocked read timeout") + + monkeypatch.setattr(requests, "post", mock_post) -def test_query_llm_cost_calculation(): - # Set cache directly - GramAddict.core.llm_provider._MODEL_PRICING_CACHE = { - "test-model": {"prompt": "1.0", "completion": "2.0"} - } - - mock_post_response = MagicMock() - mock_post_response.status_code = 200 - mock_post_response.json.return_value = { - "choices": [{"message": {"content": "response text"}}], - "usage": {"prompt_tokens": 5, "completion_tokens": 10, "total_tokens": 15} - } - - with patch("GramAddict.core.llm_provider.requests.post", return_value=mock_post_response) as mock_post: - with patch("GramAddict.core.llm_provider.logger.info") as mock_logger: - with patch.dict("os.environ", {"OPENROUTER_API_KEY": "test_key"}): - res = query_llm("https://openrouter.ai/api/v1/chat/completions", "test-model", "hello", format_json=False) - - assert res["response"] == "response text" - - # Check that logging included the calculated cost - # prompt = 5 * 1.0 = 5.0 - # completion = 10 * 2.0 = 20.0 - # total = 25.0 - mock_logger.assert_called_with("πŸͺ™ [LLM Burn] test-model -> In: 5 | Out: 10 | Total: 15 | πŸ’Έ Cost: $25.000000", extra={"color": "\x1b[38;5;208m\x1b[1m"}) + # Test Cloud URL + result = query_telepathic_llm( + model="openrouter/qwen3.5:latest", + url="https://openrouter.ai/api/v1/chat/completions", + system_prompt="sys", + user_prompt="user", + use_local_edge=False + ) + + assert result == "{}" + + def test_vlm_timeout_allows_local_processing(self, monkeypatch): + """ + Localhost (Ollama) connections must have a significantly higher timeout (180s) + so cold starts loading into VRAM don't drop. + """ + def mock_post(*args, **kwargs): + url = kwargs.get("url") or args[0] + timeout = kwargs.get("timeout") + assert timeout == 180, f"Expected 180s extended timeout for localhost, got {timeout}" + + mock_resp = MagicMock() + mock_resp.status_code = 200 + # For ollama/vllm mimic structure + mock_resp.json.return_value = {"response": '{"clicked": "true"}'} + return mock_resp + + monkeypatch.setattr(requests, "post", mock_post) + + # Test Local URL + result = query_telepathic_llm( + model="qwen3.5:latest", + url="http://localhost:11434/api/generate", + system_prompt="sys", + user_prompt="user", + use_local_edge=False + ) + + assert result == '{"clicked": "true"}' + + def test_local_edge_override_applies_timeout(self, monkeypatch): + """ + If use_local_edge=True is set, it overrides the URL to localhost and MUST apply 180s. + """ + def mock_post(*args, **kwargs): + timeout = kwargs.get("timeout") + assert timeout == 180, "Expected 180s extended timeout when forced to local edge" + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"response": '{"edge": "true"}'} + return mock_resp + + monkeypatch.setattr(requests, "post", mock_post) + + result = query_telepathic_llm( + model="openrouter", + url="https://openrouter...", + system_prompt="sys", + user_prompt="user", + use_local_edge=True # Force local + ) + + assert result == '{"edge": "true"}' diff --git a/tests/unit/test_profile_interaction_edges.py b/tests/unit/test_profile_interaction_edges.py new file mode 100644 index 0000000..258f4d3 --- /dev/null +++ b/tests/unit/test_profile_interaction_edges.py @@ -0,0 +1,91 @@ +import pytest +from unittest.mock import MagicMock +from GramAddict.core.telepathic_engine import TelepathicEngine +from GramAddict.core.q_nav_graph import QNavGraph + + +class TestProfileInteractionSync: + """ + TDD Tests: Reproduces the 'Already Followed -> Favorites' and 'No Story Ring' + bugs from 2026-04-17 live run. + """ + + def setup_method(self): + self.engine = TelepathicEngine() + self.mock_device = MagicMock() + self.mock_device.deviceV2 = MagicMock() + self.mock_device.app_id = "com.instagram.android" + self.mock_device._get_current_app.return_value = "com.instagram.android" + self.nav_graph = QNavGraph(self.mock_device) + + def test_prevent_tapping_following_button(self): + """ + If the intent is to follow, but the matching node says 'following' or 'gefolgt', + the engine must skip the click to prevent opening the Favorites/Mute bottom sheet. + """ + # Simulate a profile where the user is already followed + viable_nodes = [{ + "semantic_string": "id context: 'profile header user action follow button', text: 'Following'", + "x": 500, "y": 600, + "width": 100, "height": 50, + "area": 5000, + "class_name": "android.widget.Button", + "resource_id": "com.instagram.android:id/button", + "original_attribs": {"text": "Following"} + }] + + # Test vector-based matching fallback + self.engine._blacklist = {} + + # Mock the extraction to avoid needing valid complex XML + self.engine._extract_semantic_nodes = MagicMock(return_value=viable_nodes) + self.engine._structural_sanity_check = MagicMock(return_value=True) + self.engine._is_instagram_context = MagicMock(return_value=True) + + result = self.engine.find_best_node("", "tap follow button on profile", device=self.mock_device) + + # We must intercept it in TelepathicEngine before VLM is called + # Wait, find_best_node falls back to VLM if vector score is low. + # But if we inject it into memory, it triggers stage 1 + self.engine._memory = { + "tap follow button on profile": ["id context: 'profile header user action follow button', text: 'Following'"] + } + TelepathicEngine._instance = self.engine + + result = self.engine.find_best_node("", "tap follow button on profile", device=self.mock_device) + + assert result is not None, "Engine should return a skip result, not None" + assert result.get("skip") is True, "Must return skip: True to prevent Favorites menu from opening" + assert result.get("semantic") == "already_followed" + + + def test_story_ring_not_present_skips_click(self): + """ + If no story ring is explicitly in the XML, bot_flow should not execute + the transition (simulated here by checking our XML evaluation logic). + """ + xml_without_story = ''' + + + + + ''' + + has_story = "reel_ring" in xml_without_story or "unseen story" in xml_without_story.lower() or "story von" in xml_without_story.lower() + + assert has_story is False, "Logic falsely identified a story when there is only a generic profile picture" + + def test_story_ring_present_allows_click(self): + """ + If a story ring is present, the logic should allow the interaction. + """ + xml_with_story = ''' + + + + + ''' + + has_story = "reel_ring" in xml_with_story or "unseen story" in xml_with_story.lower() or "story von" in xml_with_story.lower() + + assert has_story is True, "Logic failed to identify active story ring" diff --git a/tests/unit/test_profile_interaction_sync.py b/tests/unit/test_profile_interaction_sync.py index 8fb8b2e..735d935 100644 --- a/tests/unit/test_profile_interaction_sync.py +++ b/tests/unit/test_profile_interaction_sync.py @@ -19,6 +19,7 @@ def test_profile_grid_sync_delay_after_follow(): It now tracks the autonomous QNavGraph calls. """ mock_device = MagicMock() + mock_device.deviceV2.dump_hierarchy.return_value = "dummy hierarchy" mock_configs = FakeConfig() mock_session_state = MagicMock(spec=SessionState) diff --git a/tests/unit/test_unfollow_engine.py b/tests/unit/test_unfollow_engine.py new file mode 100644 index 0000000..80f6635 --- /dev/null +++ b/tests/unit/test_unfollow_engine.py @@ -0,0 +1,109 @@ +import pytest +from unittest.mock import MagicMock, call +from GramAddict.core.unfollow_engine import _run_zero_latency_unfollow_loop +import logging + +class TestUnfollowEngine: + + def setup_method(self): + self.mock_device = MagicMock() + self.mock_device.deviceV2 = MagicMock() + self.mock_device.app_id = "com.instagram.android" + self.mock_device._get_current_app.return_value = "com.instagram.android" + self.mock_device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400} + + self.mock_telepathic = MagicMock() + self.mock_dopamine = MagicMock() + self.mock_dopamine.is_app_session_over.return_value = False + self.mock_dopamine.wants_to_change_feed.return_value = False + # default boredom + self.mock_dopamine.boredom = 0.0 + + self.cognitive_stack = { + "telepathic": self.mock_telepathic, + "dopamine": self.mock_dopamine, + } + + self.mock_configs = MagicMock() + self.mock_configs.args.total_unfollows_limit = 50 + + self.mock_session_state = MagicMock() + self.mock_session_state.totalUnfollowed = 0 + self.mock_session_state.check_limit.return_value = False + + self.logger = logging.getLogger("test") + + def test_unfollow_loop_success(self, monkeypatch): + """ + Happy path: Finds 'Following' button -> Clicks it -> Finds 'Unfollow' confirmation -> Clicks it -> Increments totalUnfollowed. + Then dopamine signals change of feed. + """ + def fake_extract_semantic_nodes(xml, intent, **kwargs): + if "Unfollow" in intent: + return [{"semantic_string": "Unfollow Confirmation", "x": 500, "y": 1500, "bounds": "[100,200]", "skip": False}] + else: + return [{"semantic_string": "Following Button", "x": 900, "y": 600, "bounds": "[100,200]", "skip": False}] + + self.mock_telepathic._extract_semantic_nodes.side_effect = fake_extract_semantic_nodes + self.mock_dopamine.wants_to_change_feed.side_effect = [True] + + # Patch local imports inside the bot_flow namespace since they are locally imported + mock_sleep = MagicMock() + mock_click = MagicMock() + import GramAddict.core.bot_flow + monkeypatch.setattr(GramAddict.core.bot_flow, "sleep", mock_sleep) + monkeypatch.setattr(GramAddict.core.bot_flow, "_humanized_click", mock_click) + + result = _run_zero_latency_unfollow_loop( + self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack + ) + + # Verify result and clicks + assert result == "BOREDOM_CHANGE_FEED" + assert self.mock_session_state.totalUnfollowed == 1 + + # Assert humanized clicks were logged correctly + assert mock_click.call_count == 2 + mock_click.assert_has_calls([ + call(self.mock_device, 900, 600), + call(self.mock_device, 500, 1500) + ]) + + def test_unfollow_loop_scrolls_if_empty(self, monkeypatch): + """ + End of list path: If no following buttons are found, it should scroll _humanized_scroll_down. + After 5 failed scrolls, it should gracefully return BOREDOM_CHANGE_FEED without crashing. + """ + # Always return empty nodes + self.mock_telepathic._extract_semantic_nodes.return_value = [] + + # Patch the imported _humanized_scroll_down so we can track it + mock_scroll = MagicMock() + import GramAddict.core.unfollow_engine + monkeypatch.setattr(GramAddict.core.unfollow_engine, "_humanized_scroll_down", mock_scroll) + + result = _run_zero_latency_unfollow_loop( + self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack + ) + + assert result == "BOREDOM_CHANGE_FEED" + # It should scroll exactly 6 times (failed_scrolls > 5) + assert mock_scroll.call_count == 6 + assert self.mock_session_state.totalUnfollowed == 0 + + def test_unfollow_loop_respects_limits(self): + """ + Limit protection: If session limit is reached at the start, abort immediately. + No clicks or UI dumps should occur. + """ + # Tell session_state that UNFOLLOWS limit is hit + self.mock_session_state.check_limit.return_value = (True, "Unfollow limit reached") + + result = _run_zero_latency_unfollow_loop( + self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack + ) + + assert result == "BOREDOM_CHANGE_FEED" + self.mock_device.deviceV2.dump_hierarchy.assert_not_called() + self.mock_device.deviceV2.click.assert_not_called() + self.mock_session_state.totalUnfollowed = 0 diff --git a/tests/unit/test_verify_success_reels.py b/tests/unit/test_verify_success_reels.py new file mode 100644 index 0000000..a92d834 --- /dev/null +++ b/tests/unit/test_verify_success_reels.py @@ -0,0 +1,73 @@ +import pytest +from GramAddict.core.telepathic_engine import TelepathicEngine + + +class TestVerifySuccessGridReels: + """ + TDD Tests: Reproduces Bug 1 from the 2026-04-17 09:56 run. + + The Grid Fast-Path correctly clicks an explore grid item, the UI changes + (a Reel opens), but verify_success() returns False because it only looks + for row_feed_* markers which don't exist in Reel views. + """ + + def setup_method(self): + self.engine = TelepathicEngine() + # Simulate a click context so verify_success has something to check against + TelepathicEngine._last_click_context = { + "intent": "first image in explore grid", + "semantic_string": "id context: 'image button'", + "x": 178, "y": 558, + "timestamp": 0 + } + + def test_reel_view_accepted_as_valid_grid_result(self): + """A Reel opening after a grid tap must be accepted as success.""" + # Simulate post-click XML containing Reel markers but NO feed markers + reel_xml = """ + + + + + + + """ + result = self.engine.verify_success("first image in explore grid", reel_xml) + assert result is True, "verify_success rejected a valid Reel view opened from grid tap" + + def test_normal_feed_post_still_accepted(self): + """A normal feed post opening after a grid tap must still be accepted.""" + feed_xml = """ + + + + + + """ + result = self.engine.verify_success("first image in explore grid", feed_xml) + assert result is True, "verify_success rejected a valid feed post opened from grid tap" + + def test_explore_grid_still_visible_is_failure(self): + """If the grid is still showing (no post opened), verify must fail.""" + explore_xml = """ + + + + + + + """ + result = self.engine.verify_success("first image in explore grid", explore_xml) + assert result is False, "verify_success accepted the explore grid as a post view" + + def test_profile_grid_reel_accepted(self): + """Profile grid β†’ Reel must also be accepted.""" + TelepathicEngine._last_click_context["intent"] = "first image post in profile grid" + reel_xml = """ + + + + + """ + result = self.engine.verify_success("first image post in profile grid", reel_xml) + assert result is True, "verify_success rejected a Reel opened from profile grid"