This commit is contained in:
2026-04-17 00:44:41 +02:00
parent fa1d01527d
commit 89f14463c5
42 changed files with 2452 additions and 314 deletions

View File

@@ -5,7 +5,7 @@ from colorama import Fore, Style
logger = logging.getLogger(__name__)
BENCHMARKS_FILE = os.path.join(os.path.dirname(__file__), "llm_benchmarks.json")
BENCHMARKS_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "benchmarks", "data", "llm_benchmarks.json")
def check_model_benchmarks(configs):
"""

View File

@@ -164,7 +164,7 @@ def start_bot(**kwargs):
available_targets.append("ReelsFeed")
if getattr(configs.args, "stories", None):
available_targets.append("StoriesFeed")
if getattr(configs.args, "total_unfollows_limit", 0):
if getattr(configs.args, "smart_unfollow", False):
available_targets.append("FollowingList")
if not getattr(configs.args, "disable_ai_messaging", False):
available_targets.append("MessageInbox")
@@ -207,15 +207,33 @@ def start_bot(**kwargs):
result = _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack, is_reels=is_reels)
# Evaluate outcome from loop
if result == "BOREDOM_CHANGE_FEED":
if result in ("BOREDOM_CHANGE_FEED", "FEED_EXHAUSTED"):
if result == "FEED_EXHAUSTED":
logger.info(f"✅ Finished watching {current_target}. Removing from this session's options.")
if current_target in available_targets:
available_targets.remove(current_target)
# Find new targets excluding the current one
available_targets_copy = [t for t in available_targets if t != current_target]
current_target = secrets.choice(available_targets_copy) if available_targets_copy else current_target
logger.info(f"🧠 [Free Will] Spontaneous desire changed. Switching to {current_target}.")
if not available_targets_copy:
logger.info("🧠 Session natural conclusion: All desired feeds visited and exhausted.")
break # No more feeds to visit!
current_target = secrets.choice(available_targets_copy)
if result == "BOREDOM_CHANGE_FEED":
logger.info(f"🧠 [Free Will] Spontaneous desire changed. Switching to {current_target}. (Restoring Dopamine)")
dopamine.boredom = max(0.0, dopamine.boredom * 0.2) # Reset boredom so we actually execute the new feed!
else:
logger.info(f"🧠 [Free Will] Moving on to {current_target}.")
elif result == "CONTEXT_LOST":
logger.warning(f"⚠️ Context was lost in {current_target}. Forcing re-navigation to recover.")
nav_graph.current_state = "UNKNOWN"
continue
else:
logger.info(f"Session concluding due to state: {result}")
break # Session over or unhandled state
else:
logger.error(f"Aborting target {current_target} due to navigation failure.")
@@ -462,37 +480,17 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod,
except:
count = 1
telepathic = TelepathicEngine.get_instance()
xml_dump = device.deviceV2.dump_hierarchy()
story_ring_node = telepathic.find_best_node(xml_dump, "Profile picture avatar story ring", device=device)
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(device)
if story_ring_node and not story_ring_node.get("skip"):
_humanized_click(device, story_ring_node["x"], story_ring_node["y"], sleep_mod=sleep_mod)
sleep(random.uniform(2.5, 4.0) * sleep_mod)
# ── Post-Click Verification: Did the story viewer actually open? ──
post_xml = device.deviceV2.dump_hierarchy()
story_indicators = ["reel_viewer", "story_viewer", "reel_pager", "camera_settings", "story_media"]
story_opened = any(ind in post_xml for ind in story_indicators)
if not story_opened:
# Fallback: if profile elements (header/bio) disappeared, something opened
story_opened = "profile_header" in xml_dump and "profile_header" not in post_xml
if story_opened:
telepathic.confirm_click("Profile picture avatar story ring")
logger.info(f"📸 [Story] Viewing @{username}'s story ({count} times)...")
for i in range(count):
sleep(random.uniform(2.0, 5.0) * sleep_mod)
if i < count - 1:
_humanized_click(device, int(w * 0.9), int(h * 0.5), sleep_mod=sleep_mod)
device.deviceV2.press("back")
sleep(random.uniform(1.0, 2.0) * sleep_mod)
else:
telepathic.reject_click("Profile picture avatar story ring")
logger.warning(f"⚠️ [Story] Click did NOT open story viewer for @{username}. Learning from failure.")
device.deviceV2.press("back")
sleep(random.uniform(0.5, 1.0))
if 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)
if i < count - 1:
_humanized_click(device, int(w * 0.9), int(h * 0.5), sleep_mod=sleep_mod)
device.deviceV2.press("back")
sleep(random.uniform(1.0, 2.0) * sleep_mod)
# Random Follow
follow_pct = float(getattr(configs.args, "follow_percentage", 0)) / 100.0
@@ -501,23 +499,16 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod,
follow_pct = 0.0
if random.random() < follow_pct:
xml_dump = device.deviceV2.dump_hierarchy()
telepathic = TelepathicEngine.get_instance()
follow_btn = telepathic.find_best_node(xml_dump, "tap follow button on profile", device=device)
if follow_btn and not follow_btn.get("skip"):
_humanized_click(device, follow_btn["x"], follow_btn["y"], sleep_mod=sleep_mod)
sleep(random.uniform(2.0, 4.0) * sleep_mod)
# ── Post-Click Verification: Did follow state change? ──
post_xml = device.deviceV2.dump_hierarchy().lower()
# If the button now says "Following", "Abonniert", or "Requested"
if any(word in post_xml for word in ["following", "abonniert", "requested"]):
telepathic.confirm_click("tap follow button on profile")
logger.info(f"🤝 [Deep Interaction] Followed @{username}")
session_state.totalFollowed[username] = 1
else:
telepathic.reject_click("tap follow button on profile")
logger.warning(f"⚠️ [Follow] Click did not change follow state for @{username}. Learning from failure.")
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(device)
if nav_graph._execute_transition("tap_follow_button"):
logger.info(f"🤝 [Deep Interaction] Followed @{username}")
session_state.totalFollowed[username] = 1
# ── TDD Sync Guard: Profile Animations ──
# Provide buffer for follow-related animations/menus to close before parsing the grid
sleep(random.uniform(1.8, 3.2) * sleep_mod)
# Grid Likes
likes_pct = float(getattr(configs.args, "likes_percentage", 0)) / 100.0
@@ -532,42 +523,24 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod,
except:
count = 1
xml_dump = device.deviceV2.dump_hierarchy()
telepathic = TelepathicEngine.get_instance()
first_post = telepathic.find_best_node(xml_dump, "first image post in profile grid", device=device)
if first_post and not first_post.get("skip"):
_humanized_click(device, first_post["x"], first_post["y"], sleep_mod=sleep_mod)
sleep(random.uniform(2.5, 4.5) * sleep_mod)
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(device)
if nav_graph._execute_transition("tap_grid_first_post"):
logger.info(f"❤️ [Deep Interaction] Opening grid to drop {count} likes on @{username}...")
# ── Post-Click Verification: Did a post open? ──
post_xml = device.deviceV2.dump_hierarchy()
# Post view has like buttons or media groups
post_opened = any(ind in post_xml for ind in ["button_like", "button_comment", "media_group"])
if not post_opened:
# If grid is gone, something opened
post_opened = "profile_grid" in xml_dump and "profile_grid" not in post_xml
if post_opened:
telepathic.confirm_click("first image post in profile grid")
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}")
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)}")
sleep(random.uniform(1.5, 3.0) * sleep_mod)
device.deviceV2.press("back")
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}")
sleep(random.uniform(1.0, 2.0) * sleep_mod)
else:
telepathic.reject_click("first image post in profile grid")
logger.warning(f"⚠️ [Grid] Click did not open post for @{username}. Learning from failure.")
device.deviceV2.press("back")
sleep(random.uniform(0.5, 1.0))
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)}")
sleep(random.uniform(1.5, 3.0) * sleep_mod)
device.deviceV2.press("back")
sleep(random.uniform(1.0, 2.0) * sleep_mod)
# Let the native UI momentum scroll finish just like a human watching the feed
sleep(random.uniform(1.2, 2.0))
@@ -667,11 +640,15 @@ def _detect_ad_structural(context_xml: str) -> bool:
AD_RESOURCE_IDS = {
"com.instagram.android:id/ad_cta_button",
"com.instagram.android:id/clips_single_image_ads_media_content",
"com.instagram.android:id/intent_aware_ad_pivot_container"
}
GENERIC_CTA_IDS = {
"com.instagram.android:id/clips_browser_cta",
"com.instagram.android:id/universal_cta_description_layout",
"com.instagram.android:id/universal_cta_text",
"com.instagram.android:id/intent_aware_ad_pivot_container"
}
AD_CTA_WORDS = {"install", "learn more", "shop now", "sign up", "mehr dazu", "jetzt einkaufen", "installieren", "registrieren", "anmelden", "download", "herunterladen", "get offer", "abonnieren", "subscribe"}
try:
root = ET.fromstring(context_xml)
@@ -682,6 +659,14 @@ def _detect_ad_structural(context_xml: str) -> bool:
if res_id in AD_RESOURCE_IDS:
return True
# 1.5 Generic CTAs require text checking to avoid flagging 'Use template' or 'Original audio'
if res_id in GENERIC_CTA_IDS:
text = node.attrib.get("text", "").strip().lower()
desc = node.attrib.get("content-desc", "").strip().lower()
combined = text + " " + desc
if any(w in combined for w in AD_CTA_WORDS):
return True
# 2. Secondary Label Exact Match
if res_id == "com.instagram.android:id/secondary_label":
text = node.attrib.get("text", "").strip().lower()
@@ -793,9 +778,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta
logger.info("🎬 [StoriesFeed] Session completed naturally.")
device.deviceV2.press("back")
return "SESSION_OVER"
return "FEED_EXHAUSTED"
def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, job_target, cognitive_stack, is_reels=False):
"""
The ultra-fast autonomous Free Will loop.
@@ -873,28 +856,22 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
sleep(random.uniform(0.1, 0.4) * sleep_mod)
continue
# ── Boredom Engine (Notifications / DMs) ──
if random.random() < 0.03:
logger.info("🥱 [Boredom] Checking something else (Notifications/DMs) for a second...", extra={"color": f"{Fore.MAGENTA}"})
# Activity tab usually has content-desc "Activity" or resource-id "newsfeed_tab"
newsfeed_tab = device.deviceV2(resourceId="com.instagram.android:id/newsfeed_tab")
direct_tab = device.deviceV2(descriptionContains="Messaging")
# Fallbacks:
if not direct_tab.exists: direct_tab = device.deviceV2(descriptionContains="Message")
if random.random() < 0.5 and newsfeed_tab.exists:
newsfeed_tab.click()
elif direct_tab.exists:
direct_tab.click()
# ── 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)
sleep(random.uniform(3.0, 6.0) * sleep_mod)
# Return to feed natively
feed_tab = device.deviceV2(resourceId="com.instagram.android:id/feed_tab")
if feed_tab.exists:
feed_tab.click()
else:
device.deviceV2.press("back")
# 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()
@@ -988,11 +965,10 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
consecutive_ads += 1
if consecutive_ads >= 3:
logger.warning("📺 [Anti-Stuck] Stuck on ad! Executing aggressive mechanical drag.", extra={"color": f"{Fore.RED}"})
# Massive slow drag from top to bottom (finger down to up -> screen moves down? No we want to go down in feed so finger moves UP)
# h * 0.8 to h * 0.1
# 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.swipe(w // 2, int(h * 0.8), w // 2, int(h * 0.2), duration=0.8)
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)...")
@@ -1053,7 +1029,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):
if nav_graph._execute_transition("tap_post_username", zero_engine) 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)
@@ -1091,7 +1067,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
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):
if nav_graph._execute_transition("tap_post_username", zero_engine) is True:
sleep(random.uniform(1.2, 2.5) * sleep_mod)
# Extract context
@@ -1134,7 +1110,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
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 not success:
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)
@@ -1163,11 +1139,18 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
logger.info("💬 [Interaction] Entering Comment Sheet for deep engagement...")
success = nav_graph._execute_transition("tap_comment_button", zero_engine)
if success:
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()
# 🛡️ [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"]):
logger.warning("❌ [Ambiguity Guard] Transition reported success, but Comment Sheet markers not found in UI. Bailing engagement.")
did_interact = False
continue
import xml.etree.ElementTree as ET
existing_comments = []
comment_nodes = []
@@ -1285,9 +1268,9 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
from GramAddict.core.llm_provider import query_llm
from GramAddict.core.stealth_typing import ghost_type
model = getattr(configs.args, "ai_condenser_model", "google/gemini-2.5-flash-lite-preview")
url = getattr(configs.args, "ai_condenser_url", "https://openrouter.ai/api/v1/chat/completions")
response_dict = query_llm(url=url, model=model, prompt=prompt, format_json=False)
model = getattr(configs.args, "ai_condenser_model", "llama3.2:1b")
url = getattr(configs.args, "ai_condenser_url", "http://localhost:11434/api/generate")
response_dict = query_llm(url=url, model=model, prompt=prompt, format_json=False, timeout=45)
if response_dict and "response" in response_dict:
clean_comment = response_dict["response"].strip().strip('"').strip("'")
@@ -1375,7 +1358,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
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:
if success is True:
sleep(random.uniform(1.8, 3.5) * sleep_mod)
telepathic = TelepathicEngine.get_instance()
xml_dump = device.deviceV2.dump_hierarchy()
@@ -1443,7 +1426,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session
darwin.evaluate_session_end(duration_minutes, followers_gained)
logger.info("🏁 [Drive] Feed loop terminated. Session over.")
return "SESSION_OVER"
return "FEED_EXHAUSTED"
def _run_zero_latency_search_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack):

View File

@@ -22,24 +22,38 @@ class VLMCompilerEngine:
logger.warning(f"🧠 [Compiler Engine] Deterministic heuristic failed for: '{intent_description}'. Synthesizing new rule...", extra={"color": "\x1b[1m\x1b[35m"})
args = getattr(self.device, "args", None)
model = getattr(args, "ai_telepathic_model", "google/gemini-3.1-flash-lite-preview") if args else "google/gemini-3.1-flash-lite-preview"
url = getattr(args, "ai_telepathic_url", "https://openrouter.ai/api/v1/chat/completions") if args else "https://openrouter.ai/api/v1/chat/completions"
model = getattr(args, "ai_telepathic_model", "llama3.2:1b") if args else "llama3.2:1b"
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate") if args else "http://localhost:11434/api/generate"
use_local = "11434" in url or "localhost" in url
simplified_xml = self._simplify_xml(context_xml)
# --- Model Trust Logging ---
from GramAddict.core.benchmark_guard import BENCHMARKS_FILE
trust_log = f"Using {model}"
try:
if os.path.exists(BENCHMARKS_FILE):
with open(BENCHMARKS_FILE, "r") as f:
bench_data = json.load(f).get("models", {}).get(model, {})
if bench_data:
score = bench_data.get("telepathic_score", 0)
passed = "PASS" if bench_data.get("passed_all", False) else "FAIL"
unsuitable = bench_data.get("is_unsuitable", False)
trust_level = "HIGH" if score >= 80 and not unsuitable else "MEDIUM" if score >= 50 and not unsuitable else "LOW/UNSAFE"
trust_log += f" [Benchmark: {score}/100 | {passed} | Trust: {trust_level}]"
if unsuitable:
logger.error(f"⛔ [Safety Alert] {model} is marked as UNSUITABLE for this task!")
except Exception:
pass
logger.info(f"🧠 [Compiler] Intent: '{intent_description}' -> {trust_log}")
# ---------------------------
system_prompt = (
"You write Python regex rules to find Android UI elements. "
"Given UI XML, find the element matching the intent. "
"Generate a regex pattern to match its resource-id.\n\n"
"OUTPUT FORMAT (JSON only):\n"
"{\"rule_type\": \"regex\", \"target_attribute\": \"resource-id\", "
"\"pattern\": \".*your_regex.*\", \"confidence\": 0.95, "
"\"reasoning\": \"brief explanation\"}\n\n"
"RULES:\n"
"- ONLY use rule_type='regex'. NEVER use xpath.\n"
"- Target resource-id for dynamic elements, not text or usernames.\n"
"- Make patterns globally reusable, not hardcoded to specific content."
"You are a regex engineering expert for Android UI. Identify the most stable resource-id regex for the intent.\n"
"Rules:\n"
"1. Output ONLY a raw JSON object.\n"
"2. NO markdown, NO triple backticks.\n"
"3. Format: {\"rule_type\": \"regex\", \"target_attribute\": \"resource-id\", \"pattern\": \".*regex.*\", \"confidence\": 0.95, \"reasoning\": \"string\"}"
)
user_prompt = f"TARGET INTENT: {intent_description}\n\nUI XML:\n{simplified_xml[:2000]}"
@@ -55,10 +69,32 @@ class VLMCompilerEngine:
use_local_edge=use_local
)
if res_text and res_text.startswith("```"):
if not res_text:
logger.error("Compiler LLM returned empty response.")
return None
if "```json" in res_text:
res_text = res_text.split("```json")[1].split("```")[0].strip()
elif res_text.startswith("```"):
res_text = "\n".join(res_text.strip().split("\n")[1:-1])
decision = json.loads(res_text) if res_text else {}
try:
decision = json.loads(res_text)
except json.JSONDecodeError:
logger.error(f"Compiler LLM returned invalid JSON: {res_text[:100]}...")
return None
# If LLM returned a list, take the first item if it's a dict
if isinstance(decision, list):
if len(decision) > 0 and isinstance(decision[0], dict):
decision = decision[0]
else:
logger.error(f"Compiler LLM returned unexpected list format: {decision}")
return None
if not isinstance(decision, dict):
logger.error(f"Compiler LLM returned non-object response: {type(decision)}")
return None
pattern = decision.get('pattern')
if not pattern:

View File

@@ -157,10 +157,10 @@ class Config:
self.parser.add_argument("--speed-multiplier", help="Speed multiplier", default="1.0")
# AI Model Configuration (centralized — no hardcoded model names anywhere)
self.parser.add_argument("--ai-model", "--ai-text-model", help="Primary LLM model (OpenRouter or Ollama)", default="google/gemini-2.5-flash-lite-preview")
self.parser.add_argument("--ai-model-url", "--ai-text-url", help="Primary LLM endpoint URL", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-telepathic-model", help="Text-based model for Telepathic Engine Fallbacks", default="google/gemini-3.1-flash-lite-preview")
self.parser.add_argument("--ai-telepathic-url", help="Telepathic model endpoint URL", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-model", "--ai-text-model", help="Primary LLM model (OpenRouter or Ollama)", default="llama3.2:1b")
self.parser.add_argument("--ai-model-url", "--ai-text-url", help="Primary LLM endpoint URL", default="http://localhost:11434/api/generate")
self.parser.add_argument("--ai-telepathic-model", help="Text-based model for Telepathic Engine Fallbacks", default="llama3.2:1b")
self.parser.add_argument("--ai-telepathic-url", help="Telepathic model endpoint URL", default="http://localhost:11434/api/generate")
self.parser.add_argument("--ai-fallback-model", "--ai-text-fallback-model", help="Fallback model when primary fails", default="llama3.2:1b")
self.parser.add_argument("--ai-fallback-url", "--ai-text-fallback-url", help="Fallback model endpoint URL", default="http://localhost:11434/api/generate")
self.parser.add_argument("--ai-embedding-model", help="Embedding model for vector operations", default="nomic-embed-text")
@@ -176,8 +176,8 @@ class Config:
self.parser.add_argument("--scrape-profiles", action="store_true", help="Extract and store profile metadata in CRM")
# Phase 10: RAG Comment Learning & Extractor Settings
self.parser.add_argument("--ai-condenser-model", help="LLM used for condensing text/comments", default="google/gemini-2.5-flash-lite-preview")
self.parser.add_argument("--ai-condenser-url", help="URL for the condenser model", default="https://openrouter.ai/api/v1/chat/completions")
self.parser.add_argument("--ai-condenser-model", help="LLM used for condensing text/comments", default="llama3.2:1b")
self.parser.add_argument("--ai-condenser-url", help="URL for the condenser model", default="http://localhost:11434/api/generate")
self.parser.add_argument("--ai-learn-comments", action="store_true", help="Extract and learn from comment sections")
self.parser.add_argument("--ai-learn-niche-posts", action="store_true", help="Learn from niche posts")
self.parser.add_argument("--ai-learn-own-profile", action="store_true", help="Learn from your own profile interactions")
@@ -185,6 +185,9 @@ class Config:
self.parser.add_argument("--ai-vibe", help="The specific vibe to extract from comments (e.g., friendly, controversial)", default="")
self.parser.add_argument("--ai-blacklist-topics", help="Comma-separated topics heavily penalized or skipped", default="")
self.parser.add_argument("--ai-quality-filter", action="store_true", help="Use AI to strictly filter the quality of posts and comments")
self.parser.add_argument("--smart-unfollow", action="store_true", help="Enable agentic decision making for clearing the following list")
self.parser.add_argument("--ai-vision-navigation", action="store_true", help="Capture and send base64 UI screenshots to the LLM for structural element finding")
self.parser.add_argument("--ai-vision-context", action="store_true", help="Capture and send base64 post/DM screenshots to the LLM for contextual semantic generation")
# on first run, we must wait to proceed with loading
if not self.first_run:

View File

@@ -111,6 +111,19 @@ class DarwinEngine(QdrantBase):
if profile["comment_read_dwell"] > 1.0 and resonance > 0.4 and random.random() < 0.3:
if nav_graph and zero_engine:
logger.debug(f" -> Opening comments section for {profile['comment_read_dwell']:.1f}s depth simulation")
# Capture image context of post BEFORE opening comment sheet
b64_img_payload = None
if configs and getattr(configs.args, "ai_vision_context", False):
try:
import base64
raw = device.screenshot()
if raw:
b64_img_payload = [base64.b64encode(raw).decode('utf-8')]
logger.debug("👁️ [Vision Context] Captured post screenshot for True Vision semantic analysis.")
except Exception as e:
logger.warning(f"⚠️ [Vision Context] Failed to capture screenshot: {e}")
success = nav_graph._execute_transition("tap_comment_button", zero_engine)
if success:
# ---- Phase 10: RAG Comment Extraction ----
@@ -121,7 +134,7 @@ class DarwinEngine(QdrantBase):
try:
xml_data = device.deviceV2.dump_hierarchy()
t0 = time.time()
resonance_oracle.extract_and_learn_comments(xml_data, configs, author=username or "unknown")
resonance_oracle.extract_and_learn_comments(xml_data, configs, author=username or "unknown", images_b64=b64_img_payload)
t1 = time.time()
remaining_sleep = profile["comment_read_dwell"] - (t1 - t0)
if remaining_sleep > 0:

View File

@@ -148,7 +148,18 @@ class DeviceFacade:
@adb_retry()
def _get_current_app(self):
return self.deviceV2.app_current().get("package")
"""
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.
"""
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")
return pkg
@adb_retry()
def find(self, **kwargs):

View File

@@ -110,4 +110,4 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
if failed_attempts > 2:
return "CONTEXT_LOST"
return "SESSION_OVER"
return "FEED_EXHAUSTED"

View File

@@ -84,7 +84,7 @@ def query_llm(
images_b64: Optional[List[str]] = None,
system: Optional[str] = None,
format_json: bool = False,
timeout: int = 60,
timeout: int = 180,
fallback_model: Optional[str] = None,
fallback_url: Optional[str] = None
) -> Optional[dict]:
@@ -191,12 +191,14 @@ def query_llm(
return {"response": content}
else:
# Ollama returns response
content = resp_json.get("response", "")
# Ollama returns response OR thinking (for reasoning models)
content = resp_json.get("response") or resp_json.get("thinking") or ""
if format_json:
extracted = extract_json(content)
if not extracted:
raise ValueError(f"Ollama returned non-JSON content when JSON was expected: {content[:100]}...")
# Log more context if JSON extraction fails
logger.debug(f"Ollama raw content (for JSON extraction): {content[:200]}...")
raise ValueError(f"Ollama returned non-JSON content when JSON was expected.")
resp_json["response"] = extracted
return resp_json
@@ -227,12 +229,17 @@ def query_llm(
f_model = f_model or "llama3.2:1b"
f_url = f_url or "http://localhost:11434/api/generate"
else:
f_model = f_model or "google/gemini-2.5-flash-lite-preview"
f_url = f_url or "https://openrouter.ai/api/v1/chat/completions"
f_model = f_model or "llama3.2:1b"
f_url = f_url or "http://localhost:11434/api/generate"
# Circuit Breaker: If fallback is identical to primary, don't waste time retrying
if f_model == model and f_url == url:
logger.warning(f"⚠️ [Circuit Breaker] Fallback AI matches Primary AI ({model}). Skipping redundant retry.")
return None
query_llm._is_fallback = True
try:
logger.warning(f"Primary AI ({model}) failed or returned garbage. Attempting fallback to {f_model}...")
logger.warning(f"🔄 Primary AI ({model}) failed. Attempting fallback to {f_model}...")
return query_llm(
url=f_url,
model=f_model,
@@ -252,12 +259,14 @@ def query_telepathic_llm(
system_prompt: str,
user_prompt: str,
temperature: float = 0.0,
use_local_edge: bool = False
use_local_edge: bool = False,
images_b64: Optional[List[str]] = None
) -> str:
"""
Routes UI Telepathic requests purely based on textual interpretation of the screen's XML nodes.
If use_local_edge is manually enabled, routes to localhost:11434.
Otherwise honors the provided URL and model (e.g. OpenRouter).
Passes optional Base64 screenshots (images_b64) if ai-vision-navigation is enabled.
"""
target_url = url
target_model = model
@@ -277,7 +286,7 @@ def query_telepathic_llm(
url=target_url,
model=target_model,
prompt=user_prompt,
images_b64=None,
images_b64=images_b64,
system=system_prompt,
format_json=True
)

View File

@@ -109,6 +109,13 @@ class QNavGraph:
self.current_state = "HomeFeed"
return self.navigate_to(target_state, zero_engine, recovery_attempts=recovery_attempts + 1)
else:
# NEW: Attempt Back-out recovery if we are in UNKNOWN and direct tap failed
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)
# 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
if path is None:
@@ -168,83 +175,117 @@ class QNavGraph:
return None
def _execute_transition(self, action: str, zero_engine) -> bool:
def _execute_transition(self, action: str, zero_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()
context_xml = self.device.deviceV2.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)
# Re-acquire context after clearing obstacle
for attempt in range(max_retries + 1):
context_xml = self.device.deviceV2.dump_hierarchy()
# We phrase the action as an intent for the semantic engine
# e.g. "tap_explore_tab" -> "tap explore tab"
# We add some common synonyms for Instagram to help the vector engine
intent_map = {
# Navigation (Bottom Bar) — aligned with fast-path keys
"tap_home_tab": "tap home tab",
"tap_explore_tab": "tap explore tab",
"tap_profile_tab": "tap profile tab",
"tap_reels_tab": "tap reels tab",
"tap_create_tab": "tap create post tab",
# Post Interaction — aligned with fast-path keys
"tap_like_button": "tap like button",
"tap_comment_button": "tap comment button",
"tap_post_username": "tap post username",
"tap_share_button": "tap share button",
"tap_save_button": "tap save button",
# Grid & Profile
"tap_explore_grid_item": "first image in explore grid",
"tap_story_tray_item": "profile picture avatar story ring",
"tap_follow_button": "tap follow button on profile",
"tap_grid_first_post": "first image post in profile grid",
}
intent_description = intent_map.get(action, action.replace("_", " "))
# ── 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)
# Re-acquire context after clearing obstacle
context_xml = self.device.deviceV2.dump_hierarchy()
# 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)
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
current_app = self.device._get_current_app()
if current_app != self.device.app_id:
logger.warning(f"⚠️ [Context Lost] Currently in '{current_app}', expected '{self.device.app_id}'. Transition '{action}' aborted.")
return "CONTEXT_LOST"
return False
if best_node.get("skip"):
logger.info(f"⏭️ Skipping physical tap for '{action}' (Semantic Fast-Path indicated state already fulfilled)")
return True
# We phrase the action as an intent for the semantic engine
# e.g. "tap_explore_tab" -> "tap explore tab"
# We add some common synonyms for Instagram to help the vector engine
intent_map = {
# Navigation (Bottom Bar) — aligned with fast-path keys
"tap_home_tab": "tap home tab",
"tap_explore_tab": "tap explore tab",
"tap_profile_tab": "tap profile tab",
"tap_reels_tab": "tap reels tab",
"tap_create_tab": "tap create post tab",
# Post Interaction — aligned with fast-path keys
"tap_like_button": "tap like button",
"tap_comment_button": "tap comment button",
"tap_post_username": "tap post username",
"tap_share_button": "tap share button",
"tap_save_button": "tap save button",
# Grid & Profile
"tap_explore_grid_item": "first image in explore grid",
"tap_story_tray_item": "profile picture avatar story ring",
"tap_follow_button": "tap follow button on profile",
"tap_grid_first_post": "first image post in profile grid",
"tap_back": "tap back button icon arrow",
"tap_message_icon": "tap direct message icon inbox",
"tap_newsfeed_tab": "tap activity heart icon notifications",
}
intent_description = intent_map.get(action, action.replace("_", " "))
source_tag = best_node.get("source", "telepathic").replace("_", " ").title()
logger.info(f"QNavGraph executing transition '{action}' via [{source_tag}] (Score: {best_node.get('score', 1.0):.3f})")
# 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)
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
current_app = self.device._get_current_app()
if current_app != self.device.app_id:
logger.warning(f"⚠️ [Context Lost] Currently in '{current_app}', expected '{self.device.app_id}'. Transition '{action}' aborted.")
return "CONTEXT_LOST"
# Try again if within retries, UI might be animating
if attempt < max_retries:
time.sleep(1.0)
continue
return False
# Execute click
self.device.click(obj=best_node)
time.sleep(random.uniform(1.2, 2.5))
if best_node.get("skip") or (best_node.get("selected") and "tab" in action):
logger.info(f"⏭️ Skipping physical tap for '{action}' (Semantic Fast-Path indicated state already fulfilled)")
return True
source_tag = best_node.get("source", "telepathic").replace("_", " ").title()
logger.info(f"QNavGraph executing transition '{action}' via [{source_tag}] (Score: {best_node.get('score', 1.0):.3f})")
# Execute click
self.device.click(obj=best_node)
time.sleep(random.uniform(1.2, 2.5))
# ── Post-Click Verification: Did it work? ──
post_click_xml = self.device.deviceV2.dump_hierarchy()
# 1. Semantic Verification (Hardened)
is_verified = engine.verify_success(intent_description, post_click_xml)
# 2. UI Change Verification (Fallback/Navigation)
ui_changed = post_click_xml != context_xml
# ── Post-Click Verification: Did the screen change? ──
post_click_xml = self.device.deviceV2.dump_hierarchy()
# For navigation, we expect the UI to change or specific markers to appear
# Comparison of XML strings is a good baseline for navigation success
if post_click_xml != context_xml:
engine.confirm_click(intent_description)
return True
else:
logger.warning(f"⚠️ [Nav] Click on '{action}' did not change UI. Learning from failure.")
engine.reject_click(intent_description)
return False
if is_verified and ui_changed:
engine.confirm_click(intent_description)
return True
elif not ui_changed:
logger.warning(f"⚠️ [Nav] Click on '{action}' did not change UI. Learning from failure.")
engine.reject_click(intent_description)
if attempt < max_retries:
logger.info(f"🔄 [Autonomy] UI unchanged. Retrying transition '{action}' ({attempt + 1}/{max_retries})...")
continue
else:
return False
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.")
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
logger.info("🛡️ [Safety Reset] Pressing BACK to clear potential accidental menu/sub-view.")
self.device.deviceV2.press("back")
time.sleep(1.0)
if attempt < max_retries:
logger.info(f"🔄 [Autonomy] Negative learning acquired. Retrying transition '{action}' ({attempt + 1}/{max_retries})...")
continue
else:
return False
return False
def _repair_transition(self, action: str):
"""

View File

@@ -14,7 +14,12 @@ import uuid
logger = logging.getLogger(__name__)
class QdrantBase:
_connection_failed_logged = False
def __init__(self, collection_name, vector_size=768):
if QdrantBase._connection_failed_logged:
self.client = None
return
self.collection_name = collection_name
self.client = None
self._vector_size = vector_size
@@ -52,7 +57,10 @@ class QdrantBase:
)
logger.info(f"Created Qdrant collection '{collection_name}' with dimension {vector_size}.")
except Exception as e:
logger.error(f"Failed to initialize Qdrant memory for '{collection_name}': {e}")
if not QdrantBase._connection_failed_logged:
logger.error(f"Failed to initialize Qdrant memory (likely not running): {e}")
QdrantBase._connection_failed_logged = True
# No debug logs here to prevent noise in --debug mode
self.client = None
@property
@@ -745,6 +753,10 @@ class BannedPathsDB:
MAX_AGE_SECONDS = 7 * 24 * 3600
def __init__(self):
if QdrantBase._connection_failed_logged:
self._client = None
return
self._banned = {} # {goal_hash: set(element_ids)}
self._client = None
self._collection = "gramaddict_banned_paths"

View File

@@ -154,11 +154,12 @@ class ResonanceEngine:
logger.info("✨ [Resonance] NEGATIVE ALIGNMENT. Skipping profile.", extra={"color": f"{Fore.MAGENTA}"})
return False
def extract_and_learn_comments(self, xml_hierarchy: str, configs, author: str = "unknown"):
def extract_and_learn_comments(self, xml_hierarchy: str, configs, author: str = "unknown", images_b64: list = None):
"""
Phase 10: RAG Comment Learning Implementation
Extracts comments from the UI hierarchy, filters them using the assigned VLM against
configured blacklists and vibes, and stores them in Qdrant CommentMemoryDB.
Supports True Vision Context (images_b64) to mathematically align comments to visual aesthetics.
"""
if not configs or not getattr(configs.args, "ai_learn_comments", False):
return
@@ -199,24 +200,34 @@ class ResonanceEngine:
# 2. Filter via VLM Condenser
prompt = (
f"Filter these Instagram comments. Keep ONLY real comments that generally match this vibe: '{vibe}'.\n"
f"Remove comments about: {blacklist}\n"
f"Remove UI junk text (buttons, labels, timestamps).\n\n"
f"Comments:\n{chr(10).join(raw_comments)}\n\n"
"Output a JSON array of matching comment strings. If none match, output []."
f"Evaluate Instagram comments for SPAM. Your only goal is blocking bad topics.\n"
f"VIBE = '{vibe}'\n"
f"BLACKLIST = {blacklist}\n\n"
f"Comments:\n{chr(10).join(['- ' + c for c in raw_comments])}\n\n"
"Return a JSON formatting exactly like this example:\n"
"{\n"
" \"evaluations\": [\n"
" {\"text\": \"love it!\", \"has_blacklist_words\": false, \"keep\": true},\n"
" {\"text\": \"dm me for bitcoin\", \"has_blacklist_words\": true, \"keep\": false}\n"
" ]\n"
"}"
)
model = getattr(configs.args, "ai_condenser_model", "google/gemini-2.5-flash-lite-preview")
url = getattr(configs.args, "ai_condenser_url", "https://openrouter.ai/api/v1/chat/completions")
model = getattr(configs.args, "ai_condenser_model", "llama3.2:1b")
url = getattr(configs.args, "ai_condenser_url", "http://localhost:11434/api/generate")
try:
import json
system = "You are a precise JSON filtering agent."
# Fix: kwargs match query_llm signature EXACTLY to evade TypeError
response_dict = query_llm(url=url, model=model, prompt=prompt, format_json=True)
response_dict = query_llm(url=url, model=model, prompt=prompt, system=system, format_json=True, images_b64=images_b64)
if not response_dict or "response" not in response_dict:
return
response_text = response_dict["response"]
# DEBUG
logger.debug(f"DEBUG CONDENSER RAW: {response_text}")
print(f"DEBUG CONDENSER RAW: {response_text}")
# Parse json gracefully
if type(response_text) is str:
@@ -234,8 +245,19 @@ class ResonanceEngine:
# In case expect_json already returned a parsed list somehow, though extract_json returns str
learned_comments = response_text
# Filter the dict based on evaluations array
if isinstance(learned_comments, dict):
valid_list = []
evals = learned_comments.get("evaluations", [])
for ev in evals:
# Qwen 3.5 correctly identifies 'has_blacklist_words' but hallucinates 'keep': true
has_spam = ev.get("has_blacklist_words", False)
if not has_spam:
valid_list.append(ev.get("text"))
learned_comments = valid_list
if not isinstance(learned_comments, list):
logger.error("🧠 [Comment Learning] Condenser failed to return a JSON list.")
logger.error(f"🧠 [Comment Learning] Condenser failed to return a valid JSON structure: {learned_comments}")
return
if not learned_comments:

View File

@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
# ── Screen Zone Constants (fraction of screen height) ──
# Used for positional sanity checking instead of hardcoded resource-IDs.
STATUS_BAR_ZONE = 0.04 # Top 4% = Android status bar (wifi, battery, clock)
NAV_BAR_ZONE = 0.92 # Bottom 8% = Android nav bar / Instagram bottom tabs
NAV_BAR_ZONE = 0.94 # Bottom 6% = Android nav bar / Instagram bottom tabs
MAX_BUTTON_AREA = 150000 # Buttons/icons should be smaller than this (px²)
MAX_CONTAINER_AREA = 500000 # Anything above this is a full-screen container
@@ -112,8 +112,11 @@ class TelepathicEngine:
# XML Parsing
# ──────────────────────────────────────────────
def _extract_semantic_nodes(self, xml_string: str) -> list[dict]:
"""Parses Android UI XML and extracts clickable/interactive nodes."""
def _extract_semantic_nodes(self, xml_string: str, intent: str = None, threshold: float = 0.0) -> list[dict]:
"""
Parses Android UI XML and extracts clickable/interactive nodes.
If intent and threshold are provided, it filters nodes by semantic score.
"""
nodes = []
try:
clean_xml = re.sub(r'<\?xml.*?\?>', '', xml_string).strip()
@@ -160,24 +163,47 @@ class TelepathicEngine:
center_y = (top + bottom) // 2
width = right - left
height = bottom - top
area = width * height
nodes.append({
node = {
"semantic_string": semantic_string,
"x": center_x,
"y": center_y,
"width": width,
"height": height,
"area": width * height,
"area": area,
"raw_bounds": bounds_str,
"resource_id": res_id,
"class_name": class_name,
"selected": attrib.get('selected', 'false').lower() == 'true',
"original_attribs": {"text": text, "desc": content_desc}
})
}
# Apply structural filter before scoring if intent is provided
if intent:
# We use a default screen_height if we call extract directly with intent
if not self._structural_sanity_check(node, intent, screen_height=2400):
continue
# Compute score for filtering
score = self._compute_quick_score(intent, semantic_string)
if score < threshold:
continue
node["score"] = score
nodes.append(node)
except Exception as e:
logger.error(f"Telepathic XML parsing failed: {e}")
return nodes
def _compute_quick_score(self, intent: str, semantic: str) -> float:
"""Helper for legacy _extract_semantic_nodes filtering."""
intent_words = set(intent.lower().replace("_", " ").split())
semantic_words = set(semantic.lower().replace("_", " ").split())
common = intent_words.intersection(semantic_words)
return len(common) / len(intent_words) if intent_words else 0.0
# ──────────────────────────────────────────────
# Structural Sanity (app-agnostic, no hardcoded IDs)
# ──────────────────────────────────────────────
@@ -191,7 +217,15 @@ class TelepathicEngine:
"""
# 1. Reject massive containers (full-screen views, recycler views)
# UNLESS the intent explicitly targets media
is_media_intent = any(k in intent_description.lower() for k in ["video", "photo", "reel", "media", "post"])
low_intent = intent_description.lower()
is_media_intent = any(k in low_intent for k in ["video", "photo", "reel", "media", "post"])
is_grid_item_intent = any(k in low_intent for k in ["grid", "list", "first", "item", "row"])
# If the intent specifically mentions looking for an item within a grid/list,
# we must block massive parent containers even if the word 'post', 'photo' or 'video' is present.
if is_grid_item_intent:
is_media_intent = False
if node.get("area", 0) > MAX_CONTAINER_AREA and not is_media_intent:
return False
@@ -199,10 +233,59 @@ class TelepathicEngine:
if node.get("y", 0) < screen_height * STATUS_BAR_ZONE:
return False
# 3. Reject nodes with zero area (invisible)
if node.get("area", 0) == 0:
# 3. Reject nodes in the Navigation Bar zone (bottom 6% - adjusted for accuracy)
# UNLESS the intent is explicitly about navigation tabs, profile stats, OR popup modals
nav_keywords = ["tab", "navigation", "explore tab", "reels tab", "profile tab", "home tab", "message tab", "following", "follower", "followers"]
modal_keywords = ["dismiss", "ok", "cancel", "accept", "allow", "deny", "action", "obstacle", "popup"]
low_intent = intent_description.lower()
is_nav_intent = any(k in low_intent for k in nav_keywords)
is_modal_intent = any(k in low_intent for k in modal_keywords)
# Resource-ID bypass for profile header elements that sit low
safe_ids = ["following", "follower", "post_count", "button_edit_profile", "button_share_profile"]
res_id = node.get("resource_id", "").lower()
id_bypass = any(k in res_id for k in safe_ids)
threshold = screen_height * NAV_BAR_ZONE
if node.get("y", 0) > threshold and not (is_nav_intent or is_modal_intent or id_bypass):
# Not an AI error—this is the deterministic prep-filter culling the NavBar before VLM logic.
logger.debug(f"🛡️ [Pre-LLM Pruning] Ignored navbar/overlay element (y={node.get('y')}) to prevent VLM hallucination.")
return False
# 4. Reject own profile/story if the intent is not explicitly looking for it.
# Intuitively, "profile picture avatar story ring" means "click a user's story".
# If we are looking for a story/profile, we must NOT click our OWN story.
is_targeting_own_profile = any(k in low_intent for k in ["own profile", "my profile", "own story"])
if not is_targeting_own_profile:
semantic_lower = node.get("semantic_string", "").lower()
if "your story" in semantic_lower:
logger.debug(f"🛡️ [Structural Guard] Rejecting own story overlay for intent '{intent_description}': {node['semantic_string']}")
return False
bot_username = self._get_current_username()
if bot_username and bot_username in semantic_lower:
# Rejecting bot's own username to prevent clicking itself (e.g. "marisaundmarc's story, 0 of 27")
logger.debug(f"🛡️ [Structural Guard] Rejecting own username '{bot_username}' for intent '{intent_description}': {node['semantic_string']}")
return False
# 5. Language-Agnostic Modal/Menu Guard
# Prevent clicks on items inside dialogs, bottom sheets, or context menus
# UNLESS the intent explicitly targets a menu or modal interaction.
# This completely nullifies the risk of accidentally clicking "Favorites" or "Unfollow" from a dropdown.
menu_id_indicators = ["menu_item", "option", "bottom_sheet", "dialog", "action_sheet", "popup"]
is_menu_node = any(m in node.get("resource_id", "").lower() for m in menu_id_indicators)
# If the intent doesn't sound like it wants a menu...
wants_menu_intent = any(k in low_intent for k in ["menu", "option", "more", "dismiss", "cancel", "modal"])
if is_menu_node and not wants_menu_intent:
logger.debug(f"🛡️ [Modal Guard] Rejecting menu/dialog item for non-menu intent '{intent_description}': {node['semantic_string']}")
return False
# 6. Reject nodes with zero area (invisible)
if node.get("area", 0) == 0:
return False
return True
def _is_blacklisted(self, intent: str, semantic_string: str) -> bool:
@@ -214,6 +297,26 @@ class TelepathicEngine:
# App Context Guard
# ──────────────────────────────────────────────
def _get_current_username(self) -> str:
"""Helper to get the current IG username from config, safely."""
username = getattr(self, "_cached_username", None)
if not username:
try:
from GramAddict.core.config import Config
cfg = Config()
raw_u = cfg.args.username if hasattr(cfg, "args") and hasattr(cfg.args, "username") else None
if raw_u and isinstance(raw_u, list) and len(raw_u)>0:
username = str(raw_u[0]).lower()
elif isinstance(raw_u, str):
username = raw_u.lower()
else:
username = ""
except Exception:
username = ""
self._cached_username = username
return username
# ──────────────────────────────────────────────
def _is_instagram_context(self, nodes: list[dict]) -> bool:
"""
Returns True only if the extracted nodes appear to come from the target app.
@@ -249,7 +352,7 @@ class TelepathicEngine:
ZERO AI cost — runs entirely on CPU string ops.
"""
# Extract meaningful keywords from intent (strip common filler words)
filler = {"tap", "the", "button", "tab", "on", "in", "a", "an", "of", "for", "to", "and", "or", "input", "text", "box"}
filler = {"tap", "the", "button", "on", "in", "a", "an", "of", "for", "to", "and", "or", "input", "text", "box"}
intent_words = set(w.lower() for w in re.split(r'\W+', intent_description) if w and w.lower() not in filler and len(w) > 1)
if not intent_words:
@@ -292,8 +395,8 @@ class TelepathicEngine:
# Score = ratio of intent keywords matched
score = hits / len(intent_words)
# Require at least 40% keyword overlap to avoid false positives
if score >= 0.4:
# Require at least 75% keyword overlap to avoid fatal false positives (e.g. 'post username' matching 'Send post')
if score >= 0.75:
scored.append((node, score))
if not scored:
@@ -352,7 +455,11 @@ 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:
screen_height = int(max_y * 1.05)
# 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)
# Pre-filter: Remove structurally implausible nodes and blacklisted mappings
viable_nodes = []
@@ -403,6 +510,27 @@ class TelepathicEngine:
"source": "memory"
}
# ── Stage 1.25: Structural Grid Fast Path ──
# Direct resource-ID + spatial sorting for grid navigation intents.
# This bypasses keyword/vector/VLM entirely — O(n) deterministic.
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"
}
# ── Stage 1.5: Deterministic Keyword Fast Path ──
fast_path_result = self._keyword_match_score(intent_description, viable_nodes)
if fast_path_result:
@@ -529,6 +657,23 @@ class TelepathicEngine:
actual_intent = intent or ctx["intent"]
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:"
has_text = bool(re.search(r'(?<!\w)text:', sem.lower()))
has_desc = bool(re.search(r'(?<!\w)description:', sem.lower()))
is_generic = not has_text and not has_desc
if is_generic:
logger.warning(
f"⚠️ [Anti-Poisoning] Refusing to blacklist generic semantic: '{sem}'. "
f"It would poison all similar nodes. Skipping blacklist for '{actual_intent}'."
)
TelepathicEngine._last_click_context = None
return
# Add to blacklist
if actual_intent not in self._blacklist:
self._blacklist[actual_intent] = []
@@ -545,6 +690,74 @@ class TelepathicEngine:
TelepathicEngine._last_click_context = None
def verify_success(self, intent: str, post_click_xml: str) -> bool:
"""
Hardened verification. Does NOT rely on raw XML changes.
Inspects the post-click XML for semantic markers that prove the intent worked.
"""
ctx = TelepathicEngine._last_click_context
if not ctx:
return True # No context to verify against
# 1. Positional Consistency (Atomic Guard)
# If the screen changed so much that NO elements are near the original click,
# it might be a context-switch (navigation success).
# 2. Intent-Specific Markers
low_intent = intent.lower()
low_xml = post_click_xml.lower()
# 0. Global Hallucination Guards
# If we didn't intend to share/send and the share sheet opened, the click was a hallucination.
if "share" not in low_intent and "send" not in low_intent and "share_sheet" in low_xml:
logger.warning(f"❌ [Semantic Verification] FAILED: Hallucinated click opened the direct share sheet. ({intent})")
return False
if "poll" not in low_intent and "survey" not in low_intent and ("_poll_" in low_xml or "survey_" in low_xml):
logger.warning(f"❌ [Semantic Verification] FAILED: Hallucinated click opened a survey/poll. ({intent})")
return False
# Success markers for common actions
if "like" in low_intent:
# Check for "Liked" or "gefällt mir nicht mehr" in content-desc or text
marker_found = re.search(r"\b(liked|gefällt mir nicht mehr|gefällt mir am)\b", low_xml)
if marker_found:
logger.debug("✅ [Semantic Verification] Success confirmed: 'Liked' state detected.")
return True
else:
logger.warning("❌ [Semantic Verification] FAILED: Post does not report 'Liked' state after click.")
return False
if "follow" in low_intent:
# Check if button changed to "Following" or "Requested"
marker_found = re.search(r"\b(following|requested|folgst du|angefragt)\b", low_xml)
if marker_found:
logger.debug("✅ [Semantic Verification] Success confirmed: 'Following/Requested' state detected.")
return True
else:
logger.warning("❌ [Semantic Verification] FAILED: Profile does not report 'Following' state.")
return False
if any(k in low_intent for k in ["explore grid", "profile grid", "first image"]):
# Clicking a grid item MUST open a post view.
# Posts have feed markers. Reels have clips markers.
feed_markers = [
"row_feed_button_like", "row_feed_button_comment", "row_feed_button_share",
"row_feed_comment_textview_layout", "row_feed_view_group",
"clips_media_component", "row_feed_photo_profile_name", "row_feed_photo_imageview"
]
marker_found = any(m in low_xml for m in feed_markers)
if marker_found:
logger.debug("✅ [Semantic Verification] Success confirmed: Post opened from grid (Feed markers detected).")
return True
else:
logger.warning("❌ [Semantic Verification] FAILED: Grid tap did not open a valid post view.")
return False
# For general navigation, raw XML change is still the fallback
# (covered by the caller in q_nav_graph for now)
return True
# ──────────────────────────────────────────────
# Vision Cortex Fallback (VLM)
# ──────────────────────────────────────────────
@@ -552,15 +765,37 @@ class TelepathicEngine:
def _vision_cortex_fallback(self, intent: str, nodes: list[dict], device, screen_height: int = 2400) -> Optional[dict]:
"""
Uses a Language Model to identify the correct node from parsed screen XML
when embeddings are insufficient. 100% Screenshot-free for maximum speed and zero hallucination.
when embeddings are insufficient. Opt-in Native Vision Processing via Device Screenshots!
Guards are STRUCTURAL (size, position, class) not ID-based.
Learning happens via the confirm/reject feedback loop, not here.
"""
try:
# Limit to 20 nodes for token efficiency
from GramAddict.core.config import Config
args = getattr(Config(), "args", None)
use_vision = getattr(args, "ai_vision_navigation", False) if args else False
images_payload = None
if use_vision and device is not None:
try:
logger.debug("👁️ [Vision Inference] Capturing screen for spatial understanding...")
img_obj = device.screenshot()
if img_obj:
if hasattr(img_obj, "save"):
import io
buf = io.BytesIO()
img_obj.save(buf, format='JPEG')
raw_bytes = buf.getvalue()
else:
raw_bytes = img_obj
b64_str = base64.b64encode(raw_bytes).decode('utf-8')
images_payload = [b64_str]
except Exception as e:
logger.warning(f"⚠️ [Vision Inference] Failed to capture or encode screenshot: {e}")
# Expand context to 150 nodes to harness local 8k-32k models natively
simplified_nodes = []
for i, n in enumerate(nodes[:20]):
for i, n in enumerate(nodes[:150]):
simplified_nodes.append({
"index": i,
"bounds": n["raw_bounds"],
@@ -568,21 +803,39 @@ class TelepathicEngine:
})
# Get model config
from GramAddict.core.config import Config
try:
args = Config().args
except Exception:
args = None
model = getattr(args, "ai_telepathic_model", "google/gemini-3.1-flash-lite-preview") if args else "google/gemini-3.1-flash-lite-preview"
url = getattr(args, "ai_telepathic_url", "https://openrouter.ai/api/v1/chat/completions") if args else "https://openrouter.ai/api/v1/chat/completions"
if device and hasattr(device, 'args') and device.args:
model = getattr(device.args, "ai_telepathic_model", model)
url = getattr(device.args, "ai_telepathic_url", url)
args = getattr(device, "args", None)
model = getattr(args, "ai_telepathic_model", "llama3.2:1b") if args else "llama3.2:1b"
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate") if args else "http://localhost:11434/api/generate"
# --- Model Trust Logging ---
from GramAddict.core.benchmark_guard import BENCHMARKS_FILE
trust_log = f"Using {model}"
try:
if os.path.exists(BENCHMARKS_FILE):
with open(BENCHMARKS_FILE, "r") as f:
bench_data = json.load(f).get("models", {}).get(model, {})
if bench_data:
score = bench_data.get("telepathic_score", 0)
passed = "PASS" if bench_data.get("passed_all", False) else "FAIL"
unsuitable = bench_data.get("is_unsuitable", False)
trust_level = "HIGH" if score >= 80 and not unsuitable else "MEDIUM" if score >= 50 and not unsuitable else "LOW/UNSAFE"
trust_log += f" [Benchmark: {score}/100 | {passed} | Trust: {trust_level}]"
if unsuitable:
logger.error(f"⛔ [Safety Alert] {model} is marked as UNSUITABLE for this task!")
except Exception:
pass
logger.info(f"🧠 [Telepathic] Intent: '{intent}' -> {trust_log}")
# ---------------------------
system_prompt = (
"You identify which UI element to tap based ONLY on a JSON array of parsed Android elements. "
"Each element has an 'index', structural 'bounds', and a 'semantic' description. "
"Output ONLY valid JSON containing the exact `index` to interact with, and a `reason`. "
"You are an Android UI expert. Identify the correct element index to tap based on the provided JSON.\n"
"Rules:\n"
"1. Output ONLY a raw JSON object.\n"
"2. NO markdown formatting, NO triple backticks, NO explanation.\n"
"3. Format: {\"index\": number, \"reason\": \"string\"}\n"
"4. If no element matches, return {\"index\": -1, \"reason\": \"no match\"}\n"
"5. FATAL RULES: NEVER select 'Share', 'Send post', 'Poll', or 'Survey' buttons UNLESS the intent explicitly commands it!\n"
"6. ACCOUNT SAFETY: NEVER select buttons that modify account state (Favorite, Mute, Block, Unfollow, Restrict) unless specifically commanded."
)
user_prompt = (
@@ -592,12 +845,27 @@ class TelepathicEngine:
"- Pick the SMALLEST, most specific button or icon\n"
"- NEVER pick large containers, full-screen views, or recycler views\n"
"- NEVER pick system icons (wifi, battery, status bar, clock)\n"
"- IGNORE BOTTOM NAVIGATION TABS (Home, Search, Reels, Message, Profile) if the intent is to interact with a post or comment.\n"
"- A 'Comment input' is usually an EditText or a region near the bottom but ABOVE the navigation bar.\n"
"- A 'story tray' or 'story ring' is ALWAYS located at the very TOP of the screen (low Y coordinates).\n"
"Return: {\"index\": number, \"reason\": \"...\"}"
)
resp_str = query_telepathic_llm(model, url, system_prompt, user_prompt)
resp_str = query_telepathic_llm(model, url, system_prompt, user_prompt, images_b64=images_payload)
data = json.loads(resp_str)
# ── Robustness: Handle list responses from LLM ──
if isinstance(data, list):
if len(data) > 0 and isinstance(data[0], dict):
data = data[0]
else:
logger.error(f"VLM returned unexpected list format: {data}")
return None
if not isinstance(data, dict):
logger.error(f"VLM returned non-dict response: {type(data)}")
return None
idx = data.get("index")
if idx is not None and 0 <= idx < len(nodes):
match = nodes[idx]
@@ -617,12 +885,14 @@ class TelepathicEngine:
})
return None
# ── Structural Guard 2: Position (status bar) ──
# ── Structural Guard 2: Position (status / nav bar) ──
if match.get("y", 0) < screen_height * STATUS_BAR_ZONE:
logger.error(
f"❌ [Structural Guard] VLM selected element in status bar zone "
f"(y={match.get('y')}): {match['semantic_string']}. REJECTING."
)
logger.error(f"❌ [Structural Guard] VLM selected element in status bar zone: {match['semantic_string']}. REJECTING.")
return None
is_nav_intent = any(k in intent.lower() for k in ["tab", "navigation", "search and explore", "reels", "profile", "home", "message"])
if match.get("y", 0) > 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
# ── Structural Guard 3: Already blacklisted ──

View File

@@ -110,4 +110,4 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses
if failed_scrolls > 3:
return "CONTEXT_LOST"
return "SESSION_OVER"
return "FEED_EXHAUSTED"

View File

@@ -0,0 +1,196 @@
import os
import sys
import json
import time
# Root path alignment
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(ROOT_DIR)
def colored(text, color, attrs=None):
colors = {
"red": "\033[91m", "green": "\033[92m", "yellow": "\033[93m",
"blue": "\033[94m", "magenta": "\033[95m", "cyan": "\033[96m",
"white": "\033[97m"
}
reset = "\033[0m"
bold = "\033[1m" if attrs and "bold" in attrs else ""
return f"{bold}{colors.get(color, '')}{text}{reset}"
from GramAddict.core.config import Config
from GramAddict.core.qdrant_memory import ParasocialCRMDB, CommentMemoryDB
from GramAddict.core.resonance_engine import ResonanceEngine
import xml.etree.ElementTree as ET
class MockArgs:
def __init__(self):
self.ai_vibe = "friendly, authentic, travel, photography"
self.ai_blacklist_topics = "onlyfans, bitcoin, crypto, nsfw, spam, give-away"
self.ai_learn_comments = True
self.ai_condenser_model = "qwen3.5:latest"
self.ai_condenser_url = "http://localhost:11434/api/generate"
self.ai_vision_navigation = False
self.ai_vision_context = False
class MockConfig:
def __init__(self):
self.args = MockArgs()
class AIMemoryDiagnosticRunner:
def __init__(self):
self.configs = MockConfig()
Config().args = self.configs.args
self.crm_db = ParasocialCRMDB()
self.comment_db = CommentMemoryDB()
self.resonance_oracle = ResonanceEngine("benchmark_agent", crm=self.crm_db)
def setup(self):
print(colored("🧹 Initializing benchmark data...", "cyan"))
# We handle unique targets so we don't wipe the DB
pass
def test_rag_comment_extraction(self) -> dict:
"""
Challenge: Pass an XML dump with real comments and toxic OnlyFans/Crypto spam.
Verify that the LLM Condenser drops the junk and stores the good comments.
"""
fixture_path = os.path.join(ROOT_DIR, "tests", "fixtures", "comments_mock.xml")
with open(fixture_path, "r", encoding="utf-8") as f:
xml_data = f.read()
print(colored(f" -> Extracing comments using RAG Condenser ({self.configs.args.ai_condenser_model})...", "yellow"))
start = time.time()
# Intercept the database write to bypass Qdrant indexing limits and solely test RAG filter logic
intercepted_comments = []
def mock_log(self, text: str, vibe: str, author: str = "unknown"):
intercepted_comments.append(text)
try:
from unittest.mock import patch
with patch.object(CommentMemoryDB, 'store_comment', new=mock_log):
# Override the author logic
test_author = f"benchmark_source_{int(time.time())}"
self.resonance_oracle.extract_and_learn_comments(xml_data, self.configs, author=test_author)
time.sleep(1.0)
except Exception as e:
print(f"❌ EXCEPTION: {e}")
return {"passed": False, "reason": str(e)}
try:
learned_texts = [c.lower() for c in intercepted_comments]
dur = time.time() - start
print(colored(f" -> Intercepted: {learned_texts}", "yellow"))
toxic_count = sum(1 for t in learned_texts if "onlyfans" in t or "bitcoin" in t or "dm" in t or "$" in t)
good_count = sum(1 for t in learned_texts if "majestic" in t or "lighting" in t)
if toxic_count > 0:
print(colored(" ❌ [Sub-Test] LLM Condenser hallucinated or failed to block toxic queries (OnlyFans/Crypto).", "red"))
return {"passed": False, "reason": "Toxic comments leaked"}
if good_count == 0:
print(colored(" ❌ [Sub-Test] LLM Condenser stripped everything or crashed. No good comments persisted.", "red"))
return {"passed": False, "reason": "Good comments dropped"}
print(colored(f" ✅ [Sub-Test] RAG Filter passed! 0 toxic comments, {good_count} valid comments mapped. Latency {dur:.2f}s", "green"))
return {"passed": True, "reason": "Toxic filtered, good preserved."}
except Exception as e:
return {"passed": False, "reason": f"DB Error: {e}"}
def test_crm_profile_context(self) -> dict:
"""
Challenge: Parse and persist profile data into the CRM safely.
"""
target = "benchmark_target"
context_string = "234 Posts | 1.2M Followers | 🏔️ Alpine Photographer | Link in bio"
try:
self.crm_db.log_profile_context(target, context_string)
time.sleep(0.5) # indexing buffer
history = self.crm_db.get_conversation_context(target)
if context_string in history or "1.2M Followers" in history:
print(colored(" ✅ [Sub-Test] Profile context cleanly injected into RAG CRM payload.", "green"))
return {"passed": True, "reason": "Context string found."}
else:
return {"passed": False, "reason": "Profile context missing from CRM retrieval."}
except Exception as e:
return {"passed": False, "reason": str(e)}
def test_crm_interaction_evolution(self) -> dict:
"""
Challenge: Push 3 sequential interactions for a user to see if the CRM stage evolves (0 -> 1 -> 2 -> 3).
"""
target = "benchmark_target"
try:
print(" -> Interacting: 'Like'")
self.crm_db.log_interaction(target, "tap_like_button", new_stage=1)
time.sleep(0.1)
print(" -> Interacting: 'Follow'")
self.crm_db.log_interaction(target, "tap_follow_button", new_stage=2)
time.sleep(0.1)
print(" -> Interacting: 'Comment'")
self.crm_db.log_generated_comment(target, "Wow great photo!")
self.crm_db.log_interaction(target, "tap_comment_button", new_stage=3)
time.sleep(0.5)
stage_info = self.crm_db.get_relationship_stage(target)
stage = stage_info.get("stage", 0)
if stage >= 3:
print(colored(f" ✅ [Sub-Test] CRM safely advanced state memory to Stage {stage}.", "green"))
return {"passed": True, "reason": "Evolution logic passed."}
else:
print(colored(f" ❌ [Sub-Test] CRM stalled at Stage {stage}!", "red"))
return {"passed": False, "reason": "Failed to evolve stage"}
except Exception as e:
return {"passed": False, "reason": str(e)}
def execute_all(self):
self.setup()
results = {
"timestamp": time.time(),
"model": self.configs.args.ai_condenser_model,
"scenarios": {}
}
def run_and_log(name, func):
print(colored(f"\n--- SCENARIO: {name} ---", "magenta"))
start_time = time.time()
data = {"passed": False, "reason": "Unknown error", "latency_ms": 0}
try:
res = func()
if isinstance(res, dict): data.update(res)
elif res is True: data["passed"] = True
except Exception as e:
print(colored(f"❌ EXCEPTION: {e}", "red"))
data["reason"] = str(e)
dur = time.time() - start_time
data["latency_ms"] = int(dur * 1000)
results["scenarios"][name] = data
if data["passed"]:
print(colored(f"🏁 {name} completed successfully in {dur:.2f}s", "green"))
else:
print(colored(f"🚨 {name} FAILED! (Elapsed: {dur:.2f}s)", "red", attrs=["bold"]))
print(colored(f" Reason: {data['reason']}", "yellow"))
run_and_log("RAG Comment Blacklist Extraction", self.test_rag_comment_extraction)
run_and_log("CRM Profile Context Injection", self.test_crm_profile_context)
run_and_log("CRM Sequential Evolution", self.test_crm_interaction_evolution)
self.setup() # Teardown
out_path = os.path.join(ROOT_DIR, "benchmarks", "data", "ai_memory_results.json")
with open(out_path, "w") as f:
json.dump(results, f, indent=4)
print(colored(f"\n📄 Saved AI Memory Benchmark results to: {out_path}", "cyan", attrs=["bold"]))
if __name__ == "__main__":
runner = AIMemoryDiagnosticRunner()
runner.execute_all()

View File

@@ -0,0 +1,231 @@
import os
import sys
import time
import logging
import json
from colorama import Fore, Style, init
# Init Colorama for cross-platform color support
init(autoreset=True)
# Ensure root is in path
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT_DIR)
from GramAddict.core.telepathic_engine import TelepathicEngine
from GramAddict.core.qdrant_memory import UIMemoryDB
# Mute noisy loggers
logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
def colored(text, color, attrs=None):
c = getattr(Fore, color.upper(), "")
attr_str = ""
if attrs and "bold" in attrs:
attr_str = Style.BRIGHT
return f"{attr_str}{c}{text}"
class MockArgs:
def __init__(self):
self.ai_telepathic_model = "qwen3.5:latest"
self.ai_telepathic_url = "http://localhost:11434/api/generate"
self.ai_embedding_model = "nomic-embed-text"
self.ai_embedding_url = "http://localhost:11434/api/embeddings"
self.ai_vision_navigation = True
self.ai_vision_context = True
import base64
class MockDevice:
def __init__(self):
self.args = MockArgs()
self.app_id = "com.instagram.android"
def screenshot(self):
# Return a simple 1x1 black pixel PNG to test the True Vision payload mapping
# without crashing on invalid image data.
return base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAAXNSR0IArs4c6QAAAARnQU1BAACxjwv8YQUAAAAJcEhZcwAADsMAAA7DAcdvqGQAAAAXSURBVBhXY3jP4PgfAAWEAziO3O8MAAAAASUVORK5CYII=")
from GramAddict.core.config import Config
Config().args = MockArgs()
class BrainDiagnosticRunner:
"""
Professional diagnostic suite for Live integration testing of the
Singularity LLM Cognitive Stack and Vector DB (Qdrant) persistence.
Tested against heavy real-world XML dumps from Instagram.
"""
def __init__(self):
self.device = MockDevice()
self.engine = TelepathicEngine.get_instance()
self.mem_db = UIMemoryDB()
# Test Namespaces
self.intents = {
"modal": "diagnostics_dismiss_obstacle",
"ad": "diagnostics_find_sponsored",
"hallucination": "diagnostics_tap_like_button",
"unfollow": "diagnostics_tap_following_button"
}
# Load heavy real-world XML files
self.fixtures_dir = os.path.join(ROOT_DIR, "tests", "fixtures")
self.xmls = {
"modal": self._load_fixture("blocked_ui.xml"),
"ad": self._load_fixture("peugeot_ad.xml"),
"hallucination": self._load_fixture("vlm_hallucination.xml"),
"unfollow": self._load_fixture("unfollow_list_dump.xml")
}
def _load_fixture(self, filename) -> str:
path = os.path.join(self.fixtures_dir, filename)
if not os.path.exists(path):
raise FileNotFoundError(f"Fixture {filename} completely missing. Cannot run parcours.")
with open(path, "r", encoding="utf-8") as f:
return f.read()
def setup(self):
print(colored("🧠 STARTING LIVE BRAIN 'PARCOURS' DIAGNOSTICS (Qdrant + Qwen 3.5)", "cyan", attrs=["bold"]))
if not self.mem_db.is_connected:
logger.error("❌ Qdrant is offline! Diagnostics cannot proceed.")
sys.exit(1)
print(colored("🧹 Initializing diagnostic namespace (clearing old cache)...", "yellow"))
for intent in self.intents.values():
pt_id = self.mem_db._deterministic_id(intent)
self.mem_db._delete_point(pt_id)
def teardown(self):
print(colored("🧹 Tearing down diagnostic namespace...", "yellow"))
for intent in self.intents.values():
pt_id = self.mem_db._deterministic_id(intent)
self.mem_db._delete_point(pt_id)
print(colored("✅ Diagnostics Complete.", "green", attrs=["bold"]))
def test_modal_trap(self) -> dict:
"""
Challenge: Find the 'OK' or 'Dismiss' button on a blocking popup.
"""
xml = self.xmls["modal"]
intent = self.intents["modal"]
node = self.engine.find_best_node(xml, intent, min_confidence=0.8, device=self.device)
if not node:
print(colored(" ❌ LLM failed to find the dismiss button entirely.", "red"))
return {"passed": False, "reason": "No node found"}
semantic = str(node.get("semantic", "")).lower()
if "try again later" in semantic or "action block" in semantic:
print(colored(" ❌ LLM selected the title text instead of the dismiss button.", "red"))
return {"passed": False, "reason": "Selected title instead of button"}
if "dismiss" in semantic or "ok" in semantic:
print(colored(f" ✅ VLM correctly reasoned the popup OK/Dismiss button: {semantic}", "green"))
return {"passed": True, "reason": f"Found correct button: {semantic}"}
return {"passed": False, "reason": f"Selected unrelated element: {semantic}"}
def test_ad_deception(self) -> dict:
"""
Challenge: Identify if the post is an ad by finding 'Sponsored' text.
"""
xml = self.xmls["ad"]
intent = self.intents["ad"]
node = self.engine.find_best_node(xml, intent, min_confidence=0.8, device=self.device)
if not node:
print(colored(" ❌ LLM failed to identify the sponsored indicator.", "red"))
return {"passed": False, "reason": "Missed sponsored text"}
semantic = str(node.get("semantic", "")).lower()
if "sponsored" in semantic:
print(colored(" ✅ VLM correctly identified the tiny 'Sponsored' label amidst a huge post.", "green"))
# --- Test Fast Path Recall Sub-Scenario ---
# Save it
self.engine.confirm_click(intent)
self.mem_db.store_memory(intent, xml, node)
import time
time.sleep(0.5)
# Try to grab it again
start = time.time()
recall_node = self.engine.find_best_node(xml, intent, min_confidence=0.8, device=self.device)
dur = time.time() - start
if recall_node and recall_node.get("source") == "memory":
print(colored(f" ✅ [Sub-Test] Instant Qdrant Memory Recall verified! Latency: {dur:.3f}s", "green"))
return {"passed": True, "reason": "Identified sponsored text and verified memory loop."}
else:
print(colored(" ❌ [Sub-Test] Memory recall failed.", "red"))
return {"passed": False, "reason": "Found ad, but memory persistence failed."}
return {"passed": False, "reason": f"Picked wrong node: {semantic}"}
def test_vlm_hallucination(self) -> dict:
"""
Challenge: Find the like heart icon, ignoring caption text that says 'LIKE'.
"""
xml = self.xmls["hallucination"]
intent = self.intents["hallucination"]
node = self.engine.find_best_node(xml, intent, min_confidence=0.8, device=self.device)
if not node:
print(colored(" ❌ LLM failed to find any like button.", "red"))
return {"passed": False, "reason": "No node found"}
semantic = str(node.get("semantic", "")).lower()
is_caption = ("double tap" in semantic or "like" in semantic) and "row feed button" not in semantic
if is_caption:
print(colored(" ❌ LLM fell for the semantic hallucination gap and selected the text caption!", "red"))
return {"passed": False, "reason": "Fell for caption text trap"}
if "row feed button like" in semantic or "heart" in semantic:
print(colored(" ✅ VLM successfully ignored the deceptive caption and found the structural like button.", "green"))
return {"passed": True, "reason": "Ignored text trap, clicked structural button"}
return {"passed": False, "reason": f"Picked unrelated node: {semantic}"}
def execute_all(self):
self.setup()
results = {
"timestamp": time.time(),
"model": self.device.args.ai_telepathic_model,
"scenarios": {}
}
def run_and_log(name, func):
print(colored(f"\n--- SCENARIO: {name} ---", "magenta"))
start_time = time.time()
data = {"passed": False, "reason": "Unknown error", "latency_ms": 0}
try:
res = func()
if isinstance(res, dict): data.update(res)
elif res is True: data["passed"] = True
except Exception as e:
print(colored(f"❌ EXCEPTION: {e}", "red"))
data["reason"] = str(e)
dur = time.time() - start_time
data["latency_ms"] = int(dur * 1000)
results["scenarios"][name] = data
if data["passed"]:
print(colored(f"🏁 {name} completed successfully in {dur:.2f}s", "green"))
else:
print(colored(f"🚨 {name} FAILED! (Elapsed: {dur:.2f}s)", "red", attrs=["bold"]))
run_and_log("The Modal Trap (Blocked UI)", self.test_modal_trap)
run_and_log("The Ad Deception (Sponsored)", self.test_ad_deception)
run_and_log("The VLM Hallucination Gap (Text Trap)", self.test_vlm_hallucination)
self.teardown()
out_path = os.path.join(ROOT_DIR, "benchmarks", "data", "live_learning_results.json")
with open(out_path, "w") as f:
json.dump(results, f, indent=4)
print(colored(f"\n📄 Saved intensive learning results to: {out_path}", "cyan", attrs=["bold"]))
if __name__ == "__main__":
runner = BrainDiagnosticRunner()
runner.execute_all()

View File

@@ -3,6 +3,7 @@ import sys
import json
import time
import argparse
import subprocess
from datetime import datetime
# Add root project path so we can import internal modules safely
@@ -10,8 +11,8 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from GramAddict.core.llm_provider import query_telepathic_llm
BENCHMARKS_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "GramAddict/core/llm_benchmarks.json")
SCENARIOS_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "GramAddict/core/benchmark_scenarios.json")
BENCHMARKS_FILE = os.path.join(os.path.dirname(__file__), "data/llm_benchmarks.json")
SCENARIOS_FILE = os.path.join(os.path.dirname(__file__), "data/benchmark_scenarios.json")
def load_json(path):
if os.path.exists(path):
@@ -35,6 +36,9 @@ def normalize_scores(db):
leader_model = None
for name, data in db["models"].items():
if data.get("is_unsuitable"):
continue
raw = data.get("raw_score", 0)
if raw > max_raw:
max_raw = raw
@@ -57,6 +61,40 @@ def normalize_scores(db):
return db
def get_installed_ollama_models():
"""
Finds truly local Ollama models by parsing 'ollama list'.
Strictly excludes remote/cloud endpoints or embedding-only models.
"""
try:
output = subprocess.check_output(["/usr/local/bin/ollama", "list"]).decode("utf-8")
models = []
for line in output.split("\n")[1:]:
if line.strip():
# Format: NAME, ID, SIZE, MODIFIED
parts = line.split()
if len(parts) >= 3:
name = parts[0]
size = parts[2]
# 1. Skip if size is '-' (remote/cloud model)
if size == "-":
continue
# 2. Skip ':cloud' tagged models explicitly
if ":cloud" in name:
continue
# 3. Filter out purely embedding models
if any(k in name.lower() for k in ["embed", "minilm", "rerank"]):
continue
models.append(name)
return models
except Exception as e:
print(f"⚠️ Could not list Ollama models: {e}")
return []
def benchmark_model(model_name: str, url: str, force: bool = False):
db = load_json(BENCHMARKS_FILE) or {"models": {}}
scenarios_data = load_json(SCENARIOS_FILE)
@@ -66,22 +104,25 @@ def benchmark_model(model_name: str, url: str, force: bool = False):
if not force and model_name in db.get("models", {}):
pct = db["models"][model_name].get("relative_performance_pct", "N/A")
print(f"Typical execution skip for {model_name} (Rel: {pct}%). Use --force.")
return
if not db["models"][model_name].get("is_unsuitable"):
print(f"Typical execution skip for {model_name} (Rel: {pct}%). Use --force.")
return
print(f"🚀 [Competitive Benchmarking] Model: {model_name}")
print(f"\n🚀 [Competitive Benchmarking] Model: {model_name}")
total_raw = 0
total_latency = 0
results_detail = {}
passed_all = True
blank_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
system_prompt = (
"You identify which UI element to tap on an Android screen. "
"You identify which UI element to tap based ONLY on a JSON array of parsed Android elements. "
"Output ONLY valid JSON: {\"index\": number, \"reason\": \"brief reason\"}"
)
for scenario in scenarios_data["scenarios"]:
scenarios = scenarios_data["scenarios"]
for scenario in scenarios:
print(f"--- Running: {scenario['name']} ---")
user_prompt = (
@@ -100,6 +141,7 @@ def benchmark_model(model_name: str, url: str, force: bool = False):
total_latency += latency
except Exception as e:
print(f" ❌ API Request failed for scenario {scenario['id']}: {e}")
passed_all = False
continue
raw_points = 0
@@ -118,36 +160,37 @@ def benchmark_model(model_name: str, url: str, force: bool = False):
raw_points += 60
print(f" ✅ Correct index ({data['index']}).")
else:
passed_all = False
print(f" ❌ Wrong index ({data['index']}). Target was {scenario['target_index']}.")
else:
passed_all = False
print(" ❌ JSON missing fields.")
except Exception:
passed_all = False
print(" ❌ JSON Parsing failed.")
results_detail[scenario["id"]] = raw_points
total_raw += raw_points
print(f"\n📊 Total Raw Score for {model_name}: {total_raw}")
avg_latency = total_latency // len(scenarios) if scenarios else 0
print(f"\n📊 {model_name} Result: {'PASS' if passed_all else 'FAIL'} | Score: {total_raw} | Latency: {avg_latency}ms")
if model_name not in db["models"]:
db["models"][model_name] = {}
db["models"][model_name].update({
"raw_score": total_raw,
"latency_ms": total_latency // len(scenarios_data["scenarios"]),
"telepathic_score": int((total_raw / (len(scenarios) * 100)) * 100) if scenarios else 0,
"latency_ms": avg_latency,
"last_tested": datetime.utcnow().isoformat() + "Z",
"details": results_detail
"details": results_detail,
"passed_all": passed_all,
"is_unsuitable": not passed_all
})
# Recalculate relative scores across all models
db = normalize_scores(db)
save_json(BENCHMARKS_FILE, db)
leader_name = [n for n, d in db["models"].items() if d.get("is_leader")][0]
rel_pct = db["models"][model_name]["relative_performance_pct"]
print(f"🏆 Current Leader: {leader_name}")
print(f"✨ Relative Performance for {model_name}: {rel_pct}%")
if __name__ == "__main__":
from GramAddict.core.config import Config
@@ -157,12 +200,17 @@ if __name__ == "__main__":
parser.add_argument("--model", type=str, help="Explicit model name")
parser.add_argument("--url", type=str, help="Explicit endpoint URL")
parser.add_argument("--force", action="store_true", help="Force re-testing")
parser.add_argument("--all-ollama", action="store_true", help="Automatically find and test all local Ollama models")
args, unknown = parser.parse_known_args()
models_to_test = []
if args.model and args.url:
if args.all_ollama:
ollama_models = get_installed_ollama_models()
for m in ollama_models:
models_to_test.append((m, "http://localhost:11434/api/generate"))
elif args.model and args.url:
models_to_test.append((args.model, args.url))
elif args.config:
configs = Config(first_run=True, config=args.config)
@@ -170,11 +218,11 @@ if __name__ == "__main__":
for attr, pref in [("ai_telepathic_model", "ai_telepathic_url"), ("ai_model", "ai_model_url"), ("ai_condenser_model", "ai_condenser_url")]:
m = getattr(configs.args, attr, None)
u = getattr(configs.args, pref, "https://openrouter.ai/api/v1/chat/completions")
u = getattr(configs.args, pref, "http://localhost:11434/api/generate")
if m:
models_to_test.append((m, u))
else:
print("❌ Syntax: --config test_config.yml or --model x --url y")
print("❌ Syntax: --all-ollama OR --config test_config.yml OR --model x --url y")
sys.exit(1)
for m, u in set(models_to_test):

8
run.py
View File

@@ -1,6 +1,14 @@
import sys
import warnings
import GramAddict
warnings.filterwarnings("ignore", category=UserWarning, module="urllib3")
try:
from urllib3.exceptions import NotOpenSSLWarning
warnings.filterwarnings("ignore", category=NotOpenSSLWarning)
except ImportError:
pass
if __name__ == "__main__":
try:
GramAddict.run()

View File

@@ -0,0 +1,132 @@
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()

18
scratch/verify_silence.py Normal file
View File

@@ -0,0 +1,18 @@
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 ---")

View File

@@ -1,7 +1,7 @@
username:
- marisaundmarc
# - marcmintel
device: 192.168.1.206:33055
device: 192.168.1.206:35111
app-id: com.instagram.android
feed: 5-8
explore: 3-5
@@ -18,27 +18,31 @@ dry-run-comments: true
interact-percentage: 100
follow-percentage: 100
follow-limit: 50
likes-count: 2-3
likes-count: 3-5
likes-percentage: 100
stories-count: 2-3
stories-percentage: 30
carousel-count: 2-3
stories-count: 5-8
stories-percentage: 40
carousel-count: 3-4
carousel-percentage: 70
repost-percentage: 5
# --- Projekt Singularity V8: Ultra-Smarte Config ---
ai-model: qwen3.5:latest # Dein bestes lokales Modell für Kommentare & Vibe
# WICHTIG: Wir nutzen ÜBERALL exakt das gleiche Modell (qwen3.5:latest).
# Dadurch muss Ollama das Modell nicht im VRAM hin- und herswappen, was den Bot extrem schnell macht!
ai-model: qwen3.5:latest # Generative AI (Comments, CRM)
ai-model-url: http://localhost:11434/api/generate
ai-telepathic-model: google/gemini-3.1-flash-lite-preview # Der Navigations-Champion
ai-telepathic-url: https://openrouter.ai/api/v1/chat/completions
ai-telepathic-model: qwen3.5:latest # Der neue lokale Navigations-Champion (Benchmark: 100/100)
ai-telepathic-url: http://localhost:11434/api/generate
ai-fallback-model: qwen3.5:latest # Kein "Halluzinations-Risiko" mehr im Fallback
ai-fallback-model: qwen3.5:latest # Visuelle Notfall-Erkennung
ai-fallback-url: http://localhost:11434/api/generate
ai-condenser-model: llama3.2:1b # Reicht für reine Zusammenfassung (spart VRAM)
ai-condenser-model: qwen3.5:latest # RAG Comment Learning & Spam Filter (100% Pass)
ai-condenser-url: http://localhost:11434/api/generate
# -------------------------------
ai-vision-navigation: true # Sende UI-Screenshots an LLM für visuelles Fallback-Navigieren
ai-vision-context: true # Sende Post/Account-Screenshots an LLM für visuelle DM/Content-Analyse
ai-quality-filter: true
ai-learn-own-profile: true
@@ -48,12 +52,12 @@ ai-learn-only: false
ai-vibe: "friendly, authentic, helpful"
ai-target-audience: "travel, landscape, nature, mountain, photography, adventure, wanderlust, explore"
ai-blacklist-topics: "onlyfans, nsfw, sale, discount, promo, 18+, giveaway"
smart-unfollow: true
smart-unfollow: false
total-comments-limit: 5000
dry-run: false
speed-multiplier: 1.0
watch-photo-time: 1-3
watch-video-time: 3-8
watch-photo-time: 3-6
watch-video-time: 5-12
dont-type: false
skipped-posts-limit: 10
skipped-posts-limit: 15
account-switch-delay: 10-20

View File

@@ -58,7 +58,7 @@ class MockTelepathicEngine:
return {"x": 300, "y": 300, "description": "Grid Image", "score": 1.0}
return None
def _extract_semantic_nodes(self, xml):
def _extract_semantic_nodes(self, xml, intent=None, threshold=0.0):
return [{"x": 10, "y": 10}]
def confirm_click(self, *args, **kwargs):

View File

@@ -38,11 +38,25 @@ def e2e_device_dump_injector():
return _inject_dump
class VirtualClock:
def __init__(self):
self.time = 0.0
self.animation_target_time = 0.0
def sleep(self, seconds):
if hasattr(seconds, '__iter__'):
return # For edge case where something weird is passed
self.time += float(seconds)
clock = VirtualClock()
@pytest.fixture
def dynamic_e2e_dump_injector(monkeypatch):
"""
State-Machine Injector: Replaces dump_hierarchy dynamically when transitions occur.
Validates that the Telepathic Engine's pathfinding truly worked.
It now inherently simulates UI animation delays. If a dump is requested
LESS than 1.5 virtual seconds after a transition, it returns a garbage animating UI.
"""
def _inject(device_mock, state_map, initial_xml):
fix_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -54,22 +68,54 @@ def dynamic_e2e_dump_injector(monkeypatch):
with open(path, "r") as f:
return f.read()
device_mock.deviceV2.dump_hierarchy.return_value = load_xml(initial_xml)
# The current active state XML
device_mock._current_active_xml = load_xml(initial_xml)
def _dump_hierarchy_hook():
# If the clock hasn't advanced past the UI animation time, return garbage
# Actually, explicitly fail the E2E test because the bot missed a sync guard!
if clock.time < clock.animation_target_time:
pytest.fail(f"UI SYNCHRONIZATION FAILURE: dump_hierarchy() called mid-animation! "
f"Virtual Clock is at {clock.time:.1f}s but UI needs until {clock.animation_target_time:.1f}s to settle. "
f"Add a time.sleep() guard before interacting with the UI after a click.", pytrace=False)
return device_mock._current_active_xml
device_mock.deviceV2.dump_hierarchy.side_effect = _dump_hierarchy_hook
from GramAddict.core.telepathic_engine import TelepathicEngine
def _mock_find_best_node(*args, **kwargs):
return {"x": 500, "y": 500, "skip": False, "score": 1.0, "source": "e2e_mock"}
monkeypatch.setattr(TelepathicEngine, "find_best_node", _mock_find_best_node)
monkeypatch.setattr(TelepathicEngine, "verify_success", lambda *args, **kwargs: True)
from GramAddict.core.q_nav_graph import QNavGraph
original_execute = QNavGraph._execute_transition
def _mock_execute_transition(nav_self, action, zero_engine):
def _mock_execute_transition(nav_self, action, zero_engine=None, max_retries=2):
if action == 'tap_post_username':
return True
# Evaluate using the real internal LLM/Keyword logic against the current mock XML!
success = original_execute(nav_self, action, zero_engine)
if success is True and action in state_map:
# The node was clicked successfully! Swap the XML to the target state.
device_mock.deviceV2.dump_hierarchy.return_value = load_xml(state_map[action])
return success
# We need to trigger the UI change exactly when the robot clicks physically
original_click = nav_self.device.click
def _click_hook(obj=None, *args, **kwargs):
original_click(obj, *args, **kwargs)
if action in state_map:
device_mock._current_active_xml = load_xml(state_map[action])
clock.animation_target_time = clock.time + 1.5
nav_self.device.click = _click_hook
try:
# Evaluate using the real internal LLM/Keyword logic against the current mock XML!
# Note: max_retries parameter needs to be passed through
success = original_execute(nav_self, action, zero_engine, max_retries=max_retries)
return success
finally:
nav_self.device.click = original_click
monkeypatch.setattr(QNavGraph, "_execute_transition", _mock_execute_transition)
return _inject
@@ -77,12 +123,34 @@ def dynamic_e2e_dump_injector(monkeypatch):
@pytest.fixture(autouse=True)
def mock_all_delays(monkeypatch):
"""
Strips out all humanized hardware delays specifically for the E2E test suite.
Ensures loops evaluate instantly using the injected dumps.
Replaces all humanized hardware delays specifically for the E2E test suite
with a Virtual Clock. Ensures loops evaluate instantly but preserves chronological
dependency for our Animation Simulator.
"""
monkeypatch.setattr(time, "sleep", lambda x: None)
monkeypatch.setattr(utils, "random_sleep", lambda *args, **kwargs: None)
monkeypatch.setattr(utils, "sleep", lambda x: None)
global clock
clock.time = 0.0 # reset for test
clock.animation_target_time = 0.0
def simulate_sleep(seconds):
clock.sleep(seconds)
money_sleep = lambda x: simulate_sleep(x)
random_sleep = lambda *args, **kwargs: simulate_sleep(1.0) # Assume 1.0 minimum for randoms
monkeypatch.setattr(time, "sleep", money_sleep)
monkeypatch.setattr(utils, "random_sleep", random_sleep)
monkeypatch.setattr(utils, "sleep", money_sleep)
# Needs to capture specific module sleeps depending on how they imported it
try:
from GramAddict.core import bot_flow
monkeypatch.setattr(bot_flow, "sleep", money_sleep)
monkeypatch.setattr(bot_flow.random, "uniform", lambda a, b: float(a)) # deterministic lower bound
from GramAddict.core import q_nav_graph
monkeypatch.setattr(q_nav_graph.random, "uniform", lambda a, b: float(a))
except Exception:
pass
# Standardize DarwinEngine across tests to prevent mockup math errors on session end
try:

View File

@@ -0,0 +1,26 @@
import pytest
import time
from unittest.mock import MagicMock
from GramAddict.core.q_nav_graph import QNavGraph
def test_animation_sync_guard_catches_missing_sleep(dynamic_e2e_dump_injector):
"""
Proves that the new Animation Simulator built into conftest.py
properly throws an error if we query the UI without waiting for animations.
"""
device = MagicMock()
# Inject dummy states
dynamic_e2e_dump_injector(device, {'tap_explore_tab': 'explore_feed_dump.xml'}, "home_feed_with_ad.xml")
# Simulate a raw bug where the developer clicked but didn't sleep
# We will simulate exactly what _execute_transition tries to do
nav = QNavGraph(device)
# We call transition. QNavGraph internally clicks and sleeps for 1.2s minimum.
# Our Animation target is 1.5s, so the dump inside _execute_transition will hit the fail guard!
from _pytest.outcomes import Failed
with pytest.raises(Failed) as exc_info:
nav._execute_transition("tap_explore_tab")
assert "UI SYNCHRONIZATION FAILURE" in str(exc_info.value), "The simulator failed to catch the missing sleep guard!"

View File

@@ -0,0 +1,49 @@
import pytest
from unittest.mock import MagicMock, patch
from GramAddict.core.q_nav_graph import QNavGraph
@pytest.fixture
def mock_device():
device = MagicMock()
device.app_id = "com.instagram.android"
device.deviceV2 = MagicMock()
# Mock current app to be Instagram
device._get_current_app.return_value = "com.instagram.android"
return device
def test_recovery_from_dm_view(mock_device):
"""
Test Case: Bot starts in a DM thread (UNKNOWN state).
It wants to go to ReelsFeed.
Global nav bar is missing in DMs, so first 'tap_reels_tab' will fail.
Bot should then press 'back' and try again.
"""
nav = QNavGraph(mock_device)
nav.current_state = "UNKNOWN"
# Sequence of dumps (exactly 1 per failed attempt, 2 per successful attempt):
# 1. Attempt 1 (DM): _execute_transition calls dump(1) -> find_best_node returns None -> Returns False
# 2. QNavGraph calls press("back")
# 3. Attempt 2 (Home): _execute_transition calls dump(2) -> find_best_node returns Node
# 4. _execute_transition calls dump(3) -> post-click != pre-click -> Returns True
mock_device.deviceV2.dump_hierarchy.side_effect = ["<DM />", "<Home />", "<ReelsFeed />"]
zero_engine = MagicMock()
with patch('GramAddict.core.telepathic_engine.TelepathicEngine.get_instance') as mock_get:
mock_engine = MagicMock()
mock_get.return_value = mock_engine
# 1st call: DM (nothing found)
# 2nd call: Home (reels tab found)
mock_engine.find_best_node.side_effect = [None, {"x": 50, "y": 50, "score": 0.95, "source": "keyword"}]
success = nav.navigate_to("ReelsFeed", zero_engine)
# Verify
assert success is True
assert nav.current_state == "ReelsFeed"
# Verify recovery was triggered
mock_device.deviceV2.press.assert_called_with("back")
assert mock_device.deviceV2.dump_hierarchy.call_count == 3

View File

@@ -41,6 +41,8 @@ class TestNodeExtraction:
"""
engine = TelepathicEngine()
xml = load_fixture("home_feed_with_ad.xml")
# Test raw extraction (backward compatibility)
nodes = engine._extract_semantic_nodes(xml)
like_nodes = [n for n in nodes if "row feed button like" in n["semantic_string"]]

View File

@@ -0,0 +1,25 @@
import unittest
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestAPIMismatch(unittest.TestCase):
def test_repro_extract_semantic_nodes_type_error(self):
"""
VERIFICATION TEST: Verifies that _extract_semantic_nodes now accepts
extra arguments (threshold), fixing the regression.
"""
engine = TelepathicEngine.get_instance()
xml = "<hierarchy><node resource-id='test' class='android.widget.Button' clickable='true' bounds='[0,0][10,10]' /></hierarchy>"
# This SHOULD now pass
try:
nodes = engine._extract_semantic_nodes(xml, "find buttons", threshold=0.1)
print("\n[V] VERIFICATION SUCCESSFUL: _extract_semantic_nodes accepted extra arguments.")
success = True
except TypeError as e:
print(f"\n[!] BUG STILL PRESENT: Caught TypeError: {e}")
success = False
self.assertTrue(success, "Should NOT have failed with TypeError")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,33 @@
import unittest
import os
import json
from unittest.mock import MagicMock, patch
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestCommentHallucination(unittest.TestCase):
def setUp(self):
self.engine = TelepathicEngine()
self.fixture_path = "/Volumes/Alpha SSD/Coding/bot/debug/xml_dumps/manual_interrupt__2026-04-16_19-19-38.xml"
with open(self.fixture_path, 'r') as f:
self.xml = f.read()
def test_repro_vlm_tab_hallucination(self):
"""
Verify that a navigation tab is REJECTED as a 'Comment input field'
due to structural guards.
"""
# Mock LLM response (picking index 113 which is the DM tab)
mock_llm_json = json.dumps({"index": 0, "reason": "It says Message"})
with patch('GramAddict.core.telepathic_engine.query_telepathic_llm', return_value=mock_llm_json):
# Inspect what find_best_node considers 'viable'
# (We cannot easily intercept internal local variables, so lets just run and see failures)
node = self.engine.find_best_node(self.xml, "Comment input text box editfield", device=MagicMock())
# ASSERTION: The node should be None because the structural guard REJECTED the VLM result
self.assertIsNone(node, f"Found {node}! Structural guard should have rejected the DM tab in Nav Bar zone.")
print("\n[V] VERIFICATION SUCCESSFUL: Structural Guard successfully rejected DM tab.")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,47 @@
import os
import sys
import unittest
import json
from unittest.mock import MagicMock, patch
# Add project root to path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
from GramAddict.core.compiler_engine import VLMCompilerEngine
class TestReproCompilerCrash(unittest.TestCase):
def setUp(self):
self.device = MagicMock()
self.compiler = VLMCompilerEngine(self.device)
self.xml = "<hierarchy><node index='0' resource-id='test_id' /></hierarchy>"
@patch('GramAddict.core.llm_provider.query_telepathic_llm')
def test_list_response_crash(self, mock_query):
"""
Verify that the compiler does NOT crash when the LLM returns a list.
It should handle it or return None gracefully.
"""
# Scenario: LLM returns a list of dictionaries (common with some models)
mock_query.return_value = json.dumps([{"rule_type": "regex", "target_attribute": "resource-id", "pattern": "test.*", "confidence": 0.9}])
try:
result = self.compiler.generate_heuristic("test intent", self.xml)
# If the current code handles lists correctly, this should pass.
# But the user reported a crash.
print(f"Result for list of dicts: {result}")
except Exception as e:
self.fail(f"Compiler crashed with list of dicts: {e}")
# Scenario: LLM returns a raw list (not of dicts)
mock_query.return_value = json.dumps(["pattern", "test.*"])
try:
result = self.compiler.generate_heuristic("test intent", self.xml)
self.assertIsNone(result, "Should return None for invalid list format")
except Exception as e:
# THIS is what the user reported: "'list' object has no attribute 'get'"
# which happens if it tries to call .get() on the list ["pattern", "test.*"]
self.fail(f"Compiler crashed with raw list: {e}")
if __name__ == '__main__':
unittest.main()

View File

@@ -0,0 +1,26 @@
import unittest
from unittest.mock import MagicMock, patch
class TestContextTruthiness(unittest.TestCase):
def test_repro_context_truthiness_bug(self):
"""
Verifies that 'CONTEXT_LOST' string is NOT treated as True when using 'is True' check.
"""
# Mock nav_graph
nav_graph = MagicMock()
nav_graph._execute_transition.return_value = "CONTEXT_LOST"
# Simulate the logic in bot_flow.py (FIXED)
success = nav_graph._execute_transition("tap_comment_button", MagicMock())
# This is the FIXED logic
if success is True:
is_buggy = True
else:
is_buggy = False
self.assertFalse(is_buggy, "Should NOT be buggy: 'CONTEXT_LOST' is not 'True'")
print("\n[V] VERIFICATION SUCCESSFUL: 'CONTEXT_LOST' string rejected by 'is True' check.")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,72 @@
import unittest
import os
from unittest.mock import MagicMock, patch
from GramAddict.core.q_nav_graph import QNavGraph
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestFalseLearning(unittest.TestCase):
def setUp(self):
# Ensure we start with clean caches for tests
if os.path.exists("telepathic_memory.json"):
os.remove("telepathic_memory.json")
if os.path.exists("telepathic_blacklist.json"):
os.remove("telepathic_blacklist.json")
self.device = MagicMock()
self.device.app_id = "com.instagram.android"
self.device._get_current_app.return_value = "com.instagram.android"
# Load Reels dump
with open("tests/fixtures/reels_feed_dump.xml", "r") as f:
self.reels_xml = f.read()
def test_repro_accidental_learning_on_mismatch(self):
"""
REPRO TEST: Verifies that QNavGraph/TelepathicEngine incorrectly learns
a mapping if a tap on the WRONG element changes the screen.
"""
nav = QNavGraph(self.device)
engine = TelepathicEngine.get_instance()
# 1. Setup: The bot wants to 'tap_like_button'
# But we mock the engine to mistakenly return the 'Reels Tab' icon instead
# Reels Tab icon bounds in fixture: [292,2266][355,2329]
fake_node = {
"x": 323, "y": 2297,
"score": 0.85,
"semantic": "id context: 'tab icon'",
"source": "agentic_fallback"
}
# Define a side effect that simulates find_best_node's internal tracking
def mock_find_best_node(xml, intent, **kwargs):
TelepathicEngine._last_click_context = {
"intent": intent,
"semantic_string": fake_node["semantic"],
"x": fake_node["x"],
"y": fake_node["y"],
"timestamp": 12345
}
return fake_node
with patch.object(TelepathicEngine, "find_best_node", side_effect=mock_find_best_node):
# Simulate a UI change happening after the tap (e.g. some animation or tab switch)
# We mock dump_hierarchy to return something DIFFERENT after the click
self.device.deviceV2.dump_hierarchy.side_effect = [
self.reels_xml, # Pre-click
"<root>UI CHANGED</root>" # Post-click
]
# Execute transition
success = nav._execute_transition("tap_like_button", MagicMock())
self.assertFalse(success, "Transition should be REJECTED because semantic verification failed")
# 2. Assert: The bot should NOT have learned the wrong mapping
memory = engine._load_json("telepathic_memory.json")
self.assertNotIn("tap like button", memory, "Should NOT have learned 'tap like button' because fix is working")
print("\n[!] VERIFICATION SUCCESSFUL: Hardened bot rejected wrong mapping for 'tap like button'")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,68 @@
import unittest
import os
from unittest.mock import MagicMock, patch
from GramAddict.core.q_nav_graph import QNavGraph
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestGridHallucination(unittest.TestCase):
def setUp(self):
if os.path.exists("telepathic_memory.json"):
os.remove("telepathic_memory.json")
if os.path.exists("telepathic_blacklist.json"):
os.remove("telepathic_blacklist.json")
self.device = MagicMock()
self.device.app_id = "com.instagram.android"
self.device._get_current_app.return_value = "com.instagram.android"
def test_repro_grid_tap_hallucination(self):
"""
REPRO TEST: Verifies that a generic fallback 'True' in verify_success
causes false learning if the UI changed but we didn't actually open a post.
"""
nav = QNavGraph(self.device)
engine = TelepathicEngine.get_instance()
# VLM picked node 8 which was an 'image button'
fake_node = {
"x": 100, "y": 100,
"score": 0.85,
"semantic": "id context: 'image button'",
"source": "agentic_fallback"
}
def mock_find_best_node(xml, intent, **kwargs):
TelepathicEngine._last_click_context = {
"intent": intent,
"semantic_string": fake_node["semantic"],
"x": fake_node["x"],
"y": fake_node["y"],
"timestamp": 12345
}
return fake_node
with patch.object(TelepathicEngine, "find_best_node", side_effect=mock_find_best_node):
# Pre-click XML (Explore Grid)
self.device.deviceV2.dump_hierarchy.side_effect = [
"<root><node content-desc='Explore' /></root>",
# Post click XML changes (maybe a modal opens), but NO FEED MARKERS
"<root><node content-desc='Something else' /></root>"
]
# Execute transition for explore grid item
# The bug was that verify_success returns True by default.
# If UI changed, it confirms the bad click!
success = nav._execute_transition("tap_explore_grid_item", MagicMock())
# This SHOULD be False if the bot correctly realizes no post was opened.
# If it's True, the test detects the bug.
if success:
print("\n[!] BUG REPRODUCED: Bot learned 'image button' as explore grid item even though no post was opened.")
is_buggy = True
else:
print("\n[V] VERIFICATION SUCCESSFUL: Bot rejected 'image button' because no post was opened.")
is_buggy = False
self.assertFalse(is_buggy, "Should NOT learn mapping if opening post failed.")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,34 @@
import unittest
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestPositionRejection(unittest.TestCase):
def test_repro_following_button_rejection_fix(self):
"""
VERIFICATION TEST: Verifies that a node at y=2182 (on screen height 2424)
is NOT rejected anymore by the refined structural guard.
"""
engine = TelepathicEngine.get_instance()
# This was the problematic node from the logs
node = {
"semantic_string": "description: '2.270following', id context: 'profile header following stacked familiar'",
"x": 800,
"y": 2182,
"resource_id": "com.instagram.android:id/profile_header_following_stacked_familiar",
"area": 5000 # Normal button size
}
# Test 1: Intent is 'tap following list' (Should pass due to keyword and threshold)
passed_keyword = engine._structural_sanity_check(node, "tap following list", screen_height=2424)
print(f"\n[DEBUG] Intent: 'tap following list', Passed: {passed_keyword}")
# Test 2: Intent is something else, but it's a 'safe' ID (Following)
passed_id = engine._structural_sanity_check(node, "some other intent", screen_height=2424)
print(f"\n[DEBUG] Intent: 'some other intent', Passed: {passed_id}")
self.assertTrue(passed_keyword, "Following button should be allowed for following intent")
self.assertTrue(passed_id, "Following button should be allowed due to safe ID bypass")
print("\n[V] VERIFICATION SUCCESSFUL: Position Rejection Fix confirmed.")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,50 @@
import os
import sys
import unittest
from unittest.mock import MagicMock
# Add project root to path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
from GramAddict.core.telepathic_engine import TelepathicEngine
class TestReproReelsTabHallucination(unittest.TestCase):
def setUp(self):
self.engine = TelepathicEngine()
# Path to home feed fixture
self.fixture_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../fixtures/home_feed_with_ad.xml'))
with open(self.fixture_path, 'r', encoding='utf-8') as f:
self.xml_content = f.read()
def test_reels_tab_selection(self):
"""
Verify that the engine selects the actual Reels tab (clips_tab)
and NOT the "Add to story" badge (reel_empty_badge).
"""
intent = "tap reels tab"
# We need to simulate the environment where this fails.
# Currently, 'tab' is in the filler list, so "tap reels tab" -> ["reels"]
# "Add to story" (id: reel_empty_badge) matches "reels" (via alias "reel").
# "Reels" (id: clips_tab) matches "reels" (via content-desc or rid).
result = self.engine.find_best_node(self.xml_content, intent)
self.assertIsNotNone(result, "Should have found a node")
# In the fixture:
# Clips tab is at [216,2235][432,2361] -> center is (324, 2298)
# Add to story? Wait, let's find it in the XML.
# Actually, let's search for "reel" in the XML to see candidates.
print(f"Target selected: {result.get('semantic')} at ({result.get('x')}, {result.get('y')})")
# The Reels tab (clips_tab) has y > 2200.
# The "Add to story" badge is usually at the top.
# If it selects something at the top, it's a hallucination.
self.assertGreater(result['y'], 2000, "Should select a tab at the bottom, not an element at the top")
self.assertIn("clips tab", result['semantic'].lower(), "Should select the clips_tab")
if __name__ == '__main__':
unittest.main()

View File

@@ -0,0 +1,32 @@
import pytest
from xml.etree import ElementTree as ET
from GramAddict.core.bot_flow import _detect_ad_structural
def generate_xml_with_node(res_id, text="", desc=""):
return f'''<?xml version='1.0' encoding='UTF-8' standalone='yes'?>
<hierarchy>
<node class="android.widget.FrameLayout">
<node resource-id="{res_id}" text="{text}" content-desc="{desc}" />
</node>
</hierarchy>'''
def test_detects_real_ad_label():
xml = generate_xml_with_node("com.instagram.android:id/secondary_label", text="Sponsored", desc="Sponsored")
assert _detect_ad_structural(xml) is True
xml = generate_xml_with_node("com.instagram.android:id/secondary_label", text="gesponsert", desc="gesponsert")
assert _detect_ad_structural(xml) is True
def test_ignores_false_positive_short_text():
# If a location or normal text happens to be short, e.g., "ad" for an audio name like "Adele"
# But wait, exact match of "Ad" is usually an Ad.
pass
def test_ignores_non_ad_reels_cta():
# Normal reels can have a CTA for "Use template" or "Use Audio".
# If they use clips_browser_cta, it might be a false positive.
xml = generate_xml_with_node("com.instagram.android:id/clips_browser_cta", text="Use template", desc="Use template")
# If _detect_ad_structural says True, it's a FALSE POSITIVE because it's just a template CTA!
# A real ad CTA usually says "Install Now", "Learn More", etc.
# Currently _detect_ad_structural returns True for ALL clips_browser_cta.
assert _detect_ad_structural(xml) is False, "False positive: 'Use template' is not a sponsored ad"

View File

@@ -0,0 +1,47 @@
import pytest
from unittest.mock import patch, MagicMock
from GramAddict.core.q_nav_graph import QNavGraph
def test_autonomous_retry_on_ambiguity_failure():
"""
Verifies that _execute_transition now uses an internal retry loop.
If the first attempt fails semantic verification (Ambiguity Guard),
it should press BACK, blacklist the node, and retry automatically.
"""
mock_device = MagicMock()
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)
]
mock_engine = MagicMock()
# Mock find_best_node to return Node A then Node B
mock_engine.find_best_node.side_effect = [
{"x": 10, "y": 10, "semantic_string": "Wrong Menu", "source": "vlm"},
{"x": 20, "y": 20, "semantic_string": "Correct Grid", "source": "vlm"}
]
# Mock verify_success to fail first time, succeed second time
mock_engine.verify_success.side_effect = [False, True]
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)
# The transition should ultimately succeed because attempt 2 passes
assert result is True, "Autonomous retry loop failed to return True."
# Verify that the engine blacklisted the first attempt
mock_engine.reject_click.assert_called_once()
# Verify that the engine confirmed the second attempt
mock_engine.confirm_click.assert_called_once()
# Verify BACK was pressed exactly once to clear the wrong menu
mock_device.deviceV2.press.assert_called_once_with("back")
# Verify two clicks were made
assert mock_device.click.call_count == 2

View File

@@ -0,0 +1,35 @@
import pytest
from unittest.mock import MagicMock, patch
from GramAddict.core.dopamine_engine import DopamineEngine
import GramAddict.core.bot_flow as bot_flow
def test_feed_switch_resets_boredom():
"""
Test driven development to prove that changing the feed does not immediately
terminate the session due to a stale boredom value.
This replicates the exact crash reported where empty Inbox resulted in immediate Session Exit.
"""
# Initialize DopamineEngine and simulate the state right before a feed switch
dopamine = DopamineEngine()
# Simulate a full inbox clear causing maximum boredom
dopamine.boredom = 100.0
# Assert that ordinarily, the session WOULD be over
assert dopamine.is_app_session_over() is True
# SIMULATE bot_flow.py logic that occurs during BOREDOM_CHANGE_FEED
result = "BOREDOM_CHANGE_FEED"
assert result == "BOREDOM_CHANGE_FEED"
# Apply the fix from bot_flow.py lines 210-215
dopamine.boredom = max(0.0, dopamine.boredom * 0.2)
# Assert that the session is NO LONGER over, and the bot can continue to the new feed
assert dopamine.boredom == 20.0
assert dopamine.is_app_session_over() is False
def test_session_limit_terminates_session():
dopamine = DopamineEngine()
dopamine.session_limit_seconds = 0 # force time limit
assert dopamine.is_app_session_over() is True

View File

@@ -0,0 +1,222 @@
"""
TDD Test Suite: Explore Grid Navigation Hardening
==================================================
Reproduces the exact production failure from 2026-04-16 22:59 where the bot:
1. Blacklisted ALL image_buttons because of generic semantic strings
2. Could not match "first image in explore grid" via the keyword fast-path
3. VLM picked row 3 instead of row 1 because the prompt lacks spatial ranking
These tests MUST fail before the fix and pass after.
"""
import pytest
import re
from GramAddict.core.telepathic_engine import TelepathicEngine
# ── Realistic node fixtures extracted from live XML dump 2026-04-16_22-59-53 ──
def make_node(x, y, bounds, semantic, res_id="com.instagram.android:id/dummy", text="", desc=""):
m = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds)
l, t, r, b = map(int, m.groups()) if m else (0, 0, 0, 0)
return {
"x": x, "y": y,
"width": r - l, "height": b - t, "area": (r - l) * (b - t),
"raw_bounds": bounds,
"semantic_string": semantic,
"resource_id": res_id,
"class_name": "android.widget.FrameLayout",
"selected": False,
"original_attribs": {"text": text, "desc": desc}
}
EXPLORE_GRID_NODES = [
# Row 1, Col 1 — this is what "first image" should match
make_node(178, 559, "[0,321][356,797]",
"description: '4 photos by Patsy Weingart at row 1, column 1', id context: 'grid card layout container'",
res_id="com.instagram.android:id/grid_card_layout_container",
desc="4 photos by Patsy Weingart at row 1, column 1"),
# Row 1, Col 1 — child image_button (same area, no semantic info)
make_node(178, 558, "[0,321][356,796]",
"id context: 'image button'",
res_id="com.instagram.android:id/image_button"),
# Row 1, Col 2
make_node(540, 559, "[362,321][718,797]",
"description: 'Photo by Barbara at Row 1, Column 2', id context: 'grid card layout container'",
res_id="com.instagram.android:id/grid_card_layout_container",
desc="Photo by Barbara at Row 1, Column 2"),
make_node(540, 558, "[362,321][718,796]",
"id context: 'image button'",
res_id="com.instagram.android:id/image_button"),
# Row 2, Col 2
make_node(540, 1041, "[362,803][718,1279]",
"description: 'Photo by Garima Bhaskar at Row 2, Column 2', id context: 'grid card layout container'",
res_id="com.instagram.android:id/grid_card_layout_container",
desc="Photo by Garima Bhaskar at Row 2, Column 2"),
make_node(540, 1040, "[362,803][718,1278]",
"id context: 'image button'",
res_id="com.instagram.android:id/image_button"),
# Row 3, Col 1 — this is what the VLM wrongly picked
make_node(178, 1523, "[0,1285][356,1761]",
"description: '4 photos by Soul Of Nature Photography at row 3, column 1', id context: 'grid card layout container'",
res_id="com.instagram.android:id/grid_card_layout_container",
desc="4 photos by Soul Of Nature Photography at row 3, column 1"),
make_node(178, 1522, "[0,1285][356,1760]",
"id context: 'image button'",
res_id="com.instagram.android:id/image_button"),
# Search bar
make_node(487, 219, "[32,173][943,265]",
"text: 'Search', id context: 'action bar search edit text'",
res_id="com.instagram.android:id/action_bar_search_edit_text",
text="Search"),
]
class TestBlacklistPoisoning:
"""
Bug: Generic semantic strings like 'id context: image button' get blacklisted,
which kills ALL grid items because they share the same semantic string.
"""
def test_generic_semantic_should_not_be_blacklistable(self):
"""
A semantic string consisting ONLY of a generic id context (no text, no
description) is too ambiguous to blacklist. The engine must refuse to
blacklist it because it would poison all similar nodes.
"""
engine = TelepathicEngine()
# Clear any persisted state to test pure logic
engine._blacklist = {}
# Simulate the flow: the engine "clicked" an image_button and it failed
TelepathicEngine._last_click_context = {
"intent": "first image in explore grid",
"semantic_string": "id context: 'image button'",
"x": 178, "y": 558,
"timestamp": 1000
}
engine.reject_click("first image in explore grid")
# The blacklist should NOT contain this generic entry
blacklisted = engine._blacklist.get("first image in explore grid", [])
assert "id context: 'image button'" not in blacklisted, (
"CRITICAL: Generic semantic 'id context: image button' was blacklisted! "
"This poisons ALL image_buttons in ALL grids."
)
class TestExploreGridFastPath:
"""
Bug: There was no fast-path to match 'first image in explore grid' to a
grid_card_layout_container node. Now the Grid Fast-Path (Stage 1.25) handles
this deterministically via resource-ID + spatial sorting.
"""
def test_grid_fastpath_matches_container(self):
"""
The Grid Fast-Path must match 'first image in explore grid' to
a grid_card_layout_container node without calling VLM/embeddings.
"""
engine = TelepathicEngine()
# Build a minimal XML that the engine can parse — but we test the fast-path
# directly by calling find_best_node with mocked extraction
from unittest.mock import patch
with patch.object(engine, '_extract_semantic_nodes', return_value=EXPLORE_GRID_NODES):
with patch.object(engine, '_is_instagram_context', return_value=True):
result = engine.find_best_node("<fake>", "first image in explore grid", min_confidence=0.82, device=None)
assert result is not None, (
"Grid Fast-Path returned None for 'first image in explore grid'. "
"This forces every explore grid tap to use the expensive VLM fallback."
)
assert "image button" in result.get("semantic", "").lower(), (
f"Grid Fast-Path selected wrong node type: {result.get('semantic')}"
)
def test_grid_fastpath_prefers_topmost_row(self):
"""
When multiple grid items match, the Grid Fast-Path must prefer the
topmost one (smallest Y = row 1) since the intent says 'first'.
"""
engine = TelepathicEngine()
from unittest.mock import patch
with patch.object(engine, '_extract_semantic_nodes', return_value=EXPLORE_GRID_NODES):
with patch.object(engine, '_is_instagram_context', return_value=True):
result = engine.find_best_node("<fake>", "first image in explore grid", min_confidence=0.82, device=None)
if result is not None:
# Row 1 items have y ≈ 559, Row 3 items have y ≈ 1523
assert result["y"] < 800, (
f"Grid Fast-Path selected a grid item at y={result['y']} (row 3+) "
f"instead of row 1 (y≈559). The intent says 'first image'!"
)
class TestVerifySuccessExploreGrid:
"""
Bug: verify_success for explore grid tap checks for feed markers, but
the post_load_timeout dump proved the bot was STILL on the explore grid.
The verification correctly returned False, but the response was to blacklist
the grid_card_layout_container — which is the WRONG reaction. The tap
just didn't register; it doesn't mean the mapping is wrong.
"""
def test_verify_success_returns_false_when_still_on_grid(self):
"""
If we tapped a grid item but the screen still shows the explore grid
(no feed markers), verify_success must return False.
"""
engine = TelepathicEngine()
# Simulate that we just clicked a grid item
TelepathicEngine._last_click_context = {
"intent": "first image in explore grid",
"semantic_string": "description: '4 photos by Patsy Weingart at row 1, column 1'",
"x": 178, "y": 559,
"timestamp": 1000
}
# Post-click XML still shows the explore grid (no feed markers)
still_on_grid_xml = """
<node class="android.widget.FrameLayout">
<node resource-id="com.instagram.android:id/explore_action_bar" />
<node resource-id="com.instagram.android:id/grid_card_layout_container"
content-desc="4 photos by Patsy Weingart at row 1, column 1" />
<node resource-id="com.instagram.android:id/image_button" />
</node>
"""
result = engine.verify_success("first image in explore grid", still_on_grid_xml)
assert result is False, "verify_success should fail when still on explore grid"
def test_verify_success_returns_true_when_post_opened(self):
"""
If the grid tap succeeded and we're now viewing a post with feed markers,
verify_success must return True.
"""
engine = TelepathicEngine()
TelepathicEngine._last_click_context = {
"intent": "first image in explore grid",
"semantic_string": "description: '4 photos by Patsy Weingart at row 1, column 1'",
"x": 178, "y": 559,
"timestamp": 1000
}
# Post-click XML shows a feed post (has feed markers)
post_view_xml = """
<node class="android.widget.FrameLayout">
<node resource-id="com.instagram.android:id/row_feed_photo_imageview" />
<node resource-id="com.instagram.android:id/row_feed_button_like" />
<node resource-id="com.instagram.android:id/row_feed_button_comment" />
<node resource-id="com.instagram.android:id/row_feed_button_share" />
</node>
"""
result = engine.verify_success("first image in explore grid", post_view_xml)
assert result is True, "verify_success should pass when post view is visible"

View File

@@ -0,0 +1,59 @@
"""
TDD Test: Feed Loop Continuation After Stories
===============================================
Reproduces the exact production failure from 2026-04-16 23:12 where the bot
watched 3-5 stories (23 seconds), and then declared the entire session over
instead of continuing to the next feed (HomeFeed, ExploreFeed, ReelsFeed).
The root cause: _run_zero_latency_stories_loop returns "SESSION_OVER" when
stories are exhausted, and the main loop interprets this as "end the entire
bot session" via `else: break`.
"""
import pytest
class TestFeedLoopContinuation:
"""
Tests that completing a sub-feed (Stories, DMs, Search) does NOT terminate
the entire session. The bot must move to the next feed.
"""
def test_stories_complete_returns_feed_exhausted(self):
"""
When stories are watched to the limit, the loop MUST return
'FEED_EXHAUSTED' (not 'SESSION_OVER'). The main loop must then
switch to another feed, not end the session.
"""
# We can't easily mock the full stories loop, but we can verify
# the return value semantics are correct.
# If stories loop returns "SESSION_OVER", the main flow breaks.
# If it returns "FEED_EXHAUSTED", the main flow can switch feeds.
# This test checks the contract: after a sub-feed completes naturally,
# the session should NOT be over unless dopamine says so.
from GramAddict.core.bot_flow import _run_zero_latency_stories_loop
import inspect
source = inspect.getsource(_run_zero_latency_stories_loop)
# The function must return FEED_EXHAUSTED when stories are done naturally
assert "FEED_EXHAUSTED" in source, (
"StoriesFeed loop still returns 'SESSION_OVER' when stories are exhausted. "
"This kills the entire session after just 3-5 stories! "
"Must return 'FEED_EXHAUSTED' so the main loop switches to another feed."
)
def test_main_loop_handles_feed_exhausted(self):
"""
The main session loop must handle 'FEED_EXHAUSTED' by switching
to another available feed target, NOT by breaking.
"""
from GramAddict.core import bot_flow
import inspect
source = inspect.getsource(bot_flow.start_bot)
assert "FEED_EXHAUSTED" in source, (
"Main loop does not handle 'FEED_EXHAUSTED' result. "
"When a sub-feed is exhausted, the bot must switch to another feed."
)

View File

@@ -0,0 +1,36 @@
import pytest
from unittest.mock import patch, MagicMock
from GramAddict.core.llm_provider import query_llm
def test_query_llm_passes_timeout_to_requests():
"""
Verifies that the query_llm wrapper passes the configured
timeout down to the requests.post call, enabling long
generation tasks like comments to complete without crashing.
"""
mock_response = MagicMock()
mock_response.json.return_value = {"response": "test"}
mock_response.raise_for_status.return_value = None
with patch("GramAddict.core.llm_provider.requests.post", return_value=mock_response) as mock_post:
# Act with custom 180s timeout
# Using format_json=False because Ollama branch accesses resp_json["response"]
query_llm("http://localhost:11434", "test_model", "test_prompt", timeout=180, format_json=False)
# Assert
mock_post.assert_called_once()
_, kwargs = mock_post.call_args
assert kwargs.get("timeout") == 180, "query_llm failed to pass the custom timeout to requests.post"
def test_query_llm_default_timeout_is_configurable():
"""
Verifies that if no timeout is passed, by default we still use a configured or sensible value (60s).
"""
mock_response = MagicMock()
mock_response.json.return_value = {"response": "test"}
with patch("GramAddict.core.llm_provider.requests.post", return_value=mock_response) as mock_post:
query_llm("http://localhost:11434", "test_model", "test_prompt", format_json=False)
_, kwargs = mock_post.call_args
assert kwargs.get("timeout") == 180, "query_llm default timeout should act as fallback"

View File

@@ -0,0 +1,66 @@
import pytest
from unittest.mock import patch, MagicMock, call
from GramAddict.core.bot_flow import _interact_with_profile
from GramAddict.core.session_state import SessionState
class FakeConfig:
def __init__(self):
class Args:
follow_percentage = 100
likes_percentage = 100
likes_count = "1-1"
self.args = Args()
def test_profile_grid_sync_delay_after_follow():
"""
Verifies that _interact_with_profile enforces a sleep delay
after a successful follow and BEFORE searching for the grid,
preventing UI automation from dumping mid-animation.
It now tracks the autonomous QNavGraph calls.
"""
mock_device = MagicMock()
mock_configs = FakeConfig()
mock_session_state = MagicMock(spec=SessionState)
mock_session_state.check_limit.return_value = False
manager = MagicMock()
with patch("GramAddict.core.q_nav_graph.QNavGraph") as MockQNavGraph, \
patch("GramAddict.core.bot_flow.sleep") as mock_sleep, \
patch("GramAddict.core.bot_flow.random.random", return_value=0.0): # Guarantee logic branches
mock_nav_instance = MagicMock()
mock_nav_instance._execute_transition.return_value = True # Always succeed transition
MockQNavGraph.return_value = mock_nav_instance
manager.attach_mock(mock_nav_instance._execute_transition, 'execute_transition')
manager.attach_mock(mock_sleep, 'sleep')
# Act
_interact_with_profile(mock_device, mock_configs, "test_user", mock_session_state, 1.0, MagicMock())
follow_idx = -1
grid_idx = -1
for i, mock_call in enumerate(manager.mock_calls):
# mock_call format: ('name', (args,), {kwargs})
if mock_call[0] == 'execute_transition':
args = mock_call[1]
if args and args[0] == "tap_follow_button":
follow_idx = i
elif args and args[0] == "tap_grid_first_post":
grid_idx = i
assert follow_idx != -1, "Follow transition was not executed"
assert grid_idx != -1, "Grid transition was not executed"
sleep_between = False
for i in range(follow_idx + 1, grid_idx):
if manager.mock_calls[i][0] == 'sleep':
sleep_between = True
assert sleep_between is True, (
"CRITICAL SYNC FAILURE: Found no sleep between Follow Confirmation and Grid search. "
"This causes VLM hallucinations mid-animation."
)

View File

@@ -0,0 +1,72 @@
import pytest
import GramAddict.core.telepathic_engine as telepathic_engine
from GramAddict.core.telepathic_engine import TelepathicEngine
def test_structural_guard_rejects_own_story_for_post_username():
"""
TDD Test: Reproduces the bug where Telepathic Engine might select the user's
OWN profile picture ("Your Story" in the Home Feed tray) when the intent
is to tap the post author's username.
"""
engine = TelepathicEngine()
screen_height = 2400
# Mock node representing the user's "Your Story" circle at the top
# It contains "story" or "your story", has low Y (top of screen)
your_story_node = {
"semantic_string": "description: 'Your Story', id context: 'row feed photo profile imageview'",
"y": 250, # Top story tray
"class_name": "android.widget.ImageView"
}
# Intent
intent = "tap post username"
# Expected behavior: Structural sanity check must REJECT this node to prevent
# clicking our own story/profile
is_valid = engine._structural_sanity_check(your_story_node, intent, screen_height)
assert is_valid is False, "Structural Guard failed to reject 'Your Story' when looking for 'post username'."
def test_structural_guard_accepts_actual_post_username():
engine = TelepathicEngine()
screen_height = 2400
actual_post_node = {
"semantic_string": "text: 'estherabad9', id context: 'row feed photo profile name'",
"y": 1200, # Middle of screen (feed post header)
"area": 5000,
"class_name": "android.widget.TextView"
}
intent = "tap post username"
is_valid = engine._structural_sanity_check(actual_post_node, intent, screen_height)
assert is_valid is True, "Structural Guard incorrectly rejected the actual post username."
def test_structural_guard_rejects_own_username_story():
"""
TDD Test: Reproduces 2026-04-16 23:18 bug where bot selected 'marisaundmarc's story'
instead of an unseen story from ANOTHER user.
"""
engine = TelepathicEngine()
screen_height = 2400
# Simulate current user is marisaundmarc
engine._get_current_username = lambda: "marisaundmarc"
# Mock node representing the user's OWN story, which contains their username
own_story_node = {
"semantic_string": "description: 'marisaundmarc\\'s story, 0 of 27, Unseen.', id context: 'avatar image view'",
"y": 250, # Top story tray
"class_name": "android.widget.ImageView"
}
intent = "profile picture avatar story ring"
# Should reject the user's own profile because clicking it means we edit/view our own story
# instead of doing interactions with prospects.
is_valid = engine._structural_sanity_check(own_story_node, intent, screen_height)
assert is_valid is False, "Structural Guard failed to reject the bot's OWN username story."

View File

@@ -0,0 +1,32 @@
import pytest
from GramAddict.core.telepathic_engine import TelepathicEngine
def test_media_intent_rejects_grid_containers():
"""
TDD Test: Reproduces the bug where intents containing "post" but
targeting specific grid items (like "first image post in profile grid")
were bypassing the MAX_CONTAINER_AREA guard, allowing massive
RecyclerView parent containers to be selected.
"""
engine = TelepathicEngine()
screen_height = 2400
# Mock node representing a massive RecyclerView containing the entire grid
# Area is 1080 * 2400 = 2592000 > MAX_CONTAINER_AREA (500000)
massive_grid_container = {
"semantic_string": "id context: 'swipeable nav view pager inner recycler view'",
"area": 2592000,
"y": 1200,
"class_name": "androidx.recyclerview.widget.RecyclerView",
"resource_id": "com.instagram.android:id/swipeable_nav_view_pager_inner_recycler_view"
}
# Intent
intent = "first image post in profile grid"
# Expected behavior: Structural sanity check must REJECT this node because
# although it's a "post" intent, it is specifically looking for an item within a grid/list,
# meaning we should NOT click massive screen-sized containers.
is_valid = engine._structural_sanity_check(massive_grid_container, intent, screen_height)
assert is_valid is False, "Structural Guard failed to reject massive Grid Container when specifically looking for a grid item."