chore: FSD stabilization, strict TDD enforcement, and unlearn mechanism

- implemented self-healing unlearn for Qdrant false positives
- centralized testing logic in conftest
- documented core rules, ai standards, and goap philosophy
- purged old dev scratchpads
This commit is contained in:
2026-04-24 13:28:32 +02:00
parent 75009d91a2
commit 30724d3c03
196 changed files with 8519 additions and 1595 deletions

View File

@@ -38,26 +38,21 @@ def verify_and_switch_account(device, nav_graph, target_username):
logger.warning(f"🔄 [Identity Guard] Account mismatch detected! Switching to '{target_username}'...")
# 3. Find the Profile Tab to long press
# 3. Find the Profile Tab to long press using Telepathic Engine (Blank Start)
profile_tab = None
try:
for elem in root.iter("node"):
res_id = elem.attrib.get("resource-id", "")
# MUST be exact match or end with /profile_tab to avoid matching profile_tab_icon_view
if res_id.endswith(":id/profile_tab"):
bounds_str = elem.attrib.get("bounds")
if bounds_str:
coords = re.findall(r"\d+", bounds_str)
if len(coords) == 4:
x = (int(coords[0]) + int(coords[2])) // 2
y = (int(coords[1]) + int(coords[3])) // 2
profile_tab = (x, y)
break
except Exception:
pass
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
# We ask the semantic engine to find the profile tab, ensuring 100% ID-agnostic behavior
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3)
if profile_tab_node:
profile_tab = (profile_tab_node["x"], profile_tab_node["y"])
except Exception as e:
logger.warning(f"Error resolving profile tab via telepathic engine: {e}")
if not profile_tab:
logger.error("❌ [Identity Guard] Cannot find profile_tab to initiate account switch!")
logger.error("❌ [Identity Guard] Cannot find profile_tab semantically to initiate account switch!")
return False
# Long press to open account selector

View File

@@ -1,3 +1,17 @@
"""
Active Inference Engine v2 — The Bot's Immune System.
Bayesian Active Inference: predicts future UI states before acting,
evaluates predictions against reality, and steers behavior based on
accumulated surprise (Free Energy).
v2 Enhancements:
- Consecutive error tracking → automatic policy escalation
- Interaction throttling → reduces follow/like probability under high surprise
- Session abort recommendation → when environment is fundamentally unstable
- Rich prediction context → tracks WHAT was expected vs. WHAT was found
"""
import logging
import time
import math
@@ -6,20 +20,32 @@ from colorama import Fore
logger = logging.getLogger(__name__)
class ActiveInferenceEngine:
"""
Bayesian Active Inference Engine.
Calculates Free Energy (Surprise) based on prediction errors in the
Instagram environment. Steers the agent's 'Thermodynamic Policy'.
Policies:
- STABLE: Free energy < 0.75. Normal operation. All interactions enabled.
- CAUTIOUS: Free energy 0.75-1.2. Reduced interaction probability. Longer waits.
- DORMANT: Free energy > 1.2. Minimal interactions. Maximum sleep. May recommend abort.
"""
def __init__(self, username):
self.username = username
self.free_energy = 0.0
self.surprise_threshold = 0.75
self.last_update = time.time()
self.policy = "STABLE" # STABLE, CAUTIOUS, DORMANT
self.policy = "STABLE" # STABLE, CAUTIOUS, DORMANT
self.expectation_history = []
# v2: Consecutive error tracking for escalation
self._consecutive_prediction_errors = 0
self._total_predictions = 0
self._total_errors = 0
self._session_start = time.time()
def calculate_surprise(self, predicted_outcome: float, observed_outcome: float):
"""
Bayesian surprise calculation (simplified Kullback-Leibler divergence).
@@ -60,20 +86,43 @@ class ActiveInferenceEngine:
"""
Evaluates the last prediction against reality.
Returns True if reality matches prediction, False otherwise (Prediction Error).
v2: Tracks consecutive errors and escalates policy automatically.
"""
if not self.expectation_history:
return True
expected_signature = self.expectation_history.pop()
self._total_predictions += 1
matched = any(sig.lower() in context_xml.lower() for sig in expected_signature)
if matched:
self._consecutive_prediction_errors = 0
self.calculate_surprise(1.0, 1.0)
return True
else:
logger.warning(f"⚖️ [Shadow Mode] Prediction Error! Did not find {expected_signature} in resulting UI.", extra={"color": f"{Fore.RED}"})
self._consecutive_prediction_errors += 1
self._total_errors += 1
logger.warning(f"⚖️ [Shadow Mode] Prediction Error #{self._consecutive_prediction_errors}! "
f"Did not find {expected_signature} in resulting UI.", extra={"color": f"{Fore.RED}"})
self.calculate_surprise(1.0, 0.0)
# v2: Consecutive error escalation
if self._consecutive_prediction_errors >= 5:
self.policy = "DORMANT"
logger.error(
f"🚨 [Active Inference] {self._consecutive_prediction_errors} consecutive prediction errors! "
f"Environment is fundamentally unstable. DORMANT mode engaged.",
extra={"color": f"{Fore.RED}"}
)
elif self._consecutive_prediction_errors >= 3:
self.policy = "CAUTIOUS"
logger.warning(
f"⚠️ [Active Inference] {self._consecutive_prediction_errors} consecutive errors. "
f"Switching to CAUTIOUS policy.",
extra={"color": f"{Fore.YELLOW}"}
)
# ── Dojo Data Engine Hook ──
# When prediction fails, explicitly submit the snapshot for shadow-compilation
try:
@@ -99,3 +148,58 @@ class ActiveInferenceEngine:
if self.policy == "CAUTIOUS":
return 2.0
return 1.0
# ──────────────────────────────────────────────
# v2: New behavioral steering methods
# ──────────────────────────────────────────────
def get_interaction_probability(self) -> float:
"""
Returns a probability multiplier [0.0 - 1.0] for interaction decisions.
Under STABLE: 1.0 (full interaction rate)
Under CAUTIOUS: 0.5 (halved interaction rate)
Under DORMANT: 0.1 (minimal interaction — only high-confidence targets)
This directly modifies follow/like/comment probability in the feed loop.
"""
if self.policy == "DORMANT":
return 0.1
if self.policy == "CAUTIOUS":
return 0.5
return 1.0
def should_abort_session(self) -> bool:
"""
Recommends session abort when the environment is fundamentally broken.
Triggers:
- 5+ consecutive prediction errors (UI is completely unexpected)
- Free energy > 2.0 (accumulated instability beyond recovery)
The caller (bot_flow) can choose to honor this or override.
"""
if self._consecutive_prediction_errors >= 5:
return True
if self.free_energy > 2.0:
return True
return False
def get_error_rate(self) -> float:
"""Returns the session-wide prediction error rate."""
if self._total_predictions == 0:
return 0.0
return self._total_errors / self._total_predictions
def get_diagnostics(self) -> dict:
"""Returns a diagnostic snapshot for logging/telemetry."""
return {
"free_energy": round(self.free_energy, 4),
"policy": self.policy,
"consecutive_errors": self._consecutive_prediction_errors,
"total_predictions": self._total_predictions,
"total_errors": self._total_errors,
"error_rate": round(self.get_error_rate(), 4),
"session_uptime_minutes": round((time.time() - self._session_start) / 60, 1),
"should_abort": self.should_abort_session(),
}

View File

@@ -0,0 +1,215 @@
"""
Behavior Plugin Architecture — Composable, Testable Bot Actions.
Design goals:
1. Each behavior is a self-contained plugin with a clear lifecycle
2. Plugins declare prerequisites (what screen state they require)
3. Plugins are registered in a priority-sorted registry
4. The feed loop queries the registry: "which plugins want to act on this post?"
5. Each plugin can be tested in complete isolation
This is the "neural pathway" system — instead of one monolithic brain (bot_flow.py),
the bot has specialized pathways that fire when their conditions are met.
Tesla analogy: Instead of one "drive" function, there are composable behaviors
(lane-keep, auto-park, summon) that activate when relevant.
"""
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
logger = logging.getLogger(__name__)
@dataclass
class BehaviorContext:
"""
Shared context passed to every behavior plugin.
Contains everything a behavior needs to make decisions and act.
"""
device: Any # Android device facade
configs: Any # User configuration
session_state: Any # Current session state
cognitive_stack: Dict[str, Any] # Cognitive engines (growth, resonance, etc.)
context_xml: str = "" # Current screen XML dump
sleep_mod: float = 1.0 # Active Inference sleep multiplier
post_data: Optional[Dict] = None # Extracted post content
username: str = "" # Current target username (if applicable)
@dataclass
class BehaviorResult:
"""
Result returned by a behavior plugin after execution.
Used by the orchestrator to decide what happens next.
"""
executed: bool = False # Did the behavior actually do something?
should_continue: bool = True # Should the feed loop continue to next post?
should_skip: bool = False # Should we skip to the next post immediately?
interactions: int = 0 # Number of interactions performed
metadata: Dict[str, Any] = field(default_factory=dict) # Plugin-specific data
class BehaviorPlugin(ABC):
"""
Base class for all behavior plugins.
Lifecycle:
1. `can_activate(ctx)` — Should this behavior fire for this context?
2. `priority` — If multiple behaviors can activate, higher priority goes first.
3. `execute(ctx)` — Run the behavior.
Rules:
- Plugins must be stateless between posts (state lives in session_state)
- Plugins must handle their own errors (never crash the feed loop)
- Plugins must respect session limits via ctx.session_state
"""
@property
@abstractmethod
def name(self) -> str:
"""Unique identifier for this behavior."""
...
@property
def priority(self) -> int:
"""
Execution priority. Higher = runs first.
Guidelines:
- 100+: Safety/guard behaviors (ad detection, block detection)
- 50-99: Primary interactions (like, follow, comment)
- 10-49: Secondary interactions (carousel, story view)
- 1-9: Observational behaviors (scraping, analytics)
"""
return 50
@property
def exclusive(self) -> bool:
"""
If True, no other behavior can run after this one on the same post.
Used for guard behaviors that abort interaction (e.g., ad detection).
"""
return False
@abstractmethod
def can_activate(self, ctx: BehaviorContext) -> bool:
"""
Returns True if this behavior should fire for the given context.
Must be cheap to evaluate (no device interactions).
"""
...
@abstractmethod
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""
Execute the behavior. Must handle all errors internally.
Returns a BehaviorResult describing what happened.
"""
...
def __repr__(self):
return f"<{self.__class__.__name__} name={self.name} priority={self.priority}>"
class PluginRegistry:
"""
Central registry for behavior plugins.
Manages plugin registration, priority sorting, and orchestrated execution.
Thread-safe singleton.
"""
_instance = None
@classmethod
def get_instance(cls) -> "PluginRegistry":
if cls._instance is None:
cls._instance = cls()
return cls._instance
@classmethod
def reset(cls):
cls._instance = None
def __init__(self):
self._plugins: List[BehaviorPlugin] = []
self._sorted = False
def register(self, plugin: BehaviorPlugin):
"""Register a behavior plugin."""
# Prevent duplicate registration
for existing in self._plugins:
if existing.name == plugin.name:
logger.debug(f"Plugin '{plugin.name}' already registered. Skipping.")
return
self._plugins.append(plugin)
self._sorted = False
logger.info(f"🧩 [Plugin] Registered: {plugin.name} (priority={plugin.priority})")
def unregister(self, name: str):
"""Remove a plugin by name."""
self._plugins = [p for p in self._plugins if p.name != name]
self._sorted = False
def _ensure_sorted(self):
"""Sort plugins by priority (highest first)."""
if not self._sorted:
self._plugins.sort(key=lambda p: p.priority, reverse=True)
self._sorted = True
@property
def plugins(self) -> List[BehaviorPlugin]:
"""Returns all plugins, sorted by priority."""
self._ensure_sorted()
return list(self._plugins)
def get_active_plugins(self, ctx: BehaviorContext) -> List[BehaviorPlugin]:
"""Returns plugins that can activate for the given context, sorted by priority."""
self._ensure_sorted()
active = []
for plugin in self._plugins:
try:
if plugin.can_activate(ctx):
active.append(plugin)
except Exception as e:
logger.error(f"🧩 [Plugin] Error checking {plugin.name}.can_activate: {e}")
return active
def execute_all(self, ctx: BehaviorContext) -> List[BehaviorResult]:
"""
Execute all active plugins in priority order.
Stops early if an exclusive plugin fires (e.g., ad guard).
Returns list of results from all executed plugins.
"""
self._ensure_sorted()
results = []
for plugin in self._plugins:
try:
if not plugin.can_activate(ctx):
continue
logger.debug(f"🧩 [Plugin] Executing: {plugin.name}")
result = plugin.execute(ctx)
results.append(result)
if plugin.exclusive and result.executed:
logger.debug(f"🧩 [Plugin] {plugin.name} is exclusive. Stopping chain.")
break
except Exception as e:
logger.error(f"🧩 [Plugin] Error executing {plugin.name}: {e}")
results.append(BehaviorResult(executed=False, metadata={"error": str(e)}))
return results
def __len__(self):
return len(self._plugins)
def __contains__(self, name: str):
return any(p.name == name for p in self._plugins)

View File

@@ -0,0 +1,103 @@
"""
Carousel Browsing Behavior — Plugin Implementation.
Migrated from bot_flow.py's _interact_with_carousel function.
Now independently testable and composable.
"""
import logging
import random
from time import sleep
from GramAddict.core.behaviors import BehaviorPlugin, BehaviorContext, BehaviorResult
from GramAddict.core.perception.feed_analysis import has_carousel_in_view
from GramAddict.core.physics.humanized_input import humanized_horizontal_swipe
logger = logging.getLogger(__name__)
class CarouselBrowsingPlugin(BehaviorPlugin):
"""
Browses carousel posts with humanized swiping and curiosity dwells.
Activation: When a carousel indicator is present in the current XML.
Priority: 20 (secondary interaction — runs after primary like/follow decisions).
"""
@property
def name(self) -> str:
return "carousel_browsing"
@property
def priority(self) -> int:
return 20 # Secondary interaction tier
def can_activate(self, ctx: BehaviorContext) -> bool:
"""Activates when carousel indicators are present on screen."""
if not ctx.context_xml:
return False
# Check config — carousel_percentage controls activation probability
carousel_pct = float(getattr(ctx.configs.args, "carousel_percentage", 0)) / 100.0
if carousel_pct <= 0:
return False
return has_carousel_in_view(ctx.context_xml)
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""Browse carousel with humanized swiping."""
from colorama import Fore
carousel_pct = float(getattr(ctx.configs.args, "carousel_percentage", 0)) / 100.0
# Probabilistic execution (config controls how often we interact)
if random.random() >= carousel_pct:
return BehaviorResult(executed=False)
# Parse swipe count from config
carousel_count_str = getattr(ctx.configs.args, "carousel_count", "1-2")
try:
min_c, max_c = map(int, carousel_count_str.split('-'))
count = random.randint(min_c, max_c)
except Exception:
count = 1
logger.info(
f"📸 [Carousel] Interacting with carousel. Swiping {count} times...",
extra={"color": f"{Fore.CYAN}"}
)
info = ctx.device.get_info()
w = info.get("displayWidth", 1080)
h = info.get("displayHeight", 2400)
# Curiosity Peak: One slide gets extra attention
curiosity_slide = random.randint(0, count - 1) if count > 0 else 0
for i in range(count):
# Normal transition wait
sleep(random.uniform(1.5, 3.5) * ctx.sleep_mod)
# ── Curiosity Dwell ──
if i == curiosity_slide:
dwell = random.uniform(3.0, 7.0)
logger.debug(
f"📸 [Carousel] Curiosity Peak hit on slide {i+1}. "
f"Gazing for {dwell:.1f}s..."
)
sleep(dwell * ctx.sleep_mod)
# Horizontal swipe: Right to left
humanized_horizontal_swipe(
ctx.device,
start_x=w * 0.8,
end_x=w * 0.2,
y=h * 0.5,
duration_ms=250
)
sleep(random.uniform(1.0, 2.0) * ctx.sleep_mod)
return BehaviorResult(
executed=True,
interactions=count,
metadata={"slides_viewed": count, "curiosity_slide": curiosity_slide}
)

View File

@@ -0,0 +1,73 @@
"""
Follow Behavior — Plugin Implementation.
Follows a target user's profile with session limit awareness.
Migrated from the follow section of _interact_with_profile.
"""
import logging
import random
from time import sleep
from GramAddict.core.behaviors import BehaviorPlugin, BehaviorContext, BehaviorResult
logger = logging.getLogger(__name__)
class FollowPlugin(BehaviorPlugin):
"""
Follows a target user from their profile page.
Activation: When follow_percentage > 0 and session follow limit not reached.
Priority: 60 (primary interaction tier).
"""
@property
def name(self) -> str:
return "follow"
@property
def priority(self) -> int:
return 60 # Primary interaction tier
def can_activate(self, ctx: BehaviorContext) -> bool:
"""Activates when follow is enabled and limits not reached."""
from GramAddict.core.session_state import SessionState
follow_pct = float(getattr(ctx.configs.args, "follow_percentage", 0)) / 100.0
if follow_pct <= 0:
return False
if ctx.session_state.check_limit(SessionState.Limit.FOLLOWS):
return False
return True
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""Follow the target user."""
follow_pct = float(getattr(ctx.configs.args, "follow_percentage", 0)) / 100.0
rnd = random.random()
logger.info(
f"⚙️ [Decision] Profile Follow -> Config: {follow_pct*100}% "
f"(Roll: {rnd:.2f}) -> Proceed: {rnd < follow_pct}"
)
if rnd >= follow_pct:
return BehaviorResult(executed=False)
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(ctx.device)
if nav_graph.do("tap follow button"):
logger.info(f"🤝 [Follow] Followed @{ctx.username}")
ctx.session_state.totalFollowed[ctx.username] = 1
# Buffer for follow animations to close
sleep(random.uniform(1.8, 3.2) * ctx.sleep_mod)
return BehaviorResult(
executed=True,
interactions=1,
metadata={"followed": ctx.username}
)
return BehaviorResult(executed=False, metadata={"reason": "nav_failed"})

View File

@@ -0,0 +1,149 @@
"""
Grid Like Behavior — Plugin Implementation.
Likes posts from a target user's profile grid.
Migrated from the grid-likes section of _interact_with_profile.
"""
import logging
import random
from time import sleep
from GramAddict.core.behaviors import BehaviorPlugin, BehaviorContext, BehaviorResult
from GramAddict.core.physics.humanized_input import humanized_click, humanized_scroll
from GramAddict.core.physics.timing import wait_for_post_loaded
logger = logging.getLogger(__name__)
class GridLikePlugin(BehaviorPlugin):
"""
Opens profile grid and likes posts with humanized behavior.
Activation: When likes_percentage > 0 and session like limit not reached.
Priority: 50 (primary interaction tier, after follow).
"""
@property
def name(self) -> str:
return "grid_like"
@property
def priority(self) -> int:
return 50 # Primary interaction, after follow
def can_activate(self, ctx: BehaviorContext) -> bool:
"""Activates when likes are enabled, limits not reached, and we are on a profile."""
from GramAddict.core.session_state import SessionState
likes_pct = float(getattr(ctx.configs.args, "likes_percentage", 0)) / 100.0
if likes_pct <= 0:
return False
if ctx.session_state.check_limit(SessionState.Limit.LIKES):
return False
# ── STRUCTURAL GUARD ──
# Prevent execution in Reels/HomeFeed. Must be on a profile.
if ctx.context_xml:
if "profile_header" not in ctx.context_xml.lower() and "followers" not in ctx.context_xml.lower():
return False
return True
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""Open grid and like posts."""
likes_pct = float(getattr(ctx.configs.args, "likes_percentage", 0)) / 100.0
rnd = random.random()
logger.info(
f"⚙️ [Decision] Profile Grid Likes -> Config: {likes_pct*100}% "
f"(Roll: {rnd:.2f}) -> Proceed: {rnd < likes_pct}"
)
if rnd >= likes_pct:
return BehaviorResult(executed=False)
# Parse like count
likes_count_str = getattr(ctx.configs.args, "likes_count", "1-2")
try:
min_l, max_l = map(int, likes_count_str.split('-'))
count = random.randint(min_l, max_l)
except Exception:
count = 1
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(ctx.device)
if not nav_graph.do("tap first image post in profile grid"):
return BehaviorResult(executed=False, metadata={"reason": "grid_nav_failed"})
if not wait_for_post_loaded(ctx.device, timeout=5):
logger.warning(f"❌ Post failed to open from profile grid of @{ctx.username}.")
return BehaviorResult(executed=False, metadata={"reason": "post_load_failed"})
logger.info(
f"❤️ [Grid Like] Opening grid to drop {count} likes on @{ctx.username}..."
)
info = ctx.device.get_info()
w = info.get("displayWidth", 1080)
h = info.get("displayHeight", 2400)
growth = ctx.cognitive_stack.get("growth_brain")
total_liked = 0
for i in range(count):
xml_dump = ctx.device.dump_hierarchy()
if not isinstance(xml_dump, str):
xml_dump = ""
xml_dump_lower = xml_dump.lower()
is_reel = "reel_viewer" in xml_dump_lower or "clips_viewer" in xml_dump_lower
is_liked = (
"gefällt mir nicht mehr" in xml_dump_lower or
"unlike" in xml_dump_lower or
'content-desc="liked"' in xml_dump_lower
)
# Double-tap ~40% of the time on standard images
use_double_tap = growth.wants_to_double_tap(is_reel=is_reel) if growth else False
if use_double_tap:
if is_liked:
logger.debug(f"Skipped liking grid post {i+1}/{count} (already liked)")
else:
offset_x = random.randint(int(w * 0.2), int(w * 0.8))
offset_y = random.randint(int(h * 0.3), int(h * 0.7))
logger.info(
f"❤️ [Grid Like] Double-Tapping organically at ({offset_x}, {offset_y})"
)
humanized_click(ctx.device, offset_x, offset_y, double=True, sleep_mod=ctx.sleep_mod)
ctx.session_state.totalLikes += 1
total_liked += 1
logger.debug(f"Liked grid post {i+1}/{count} via Double-Tap")
else:
if nav_graph.do("tap like button"):
ctx.session_state.totalLikes += 1
total_liked += 1
logger.debug(f"Liked grid post {i+1}/{count} via Heart Button")
else:
logger.debug(f"Skipped liking grid post {i+1}/{count}")
sleep(random.uniform(1.0, 2.0) * ctx.sleep_mod)
if is_reel:
logger.debug("🎬 Detected Reel. Swiping full-screen up.")
humanized_scroll(ctx.device, is_skip=True)
else:
humanized_scroll(ctx.device, is_skip=False)
sleep(random.uniform(1.5, 3.0) * ctx.sleep_mod)
ctx.device.press("back")
sleep(random.uniform(1.0, 2.0) * ctx.sleep_mod)
return BehaviorResult(
executed=True,
interactions=total_liked,
metadata={"posts_viewed": count, "posts_liked": total_liked}
)

View File

@@ -0,0 +1,119 @@
"""
Profile Guard Behavior — Plugin Implementation.
Safety guards that reject profiles before any interactions occur:
- Private accounts
- Empty accounts
- Close friends (when configured)
- Visual vibe check (AI aesthetic quality)
Priority 100 (highest, exclusive) — if a guard fires, no other behavior runs.
"""
import logging
from GramAddict.core.behaviors import BehaviorPlugin, BehaviorContext, BehaviorResult
logger = logging.getLogger(__name__)
class ProfileGuardPlugin(BehaviorPlugin):
"""
Guards against interacting with profiles that should be skipped.
Exclusive: if this fires, no further interactions happen on this profile.
"""
@property
def name(self) -> str:
return "profile_guard"
@property
def priority(self) -> int:
return 100 # Highest — runs before everything
@property
def exclusive(self) -> bool:
return True # Stop all other plugins if guard fires
def can_activate(self, ctx: BehaviorContext) -> bool:
"""Only activates on Profile screens to prevent false-positives in Feed/Reels."""
nav_graph = ctx.cognitive_stack.get("nav_graph")
is_profile = nav_graph and nav_graph.current_state == "ProfileView"
return bool(ctx.username) and is_profile
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""Check profile guards. Returns executed=True + should_skip=True if rejected."""
from colorama import Fore
xml_check = ctx.context_xml
if not xml_check:
return BehaviorResult(executed=False)
xml_check_lower = xml_check.lower()
# Self-interaction guard
if (hasattr(ctx.session_state, 'my_username') and
ctx.username == ctx.session_state.my_username):
logger.info(
f"🤝 [Profile Guard] Skipping own profile @{ctx.username}."
)
return BehaviorResult(executed=True, should_skip=True,
metadata={"reason": "self_profile"})
# Private account guard
if ("this account is private" in xml_check_lower or
"konto ist privat" in xml_check_lower):
logger.info(
f"🔒 [Profile Guard] @{ctx.username} is private.",
extra={"color": f"{Fore.YELLOW}"}
)
return BehaviorResult(executed=True, should_skip=True,
metadata={"reason": "private"})
# Empty account guard
if ("no posts yet" in xml_check_lower or
"noch keine beiträge" in xml_check_lower):
logger.info(
f"📭 [Profile Guard] @{ctx.username} has no posts.",
extra={"color": f"{Fore.YELLOW}"}
)
return BehaviorResult(executed=True, should_skip=True,
metadata={"reason": "empty"})
# Close friends guard
if getattr(ctx.configs.args, "ignore_close_friends", False):
if ("enge freunde" in xml_check_lower or
"close friend" in xml_check_lower):
logger.info(
f"💚 [Profile Guard] @{ctx.username} is a Close Friend. Ignoring.",
extra={"color": "\033[32m"}
)
return BehaviorResult(executed=True, should_skip=True,
metadata={"reason": "close_friend"})
# Visual Vibe Check (AI Aesthetic Quality Guard)
import random
vibe_check_pct = float(getattr(ctx.configs.args, "visual_vibe_check_percentage", 0)) / 100.0
if vibe_check_pct > 0 and random.random() < vibe_check_pct:
from GramAddict.core.telepathic_engine import TelepathicEngine
telepathic = ctx.cognitive_stack.get("telepathic") or TelepathicEngine.get_instance()
persona_interests = ctx.cognitive_stack.get("persona_interests", []) if ctx.cognitive_stack else []
vibe_result = telepathic.evaluate_profile_vibe(ctx.device, persona_interests)
if vibe_result:
score = vibe_result.get("quality_score", 5)
matches_niche = vibe_result.get("matches_niche", True)
if score < 5 or not matches_niche:
logger.warning(
f"🚫 [Vibe Check] Profile @{ctx.username} rejected (Score: {score}, Niche: {matches_niche}). Reason: {vibe_result.get('reason')}"
)
return BehaviorResult(executed=True, should_skip=True,
metadata={"reason": "vibe_check_failed", "score": score})
else:
logger.info(
f"✅ [Vibe Check] Profile @{ctx.username} approved (Score: {score}). Continuing interaction.",
extra={"color": "\033[36m"}
)
# All guards passed — don't block further plugins
return BehaviorResult(executed=False)

View File

@@ -0,0 +1,101 @@
"""
Story Viewing Behavior — Plugin Implementation.
Watches a target user's stories with humanized timing and navigation.
Migrated from the story-viewing section of _interact_with_profile.
"""
import logging
import random
from time import sleep
from GramAddict.core.behaviors import BehaviorPlugin, BehaviorContext, BehaviorResult
from GramAddict.core.physics.humanized_input import humanized_click
from GramAddict.core.physics.timing import wait_for_story_loaded
logger = logging.getLogger(__name__)
class StoryViewPlugin(BehaviorPlugin):
"""
Views a target user's stories from their profile.
Activation: When stories_percentage > 0 and user has stories.
Priority: 40 (runs before likes/follows since it navigates away from profile).
"""
@property
def name(self) -> str:
return "story_view"
@property
def priority(self) -> int:
return 40 # Before likes/follows (since it navigates away)
def can_activate(self, ctx: BehaviorContext) -> bool:
"""Activates when story viewing is enabled in config."""
stories_pct = float(getattr(ctx.configs.args, "stories_percentage", 0)) / 100.0
return stories_pct > 0
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
"""View stories with humanized timing."""
from colorama import Fore
stories_pct = float(getattr(ctx.configs.args, "stories_percentage", 0)) / 100.0
# Probabilistic check
if random.random() >= stories_pct:
return BehaviorResult(executed=False)
# Parse story count
stories_count_str = getattr(ctx.configs.args, "stories_count", "1-2")
try:
min_st, max_st = map(int, stories_count_str.split('-'))
count = random.randint(min_st, max_st)
except Exception:
count = 1
# Check for story ring
xml_dump = ctx.context_xml or ctx.device.dump_hierarchy()
xml_lower = xml_dump.lower()
has_story = (
"reel_ring" in xml_dump or
"'s unseen story" in xml_lower or
"has a new story" in xml_lower or
"story von" in xml_lower
)
if not has_story:
return BehaviorResult(executed=False, metadata={"reason": "no_story"})
# Navigate to story
from GramAddict.core.q_nav_graph import QNavGraph
nav_graph = QNavGraph(ctx.device)
if not nav_graph.do("tap story ring avatar"):
return BehaviorResult(executed=False, metadata={"reason": "nav_failed"})
# Wait for story to load
if not wait_for_story_loaded(ctx.device, timeout=5):
logger.warning(f"❌ Story failed to open for @{ctx.username}.")
return BehaviorResult(executed=False, metadata={"reason": "load_timeout"})
logger.info(f"📸 [Story] Viewing @{ctx.username}'s story ({count} times)...")
info = ctx.device.get_info()
w = info.get("displayWidth", 1080)
h = info.get("displayHeight", 2400)
for i in range(count):
sleep(random.uniform(2.0, 5.0) * ctx.sleep_mod)
if i < count - 1:
humanized_click(ctx.device, int(w * 0.9), int(h * 0.5), sleep_mod=ctx.sleep_mod)
ctx.device.press("back")
sleep(random.uniform(1.0, 2.0) * ctx.sleep_mod)
return BehaviorResult(
executed=True,
interactions=count,
metadata={"stories_viewed": count}
)

File diff suppressed because it is too large Load Diff

View File

@@ -185,6 +185,10 @@ class Config:
self.parser.add_argument("--profile-learning-percentage", help="Percentage of profiles to deeply scan before engaging", default="0")
self.parser.add_argument("--visual-vibe-check-percentage", help="Percentage of profiles to visually evaluate via screenshot before engaging", default="0")
self.parser.add_argument("--ignore-close-friends", action="store_true", help="Completely ignore posts, stories, and profiles of Close Friends (Enge Freunde)")
# Biomechanical Physics
self.parser.add_argument("--handedness", help="Dominant hand: 'right' or 'left'. Affects thumb arc direction and tap bias.", default="right")
# Phase 10: RAG Comment Learning & Extractor Settings
self.parser.add_argument("--ai-condenser-model", help="LLM used for condensing text/comments", default="qwen3.5:latest")

View File

@@ -7,6 +7,9 @@ import uuid
import time
from datetime import datetime
from GramAddict.core.qdrant_memory import QdrantBase
from GramAddict.core.physics.biomechanics import PhysicsBody, BezierGesture
from GramAddict.core.physics.sendevent_injector import SendEventInjector
logger = logging.getLogger(__name__)
@@ -91,6 +94,9 @@ class DarwinEngine(QdrantBase):
# 2. Non-linear cognitive latency (Micro-Jitters)
if profile["scroll_velocity"] != 1.0:
logger.debug(f" -> Simulating cognitive read latency (Micro-Jitters, Velocity: {profile['scroll_velocity']:.2f})")
body = PhysicsBody.get_session_instance(device)
injector = SendEventInjector.get_instance(device)
# Thumb starts on the right side of the screen to avoid clicking polls/tags in the center
cx = int(w * 0.8) + device.cm_to_pixels(random.uniform(-0.3, 0.3))
cy = h // 2
@@ -101,10 +107,14 @@ class DarwinEngine(QdrantBase):
start_y = int(cy + distance / 2)
end_y = int(cy - distance / 2)
# Add some x-axis noise for nonlinear human realism (~0.1 cm)
noise_x = device.cm_to_pixels(random.uniform(-0.1, 0.1))
# Use Bézier curve for the jitter
points = BezierGesture.scroll_curve(
(cx, start_y), (cx, end_y), body, n_points=6
)
timing = BezierGesture.compute_sigmoid_timing(len(points), duration * 1000)
device.swipe(cx, start_y, cx + noise_x, end_y, duration=duration)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
# 3. Micro Back-swipe (The Human Wobble)
if random.random() < profile["back_swipe_prob"]:
@@ -173,7 +183,9 @@ class DarwinEngine(QdrantBase):
# Instead of relying on a fragile bottom_sheet_container ID,
# we verify if the feed is visible. If not, the comment sheet is still open (or keyboard).
ui_dump = device.dump_hierarchy()
if 'resource-id="com.instagram.android:id/row_feed"' not in ui_dump and 'resource-id="com.instagram.android:id/button_like"' not in ui_dump:
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
if not telepath.find_best_node(ui_dump, "post like button heart", min_confidence=0.4, device=device):
logger.debug(" -> Not back on Home feed, pressing back again to close comment sheet/keyboard")
device.press("back")
time.sleep(1.0)
@@ -191,22 +203,46 @@ class DarwinEngine(QdrantBase):
"""
Simulates a thumb resting or slightly shifting on the glass.
Essential for breaking the 'robotically still' dwell periods.
Uses PhysicsBody for handedness-aware direction and fatigue-scaled amplitude.
"""
if random.random() < 0.2: # 20% chance for a wobble during dwell
if random.random() < 0.2: # 20% chance for a wobble during dwell
logger.debug("🧬 [Ghost Protocol] Micro-Wobble triggered.")
body = PhysicsBody.get_session_instance(device)
injector = SendEventInjector.get_instance(device)
info = device.get_info()
w = info.get("displayWidth", 1080)
h = info.get("displayHeight", 2400)
cx = int(w * 0.8) + device.cm_to_pixels(random.uniform(-0.3, 0.3))
# Start position from body (session-aware)
cx, cy = body.get_scroll_start()
# Override Y to center for wobble
cy = h // 2
# Keep the shift small but above Android's touch slop threshold (~8dp) so it visibly moves the UI
y_shift = device.cm_to_pixels(random.uniform(0.3, 0.6)) * random.choice([1, -1])
x_shift = device.cm_to_pixels(random.uniform(-0.2, 0.2))
# Fatigue scales wobble amplitude (tired = more sloppy)
amplitude = 1.0 + body.fatigue * 0.5
# Single slow slip (use native shell for exact OS-level injection without framework rounding)
duration_ms = int(random.uniform(150, 300))
device.shell(f"input swipe {int(cx)} {int(cy)} {int(cx + x_shift)} {int(cy + y_shift)} {duration_ms}")
# Keep the shift small but above Android's touch slop threshold (~8dp)
y_shift = device.cm_to_pixels(random.uniform(0.3, 0.6) * amplitude) * random.choice([1, -1])
x_shift = device.cm_to_pixels(random.uniform(-0.2, 0.2) * amplitude)
# Handedness bias: right-handers wobble right-down, left-handers left-down
if body.handedness == "right":
x_shift += device.cm_to_pixels(random.uniform(0, 0.1))
else:
x_shift -= device.cm_to_pixels(random.uniform(0, 0.1))
end_x = int(cx + x_shift)
end_y = int(cy + y_shift)
points = BezierGesture.scroll_curve(
(cx, cy), (end_x, end_y), body, n_points=5
)
duration_ms = random.uniform(150, 300)
timing = BezierGesture.compute_sigmoid_timing(len(points), duration_ms)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
def _get_historical_landscape(self):
try:

View File

@@ -8,6 +8,10 @@ from random import uniform
from GramAddict.core.utils import random_sleep
from functools import wraps
from GramAddict.core.physics.biomechanics import PhysicsBody, BezierGesture
from GramAddict.core.physics.sendevent_injector import SendEventInjector
logger = logging.getLogger(__name__)
def adb_retry(retries=3, delay=2.0):
@@ -150,14 +154,21 @@ class DeviceFacade:
w = right - left
h = bottom - top
# Biological fingerprint: Thumb bias (Bottom-Left cluster for Right-Handers)
cx_base = left + (w * 0.45)
cy_base = top + (h * 0.55)
# Biological fingerprint via PhysicsBody
body = PhysicsBody.get_session_instance(self)
# Thumb bias: right-handers land slightly left-below center
if body.handedness == "right":
cx_base = left + (w * 0.45)
cy_base = top + (h * 0.55)
else:
cx_base = left + (w * 0.55)
cy_base = top + (h * 0.55)
from random import gauss
# ~68% of clicks within 15% radius, 95% within 30%. Very organic.
sigma_x = max(1, w * 0.15)
sigma_y = max(1, h * 0.15)
# Fatigue increases spread
fatigue_mult = 1.0 + body.fatigue * 0.3
sigma_x = max(1, w * 0.15 * fatigue_mult)
sigma_y = max(1, h * 0.15 * fatigue_mult)
cx = int(gauss(cx_base, sigma_x))
cy = int(gauss(cy_base, sigma_y))
@@ -181,23 +192,30 @@ class DeviceFacade:
if y > 2100 or y < 200 or x < 50 or x > 1030:
self.deviceV2.shell(f"input tap {int(x)} {int(y)}")
return
from random import uniform
try:
self.deviceV2.touch.down(x, y)
# Human finger rest time (squish)
sleep(uniform(0.05, 0.15))
# Sloppy slip (Containment: Don't slip horizontally at the bottom edge, prevents Android App-Switch gestures)
slip_x = x + int(uniform(-4, 4)) if y < 2100 else x
slip_y = y + int(uniform(-4, 4))
self.deviceV2.touch.move(slip_x, slip_y)
sleep(uniform(0.01, 0.05))
self.deviceV2.touch.up(slip_x, slip_y)
body = PhysicsBody.get_session_instance(self)
injector = SendEventInjector.get_instance(self)
points = BezierGesture.tap_curve(x, y, body)
tap_duration = uniform(40, 90)
timing = BezierGesture.compute_sigmoid_timing(len(points), tap_duration)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
except Exception as e:
logger.debug(f"human_click failed, fallback: {e}")
self.deviceV2.shell(f"input tap {int(x)} {int(y)}")
logger.debug(f"human_click biomechanics failed, fallback: {e}")
try:
self.deviceV2.touch.down(x, y)
sleep(uniform(0.05, 0.15))
slip_x = x + int(uniform(-4, 4)) if y < 2100 else x
slip_y = y + int(uniform(-4, 4))
self.deviceV2.touch.move(slip_x, slip_y)
sleep(uniform(0.01, 0.05))
self.deviceV2.touch.up(slip_x, slip_y)
except Exception as e2:
logger.debug(f"human_click u2 failed, final fallback: {e2}")
self.deviceV2.shell(f"input tap {int(x)} {int(y)}")
@adb_retry()
def swipe_points(self, x1, y1, x2, y2, duration=0.1):
@@ -206,12 +224,32 @@ class DeviceFacade:
@adb_retry()
def human_swipe(self, start_x, start_y, end_x, end_y, duration=0.3):
# Simulate a realistic human swipe by keeping it simple.
# Android's ScrollView calculates fling velocity based on the final few points.
# If we use swipe_points with non-linear distances, it breaks the fling physics and produces stuttering or backwards scrolls.
# We just use native swipe with randomized small x-variance.
# 🛡️ [Gesture Guard] If swiping near the very edges, use native swipe to prevent
# system gesture clashes, unless it's a feed scroll (which is usually safe).
dur_ms = int(duration * 1000)
self.deviceV2.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {dur_ms}")
if start_x < 50 or start_x > 1030 or start_y < 200 or start_y > 2100:
self.deviceV2.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {dur_ms}")
return
try:
body = PhysicsBody.get_session_instance(self)
injector = SendEventInjector.get_instance(self)
# Use scroll_curve for vertical swipes, horizontal_swipe_curve for horizontal
is_horizontal = abs(end_x - start_x) > abs(end_y - start_y)
if is_horizontal:
points = BezierGesture.horizontal_swipe_curve((start_x, start_y), (end_x, end_y), body)
else:
points = BezierGesture.scroll_curve((start_x, start_y), (end_x, end_y), body)
# Use fling timing (J-curve) to ensure high terminal velocity so Android scroll physics works natively
timing = BezierGesture.compute_fling_timing(len(points), dur_ms)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
except Exception as e:
logger.debug(f"human_swipe biomechanics failed, fallback to native swipe: {e}")
self.deviceV2.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {dur_ms}")
@adb_retry()
def _get_current_app(self):

View File

@@ -0,0 +1,284 @@
"""
Evolution Engine — Autonomous Parameter Tuning via Genetic Algorithm.
Instead of hardcoded behavioral parameters (scroll probability, boredom decay,
resonance thresholds), this engine EVOLVES them based on real session outcomes.
Inspired by Tesla's real-time neural network weight updates from fleet data:
- Each session is a "generation"
- Session outcomes (follows gained, blocks, duration) determine "fitness"
- Winning parameters are preserved; losing parameters are mutated
- Hard safety bounds prevent the bot from evolving into dangerous territory
All parameters persist in Qdrant, surviving restarts.
"""
import logging
import random
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass, field, asdict
logger = logging.getLogger(__name__)
# ── Hard Safety Bounds ──
# These are absolute limits that CANNOT be exceeded through evolution.
# Think of them as the "physics" constraints of the system.
SAFETY_BOUNDS = {
"scroll_correction_probability": (0.05, 0.35), # Never below 5%, never above 35%
"boredom_decay_rate": (0.05, 0.5), # How fast boredom accumulates
"resonance_threshold": (0.3, 0.9), # Content quality filter
"interaction_cooldown_seconds": (1.0, 10.0), # Min pause between interactions
"max_follows_per_session": (5, 40), # Absolute follow cap
"max_likes_per_session": (10, 80), # Absolute like cap
"session_duration_target_minutes": (15, 120), # Session length target
"story_view_probability": (0.1, 0.8), # How often to view stories
}
@dataclass
class Genome:
"""
The bot's behavioral DNA — a set of evolvable parameters.
Each parameter has a current value and respects hard safety bounds.
"""
scroll_correction_probability: float = 0.15
boredom_decay_rate: float = 0.2
resonance_threshold: float = 0.7
interaction_cooldown_seconds: float = 2.5
max_follows_per_session: int = 15
max_likes_per_session: int = 30
session_duration_target_minutes: float = 45.0
story_view_probability: float = 0.4
# Metadata
generation: int = 0
best_fitness: float = 0.0
last_updated: float = field(default_factory=time.time)
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, d: dict) -> "Genome":
# Filter out unknown keys for forward-compatibility
known = {f.name for f in cls.__dataclass_fields__.values()}
filtered = {k: v for k, v in d.items() if k in known}
return cls(**filtered)
@dataclass
class SessionResult:
"""
Outcome metrics from a completed session.
Used to calculate fitness for the current genome.
"""
follows_gained: int = 0
likes_given: int = 0
stories_viewed: int = 0
blocks_received: int = 0
duration_minutes: float = 0.0
prediction_error_rate: float = 0.0 # From Active Inference
profiles_scraped: int = 0
class EvolutionEngine:
"""
Genetic algorithm for behavioral parameter optimization.
Lifecycle:
1. Load genome from Qdrant (or use defaults)
2. Bot uses genome parameters during session
3. After session, evaluate fitness
4. If fitness improved → lock genome (preserve winning params)
5. If fitness decreased → mutate genome (try new params)
6. Persist genome to Qdrant
"""
_instance = None
@classmethod
def get_instance(cls, username: str = None) -> "EvolutionEngine":
if cls._instance is None:
cls._instance = cls(username or "default")
return cls._instance
@classmethod
def reset(cls):
cls._instance = None
def __init__(self, username: str):
self.username = username
self.genome = Genome()
self._qdrant_connected = False
self._load_genome()
def _load_genome(self):
"""Load persisted genome from Qdrant, or use defaults."""
try:
from GramAddict.core.qdrant_memory import QdrantBase
self._db = QdrantBase("evolution_genomes_v1", vector_size=128)
if not self._db.is_connected:
logger.debug("[Evolution] Qdrant not available. Using default genome.")
return
self._qdrant_connected = True
# Try to recall existing genome
vec = self._db._get_embedding(f"genome_{self.username}")
if not vec:
return
results = self._db.client.query_points(
collection_name=self._db.collection_name,
query=vec,
limit=1,
score_threshold=0.95,
).points
if results:
payload = results[0].payload
genome_data = payload.get("genome", {})
if genome_data:
self.genome = Genome.from_dict(genome_data)
logger.info(
f"🧬 [Evolution] Loaded genome generation {self.genome.generation} "
f"(fitness: {self.genome.best_fitness:.3f})"
)
except Exception as e:
logger.debug(f"[Evolution] Failed to load genome: {e}")
def _save_genome(self):
"""Persist genome to Qdrant."""
if not self._qdrant_connected:
return
try:
vec = self._db._get_embedding(f"genome_{self.username}")
if not vec:
return
self.genome.last_updated = time.time()
payload = {
"username": self.username,
"genome": self.genome.to_dict(),
}
self._db.upsert_point(
f"genome_{self.username}",
payload,
vector=vec,
log_success=f"🧬 [Evolution] Saved genome generation {self.genome.generation}"
)
except Exception as e:
logger.debug(f"[Evolution] Failed to save genome: {e}")
def compute_fitness(self, result: SessionResult) -> float:
"""
Computes a fitness score [0.0 - 1.0] from session outcomes.
Reward:
- Follows gained (high value)
- Likes given (medium value)
- Stories viewed (low value)
- Longer sessions (moderate value)
Penalty:
- Blocks received (SEVERE penalty — 50% fitness reduction per block)
- High prediction error rate (moderate penalty)
"""
if result.blocks_received > 0:
# Blocks are catastrophic — any genome that triggers a block is unfit
block_penalty = 0.5 ** result.blocks_received
logger.warning(
f"🧬 [Evolution] BLOCK PENALTY: {result.blocks_received} blocks → "
f"fitness multiplier {block_penalty:.3f}"
)
else:
block_penalty = 1.0
# Normalize outcomes to [0, 1] range
follow_score = min(result.follows_gained / 20.0, 1.0) # Cap at 20
like_score = min(result.likes_given / 50.0, 1.0) # Cap at 50
story_score = min(result.stories_viewed / 20.0, 1.0) # Cap at 20
duration_score = min(result.duration_minutes / 60.0, 1.0) # Cap at 60 min
# Prediction accuracy bonus
accuracy_bonus = 1.0 - result.prediction_error_rate
# Weighted fitness
raw_fitness = (
follow_score * 0.35 + # Follows are most valuable
like_score * 0.20 + # Likes are secondary
story_score * 0.05 + # Stories are minor
duration_score * 0.15 + # Session stability matters
accuracy_bonus * 0.25 # Prediction accuracy = environmental mastery
)
fitness = raw_fitness * block_penalty
fitness = max(0.0, min(1.0, fitness)) # Clamp to [0, 1]
return round(fitness, 4)
def evolve(self, result: SessionResult):
"""
Evaluate session and evolve the genome.
If fitness improved → lock parameters (exploitation)
If fitness decreased → mutate parameters (exploration)
"""
fitness = self.compute_fitness(result)
logger.info(
f"🧬 [Evolution] Generation {self.genome.generation} fitness: {fitness:.4f} "
f"(best: {self.genome.best_fitness:.4f})"
)
if fitness >= self.genome.best_fitness:
# ── Exploitation: Lock winning parameters ──
logger.info(f"🧬 [Evolution] ✅ Fitness improved! Locking generation {self.genome.generation}.")
self.genome.best_fitness = fitness
else:
# ── Exploration: Mutate parameters ──
logger.info(f"🧬 [Evolution] 🔀 Fitness regressed. Mutating for generation {self.genome.generation + 1}.")
self._mutate()
self.genome.generation += 1
self._save_genome()
def _mutate(self, mutation_rate: float = 0.15):
"""
Mutate genome parameters within safety bounds.
Each parameter has a `mutation_rate` chance of being modified.
Mutations are small (±10-20% of current value) to ensure gradual evolution.
"""
for param_name, (low, high) in SAFETY_BOUNDS.items():
if random.random() > mutation_rate:
continue
current = getattr(self.genome, param_name, None)
if current is None:
continue
# Mutation: ±10-20% of range
param_range = high - low
delta = random.uniform(-0.2, 0.2) * param_range
new_value = current + delta
# Clamp to safety bounds
if isinstance(current, int):
new_value = int(max(low, min(high, round(new_value))))
else:
new_value = max(low, min(high, new_value))
old_value = current
setattr(self.genome, param_name, new_value)
logger.debug(f"🧬 [Mutation] {param_name}: {old_value}{new_value}")
def get_param(self, name: str, default: Any = None) -> Any:
"""Get a parameter value from the current genome."""
return getattr(self.genome, name, default)

View File

@@ -172,6 +172,14 @@ class ScreenIdentity:
def _classify_screen(self, ids, descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature=None):
"""Classify screen type using Semantic Memory with LLM fallback — NO hardcoded states."""
# Priority 0: Content-creation overlays that block ALL navigation.
# These full-screen Instagram UIs have no navigation tabs and trap the bot.
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
creation_flow_markers = ('quick_capture', 'gallery_cancel_button', 'creation_flow', 'reel_camera')
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
# Priority 1: Check Qdrant Semantic Cache
if signature and self.screen_memory and self.screen_memory.is_connected:
cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92)
@@ -263,7 +271,8 @@ class ScreenIdentity:
if screen_type == ScreenType.OWN_PROFILE or screen_type == ScreenType.OTHER_PROFILE:
if 'message' in desc_lower or 'nachricht' in desc_lower:
actions.append('tap message button')
if 'following' in desc_lower or 'abonniert' in desc_lower or 'following' in text_lower:
if ('following' in desc_lower or 'abonniert' in desc_lower or 'following' in text_lower
or 'profile_header_following' in ' '.join(resource_ids).lower()):
actions.append('tap following list')
# Grid items
@@ -466,6 +475,7 @@ class NavigationKnowledge:
# In-memory cache for rapidly avoiding traps during exploration
# In-memory cache for rapidly avoiding traps during exploration
self._learned_screen_mappings = {}
self._learned_traps = set()
def wipe(self):
"""Wipe all learned knowledge from Qdrant."""
@@ -636,6 +646,54 @@ class NavigationKnowledge:
pass
return None
def learn_trap(self, screen_type: ScreenType, action: str, trap_reason: str = "softlock"):
"""Aversively learn that an action on a screen is dangerous/useless."""
trap_key = f"{screen_type.name}_{action}"
self._learned_traps.add(trap_key)
if not self._db or not self._db.is_connected:
return
seed = f"trap_{trap_key}"
# Aversive vector is completely orthogonal to normal goals to prevent retrieval overlap
vec = self._db._get_embedding(f"trap_avoidance: {trap_key} {trap_reason}")
payload = {
"trap_screen": screen_type.name,
"trap_action": action,
"trap_reason": trap_reason,
"timestamp": time.time()
}
self._db.upsert_point(seed, payload, vector=vec)
logger.error(f"💀 [Aversive Learning] BURNED action '{action}' on {screen_type.name} due to: {trap_reason}")
def is_trap(self, screen_type: ScreenType, action: str) -> bool:
"""Check if an action on this screen is a known trap."""
trap_key = f"{screen_type.name}_{action}"
if trap_key in self._learned_traps:
return True
if not self._db or not self._db.is_connected:
return False
try:
from qdrant_client.models import Filter, FieldCondition, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(
must=[
FieldCondition(key="trap_screen", match=MatchValue(value=screen_type.name)),
FieldCondition(key="trap_action", match=MatchValue(value=action))
]
),
limit=1
)[0]
if results:
self._learned_traps.add(trap_key)
return True
except Exception:
pass
return False
class GoalPlanner:
"""
@@ -659,17 +717,17 @@ class GoalPlanner:
logger.info(f"🎯 [GOAP] Goal '{goal}' already achieved on {screen_type.value}!")
return None
# ── 2. Am I on the right screen? If not, navigate there ──
# ── 2. Execute the goal action ──
goal_action = self._plan_goal_action(goal_lower, screen_type, available, context)
if goal_action:
return goal_action
# ── 3. Am I on the right screen? If not, navigate there ──
selected_tab = screen.get('selected_tab')
nav_action = self._plan_navigation(goal_lower, screen_type, available, selected_tab, explored_nav_actions)
if nav_action:
return nav_action
# ── 3. Execute the goal action ──
goal_action = self._plan_goal_action(goal_lower, screen_type, available, context)
if goal_action:
return goal_action
return 'press back'
def _is_goal_achieved(self, goal: str, screen_type: ScreenType, context: dict) -> bool:
@@ -699,17 +757,39 @@ class GoalPlanner:
def _plan_navigation(self, goal: str, screen_type: ScreenType, available: List[str], selected_tab: Optional[str] = None, explored_nav_actions: set = None) -> Optional[str]:
"""If we're on the wrong screen, figure out how to navigate."""
# 0. Aversive Filter: Remove known traps from available actions
safe_available = []
for action in available:
if not self.knowledge.is_trap(screen_type, action):
safe_available.append(action)
else:
logger.debug(f"🛡️ [Aversive Filter] Masking trapped action: '{action}'")
available = safe_available
# 1. Get required screens for this goal from knowledge
required_screens = self.knowledge.get_requirements(goal)
# 2. Blank Start Discovery (if knowledge is empty)
if not required_screens:
logger.info(f"🧠 [Nav Discovery] No known requirements for '{goal}'. Will attempt autonomous discovery.")
# Semantic Heuristic Match on goal text vs available actions
verbs = {'open', 'tap', 'click', 'navigate', 'press', 'goto', 'view', 'feed'}
goal_words = [w.rstrip('s') for w in goal.replace('_', ' ').lower().split() if len(w) > 3 and w not in verbs]
for action in available:
if any(w in action.lower() for w in goal_words):
# Verify we don't know this leads somewhere else explicitly
known_target = self.knowledge.get_screen_for_action(action)
if not known_target:
logger.info(f"🎯 [Nav Discovery] Linguistic match on available action! '{action}' aligns with '{goal}'")
return action
# We don't know the required screen or path. Let the TelepathicEngine figure out
# what button to press based on the pure goal text!
if explored_nav_actions and goal in explored_nav_actions:
logger.info(f"🛑 [Nav Discovery] Autonomous intent '{goal}' already tried and failed/trapped. Yielding to back-tracking.")
pass # Don't return goal again, let it press back if possible
return None # Don't return goal again — force fallback to press back
else:
return goal
@@ -869,6 +949,7 @@ class GoalExecutor:
# ── Live planning ──
steps_taken = []
last_action = None
last_screen_type = None
explored_nav_actions = set()
for step_num in range(max_steps):
# PERCEIVE
@@ -899,9 +980,10 @@ class GoalExecutor:
# SAE Feedback Loop!
# If we hit this, the LAST action caused an obstacle! Mask it!
if last_action:
if last_action and last_screen_type:
self.action_failures[last_action] = self.action_failures.get(last_action, 0) + MAX_RETRIES # Instantly mask it
logger.warning(f"🛡️ [SAE Feedback] Action '{last_action}' caused an obstacle. Masking aggressively.")
self.planner.knowledge.learn_trap(last_screen_type, last_action, f"caused_obstacle_{obstacle_name}")
logger.warning(f"🛡️ [SAE Feedback] Action '{last_action}' caused an obstacle. Masking aggressively and learned trap.")
if not self._get_sae().ensure_clear_screen():
if screen_type == ScreenType.FOREIGN_APP:
@@ -923,6 +1005,7 @@ class GoalExecutor:
logger.info(f"🧭 [GOAP Step {step_num + 1}] Action: '{action}'")
last_action = action
last_screen_type = screen_type
# EXECUTE
success = self._execute_action(action, goal=goal)
@@ -936,7 +1019,17 @@ class GoalExecutor:
self.action_failures[action] = 0
else:
self.action_failures[action] = self.action_failures.get(action, 0) + 1
logger.warning(f"⚠️ [GOAP] Action '{action}' failed. Continuing with replanning...")
# Track failed actions in explored_nav_actions so the planner
# knows NOT to return the same synthetic intent again.
# Without this, synthetic intents (not in available_actions)
# bypass the masking logic and loop forever.
explored_nav_actions.add(action)
if self.action_failures[action] >= MAX_RETRIES:
self.planner.knowledge.learn_trap(screen_type, action, "repeated_failure_or_null_action")
logger.error(f"💀 [GOAP Execute] Action '{action}' failed {MAX_RETRIES} times. Marked as permanent trap.")
else:
logger.warning(f"⚠️ [GOAP Execute] Action '{action}' failed. Continuing with replanning...")
random_sleep(0.5, 1.5)
@@ -976,7 +1069,7 @@ class GoalExecutor:
xml_dump = self.device.dump_hierarchy()
best_node = engine.find_best_node(
xml_dump, action, min_confidence=0.75, device=self.device
xml_dump, action, min_confidence=0.75, device=self.device, goal=goal
)
if not best_node or best_node.get("skip"):
@@ -1007,10 +1100,37 @@ class GoalExecutor:
if is_navigation:
if ui_changed:
action_success = True
logger.info(f"✅ [GOAP Step] Navigation '{action}' successful -> {post_screen_type.name}.")
# Record dynamic knowledge: This action leads to THIS screen
self.planner.knowledge.learn_screen_mapping(action, post_screen_type)
# ── Goal-Aware Navigation Validation ──
# If we have a goal, verify we landed on the correct screen.
# E.g., "tap messages tab" + goal="open messages" must land on DM_INBOX, not REELS_FEED.
if goal:
goal_screen_map = {
'messages': ScreenType.DM_INBOX,
'explore': ScreenType.EXPLORE_GRID,
'home': ScreenType.HOME_FEED,
'profile': ScreenType.OWN_PROFILE,
'reels': ScreenType.REELS_FEED,
}
expected_screen = None
for keyword, screen in goal_screen_map.items():
if keyword in goal.lower():
expected_screen = screen
break
if expected_screen and post_screen_type != expected_screen:
logger.warning(
f"❌ [GOAP Navigation] Wanted {expected_screen.name} but landed on "
f"{post_screen_type.name}. Rejecting navigation '{action}'."
)
action_success = False
else:
action_success = True
logger.info(f"✅ [GOAP Step] Navigation '{action}' successful -> {post_screen_type.name}.")
self.planner.knowledge.learn_screen_mapping(action, post_screen_type)
else:
action_success = True
logger.info(f"✅ [GOAP Step] Navigation '{action}' successful -> {post_screen_type.name}.")
self.planner.knowledge.learn_screen_mapping(action, post_screen_type)
else:
logger.warning(f"❌ [GOAP Step] No UI change detected after '{action}'.")
action_success = False

View File

@@ -127,6 +127,45 @@ def prewarm_ollama_models(configs):
import threading
threading.Thread(target=_warmup, daemon=True).start()
def unload_ollama_models(configs):
"""
Sends keep_alive: 0 to all configured local Ollama API endpoints via a background thread
to force the models to unload from VRAM during bot shutdown.
"""
args = configs.args
def _unload():
import threading
models_to_unload = set()
# Collect unique local models
for attr, url_attr in [
("ai_telepathic_model", "ai_telepathic_url"),
("ai_fallback_model", "ai_fallback_url"),
("ai_condenser_model", "ai_condenser_url"),
("ai_model", "ai_model_url")
]:
url = getattr(args, url_attr, "")
model = getattr(args, attr, "")
if model and url and ("localhost" in url or "127.0.0.1" in url):
models_to_unload.add((url, model))
for url, model in models_to_unload:
logger.info(f"❄️ [VRAM Cleanup] Instructing local Ollama engine to unload {model} from memory...")
try:
# Fire keep_alive: 0 to unload it from VRAM
requests.post(
url,
json={"model": model, "keep_alive": 0},
timeout=5
)
except Exception as e:
logger.debug(f"Failed to unload {model}: {e}")
if hasattr(args, "ai_telepathic_model"):
import threading
threading.Thread(target=_unload, daemon=True).start()
def log_openrouter_burn():
"""Fetches and logs the current OpenRouter API key usage (money burned) ONLY if OpenRouter is actively used."""
key = os.environ.get("OPENROUTER_API_KEY")

View File

@@ -0,0 +1,16 @@
"""Perception — Feed and Content Analysis."""
from GramAddict.core.perception.feed_analysis import (
FEED_MARKERS,
CAROUSEL_INDICATORS,
has_carousel_in_view,
extract_post_content,
has_feed_markers,
)
__all__ = [
"FEED_MARKERS",
"CAROUSEL_INDICATORS",
"has_carousel_in_view",
"extract_post_content",
"has_feed_markers",
]

View File

@@ -0,0 +1,90 @@
"""
Perception — Feed Content Analysis.
Structural analysis of the feed: detecting markers, carousels,
extracting post content. Zero-AI, pure structural parsing.
Extracted from bot_flow.py to enable isolated testing.
"""
import logging
import re
import xml.etree.ElementTree as ET
logger = logging.getLogger(__name__)
# ── Feed Presence Markers ──
# Resource IDs that confirm we're looking at a loaded feed post.
# Used by _wait_for_post_loaded and other wait loops.
FEED_MARKERS = [
"row_feed_photo_profile_name",
"row_feed_profile_header",
"row_feed_photo_imageview",
"clips_media_component",
"clips_video_container",
"clips_viewer_container",
"clips_linear_layout_container",
"zoomable_view_container",
"feed_action_row",
"carousel_viewpager"
]
# ── Carousel Detection ──
CAROUSEL_INDICATORS = [
"com.instagram.android:id/carousel_page_indicator",
"com.instagram.android:id/carousel_media_group",
"com.instagram.android:id/carousel_viewpager"
]
def has_carousel_in_view(xml_dump: str) -> bool:
"""
Checks if a carousel is present on screen based on standard Android UI identifiers.
Handles 'carousel_page_indicator', 'carousel_media_group', and 'carousel_viewpager'.
"""
return any(ind in xml_dump for ind in CAROUSEL_INDICATORS)
def extract_post_content(context_xml: str) -> dict:
"""
Extracts meaningful content data from the current feed post's XML.
This is the BOT'S EYES — what it actually "sees" about each post.
Returns:
{'username': str, 'description': str, 'caption': str}
"""
result = {"username": "", "description": "", "caption": ""}
try:
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
# 1. Learn/extract post author dynamically
author_node = telepath.find_best_node(context_xml, "post author username header", min_confidence=0.75)
if author_node and author_node.get("original_attribs", {}).get("text"):
result["username"] = author_node["original_attribs"]["text"].strip()
# 2. Learn/extract post media description dynamically
media_node = telepath.find_best_node(context_xml, "post media content", min_confidence=0.35)
if media_node and media_node.get("original_attribs", {}).get("desc"):
result["description"] = media_node["original_attribs"]["desc"].strip()
# 3. Visible caption text (heuristic fallback if node isn't explicitly found)
# Search all nodes for text that contains the username to find the caption body
root = ET.fromstring(context_xml)
for node in root.iter("node"):
text = node.attrib.get("text", "").strip()
if result["username"] and len(text) > 20 and result["username"] in text:
result["caption"] = text
break
except Exception as e:
logger.warning(f"Error extracting post content autonomously: {e}")
return result
def has_feed_markers(xml_dump: str) -> bool:
"""Quick check: does this XML contain any feed presence markers?"""
return any(marker in xml_dump for marker in FEED_MARKERS)

View File

@@ -0,0 +1,30 @@
"""Physics — Humanized Input Simulation, Biomechanics & UI Timing."""
from GramAddict.core.physics.humanized_input import (
humanized_scroll,
humanized_click,
humanized_horizontal_swipe,
)
from GramAddict.core.physics.timing import (
wait_for_post_loaded,
wait_for_story_loaded,
align_active_post,
)
from GramAddict.core.physics.biomechanics import (
PhysicsBody,
BezierGesture,
)
from GramAddict.core.physics.sendevent_injector import SendEventInjector
__all__ = [
"humanized_scroll",
"humanized_click",
"humanized_horizontal_swipe",
"wait_for_post_loaded",
"wait_for_story_loaded",
"align_active_post",
"PhysicsBody",
"BezierGesture",
"SendEventInjector",
]

View File

@@ -0,0 +1,440 @@
"""
Biomechanics — Organic Thumb Kinematics & Bézier Gesture Synthesis.
Simulates the physical behavior of a human thumb across an entire bot session:
- Spatial drift (posture changes)
- Fatigue (slower, less accurate over time)
- Handedness bias (right-handers arc right)
- Non-linear Bézier touch paths with sigmoid velocity and Gaussian pressure
This module produces gesture data (point sequences) that are then injected
via SendEventInjector or fall back to adb `input swipe`.
"""
import logging
import math
import random
import time
logger = logging.getLogger(__name__)
class PhysicsBody:
"""
Kinematic model of a human thumb over a session.
Tracks anchor position (where the thumb naturally rests), session-level
spatial drift (simulating posture changes), and fatigue (affecting speed
and accuracy). Provides biomechanically plausible start/end positions
for all gestures.
"""
_session_instance = None
def __init__(self, handedness="right", device_info=None):
self.handedness = handedness
# Defensive parsing — device_info may contain MagicMock objects in tests
try:
self.w = int(device_info.get("displayWidth", 1080)) if device_info else 1080
except (TypeError, ValueError):
self.w = 1080
try:
self.h = int(device_info.get("displayHeight", 2400)) if device_info else 2400
except (TypeError, ValueError):
self.h = 2400
# Anchor Point: natural thumb rest position
# Right-handers: lower-right quadrant; Left-handers: lower-left
self.anchor_x = self.w * (0.75 if handedness == "right" else 0.25)
self.anchor_y = self.h * 0.82
# Session Drift: simulates posture shifts over time
self.drift_x = 0.0
self.drift_y = 0.0
self.gesture_count = 0
# Fatigue Model: 0.0 = fresh, 1.0 = exhausted
self.fatigue = 0.0
self.last_gesture_time = time.time()
logger.debug(
f"🦴 [PhysicsBody] Initialized: {handedness}-handed, "
f"anchor=({self.anchor_x:.0f}, {self.anchor_y:.0f}), "
f"display={self.w}x{self.h}"
)
@classmethod
def get_session_instance(cls, device=None, handedness="right"):
"""
Returns a session-persistent PhysicsBody.
The body persists across all gestures within a single bot session,
accumulating drift and fatigue realistically.
"""
if cls._session_instance is None:
device_info = {}
if device:
try:
raw = device.get_info()
# Defensive: convert to plain dict to handle MagicMock returns
if isinstance(raw, dict):
device_info = raw
else:
device_info = {}
except Exception:
pass
cls._session_instance = cls(handedness=handedness, device_info=device_info)
return cls._session_instance
@classmethod
def reset(cls):
"""Reset for testing / new session."""
cls._session_instance = None
def get_scroll_start(self):
"""
Returns a biomechanically plausible scroll start position.
Right-handers start scrolls on the right side of the screen,
with Gaussian jitter and session drift applied.
"""
self._apply_session_drift()
self._update_fatigue()
# Base position: right side for right-handers, avoiding edges
base_x = self.anchor_x + self.drift_x
# Scroll starts in the lower 70-85% of the screen
base_y = self.h * random.uniform(0.70, 0.85) + self.drift_y
# Gaussian jitter (natural inaccuracy, increases with fatigue)
fatigue_mult = 1.0 + self.fatigue * 0.5
jitter_x = random.gauss(0, self.w * 0.02 * fatigue_mult)
jitter_y = random.gauss(0, self.h * 0.015 * fatigue_mult)
x = int(max(50, min(self.w - 50, base_x + jitter_x)))
y = int(max(200, min(self.h - 200, base_y + jitter_y)))
self.gesture_count += 1
return x, y
def get_tap_position(self, target_x, target_y):
"""
Returns a biomechanically plausible tap position near the target.
Applies thumb bias (right-handers land slightly left-down of center)
and Gaussian jitter.
"""
self._update_fatigue()
# Thumb bias: right-handers hit slightly left and below center
bias_x = -3 if self.handedness == "right" else 3
bias_y = 4 # Thumb pad is below the actual contact center
fatigue_mult = 1.0 + self.fatigue * 0.3
jitter_x = random.gauss(bias_x, 5 * fatigue_mult)
jitter_y = random.gauss(bias_y, 5 * fatigue_mult)
x = int(max(5, min(self.w - 5, target_x + jitter_x)))
y = int(max(5, min(self.h - 5, target_y + jitter_y)))
self.gesture_count += 1
return x, y
def get_thumb_arc_bias(self):
"""
Returns the horizontal arc bias for scroll curves.
Right-handers naturally arc their thumb to the right during
vertical swipes; left-handers arc left.
"""
base_arc = self.w * 0.04
if self.handedness == "right":
return base_arc + random.uniform(-self.w * 0.01, self.w * 0.02)
else:
return -(base_arc + random.uniform(-self.w * 0.01, self.w * 0.02))
def get_pressure_baseline(self):
"""
Returns the baseline pressure for touch events.
Fatigued thumbs press harder (compensating for reduced precision).
"""
baseline = 0.35 + self.fatigue * 0.15
return min(0.85, baseline + random.uniform(-0.05, 0.05))
def get_touch_major(self):
"""
Returns the touch contact area (touch_major) in device units.
Fatigued thumbs have a larger contact patch (flatter press).
"""
base = 6 + int(self.fatigue * 4)
return max(4, base + random.randint(-2, 2))
def _apply_session_drift(self):
"""
Every ~15-25 gestures, apply a small posture shift.
Simulates the user adjusting their grip on the phone.
"""
drift_interval = random.randint(15, 25)
if self.gesture_count > 0 and self.gesture_count % drift_interval == 0:
old_dx, old_dy = self.drift_x, self.drift_y
self.drift_x += random.gauss(0, self.w * 0.025)
self.drift_y += random.gauss(0, self.h * 0.015)
# Clamp drift so we don't wander off the screen
self.drift_x = max(-self.w * 0.1, min(self.w * 0.1, self.drift_x))
self.drift_y = max(-self.h * 0.06, min(self.h * 0.06, self.drift_y))
if abs(self.drift_x - old_dx) > 5 or abs(self.drift_y - old_dy) > 5:
logger.debug(
f"🦴 [PhysicsBody] Posture drift: "
f"Δx={self.drift_x - old_dx:+.0f}, Δy={self.drift_y - old_dy:+.0f} "
f"(gesture #{self.gesture_count})"
)
def _update_fatigue(self):
"""
Update fatigue based on gesture frequency.
Rapid gestures increase fatigue; idle periods recover it.
"""
elapsed = time.time() - self.last_gesture_time
if elapsed < 0.5:
# Rapid-fire: fatigue increases
self.fatigue = min(1.0, self.fatigue + 0.015)
elif elapsed > 8.0:
# Long pause: recovery
self.fatigue = max(0.0, self.fatigue - 0.08)
elif elapsed > 3.0:
# Moderate pause: slight recovery
self.fatigue = max(0.0, self.fatigue - 0.02)
self.last_gesture_time = time.time()
class BezierGesture:
"""
Generates multi-point cubic Bézier curves for organic touch paths.
Replaces the linear A→B interpolation of `adb shell input swipe`
with biomechanically accurate gesture trajectories including:
- Thumb arc curvature (handedness-dependent)
- Sigmoid velocity profile (slow start → fast middle → slow end)
- Gaussian pressure curve (light touch → firm contact → light lift)
"""
@staticmethod
def scroll_curve(start, end, body: PhysicsBody, n_points=None):
"""
Generates a vertical scroll gesture curve.
Args:
start: (x, y) start position
end: (x, y) end position
body: PhysicsBody for handedness/fatigue context
n_points: Override for number of intermediate points
Returns:
List of (x, y, pressure) tuples along the Bézier curve
"""
sx, sy = start
ex, ey = end
if n_points is None:
n_points = random.randint(10, 18)
# Thumb arc: the control points bias the curve sideways
arc_bias = body.get_thumb_arc_bias()
# Two control points for cubic Bézier
# CP1: early in the gesture, slight arc
cp1_x = sx + arc_bias * random.uniform(0.2, 0.4)
cp1_y = sy + (ey - sy) * random.uniform(0.2, 0.35)
# CP2: later in the gesture, peak arc
cp2_x = sx + arc_bias * random.uniform(0.5, 0.8)
cp2_y = sy + (ey - sy) * random.uniform(0.65, 0.8)
pressure_baseline = body.get_pressure_baseline()
points = []
for i in range(n_points + 1):
t = i / n_points
# Cubic Bézier interpolation
x = (
(1 - t) ** 3 * sx
+ 3 * (1 - t) ** 2 * t * cp1_x
+ 3 * (1 - t) * t ** 2 * cp2_x
+ t ** 3 * ex
)
y = (
(1 - t) ** 3 * sy
+ 3 * (1 - t) ** 2 * t * cp1_y
+ 3 * (1 - t) * t ** 2 * cp2_y
+ t ** 3 * ey
)
# Micro-noise on each point (finger vibration)
x += random.gauss(0, 1.5)
y += random.gauss(0, 1.5)
# Pressure curve: Gaussian peak around t=0.4 (peak contact mid-gesture)
pressure = pressure_baseline + 0.3 * math.exp(
-((t - 0.4) ** 2) / 0.1
)
pressure += random.uniform(-0.04, 0.04)
pressure = max(0.08, min(0.92, pressure))
points.append((int(x), int(y), round(pressure, 3)))
return points
@staticmethod
def tap_curve(target_x, target_y, body: PhysicsBody):
"""
Generates a tap gesture (touch-down → micro-drift → touch-up).
Returns:
List of (x, y, pressure) tuples (typically 3-5 points)
"""
tx, ty = body.get_tap_position(target_x, target_y)
pressure_base = body.get_pressure_baseline()
# Touch-down (initial light contact)
p_down = max(0.1, pressure_base * 0.6 + random.uniform(-0.05, 0.05))
# Full contact
p_full = min(0.9, pressure_base + random.uniform(-0.05, 0.1))
# Release
p_up = max(0.05, pressure_base * 0.3 + random.uniform(-0.03, 0.03))
# Micro-drift: finger slides ~2-6px during contact
drift_x = random.randint(-4, 4)
drift_y = random.randint(-4, 4)
points = [
(tx, ty, round(p_down, 3)),
(tx + drift_x // 2, ty + drift_y // 2, round(p_full, 3)),
(tx + drift_x, ty + drift_y, round(p_up, 3)),
]
return points
@staticmethod
def horizontal_swipe_curve(start, end, body: PhysicsBody, n_points=None):
"""
Generates a horizontal swipe curve (e.g., carousel browsing).
Includes vertical arc (thumb drops downward when swiping left-to-right
for right-handers) and sigmoid velocity.
"""
sx, sy = start
ex, ey = end
if n_points is None:
n_points = random.randint(8, 14)
# Vertical arc for horizontal swipes
# Right-handers swiping left: thumb drops 30-90px
direction = 1 if ex < sx else -1 # 1 = swiping left
if body.handedness == "right":
y_arc = direction * random.uniform(25, 70)
else:
y_arc = -direction * random.uniform(25, 70)
# Control points
cp1_x = sx + (ex - sx) * random.uniform(0.25, 0.35)
cp1_y = sy + y_arc * 0.4
cp2_x = sx + (ex - sx) * random.uniform(0.65, 0.75)
cp2_y = sy + y_arc * 0.9
pressure_baseline = body.get_pressure_baseline()
points = []
for i in range(n_points + 1):
t = i / n_points
x = (
(1 - t) ** 3 * sx
+ 3 * (1 - t) ** 2 * t * cp1_x
+ 3 * (1 - t) * t ** 2 * cp2_x
+ t ** 3 * ex
)
y = (
(1 - t) ** 3 * sy
+ 3 * (1 - t) ** 2 * t * cp1_y
+ 3 * (1 - t) * t ** 2 * cp2_y
+ t ** 3 * ey
)
x += random.gauss(0, 2)
y += random.gauss(0, 2)
pressure = pressure_baseline + 0.25 * math.exp(
-((t - 0.45) ** 2) / 0.12
)
pressure += random.uniform(-0.04, 0.04)
pressure = max(0.08, min(0.92, pressure))
points.append((int(x), int(y), round(pressure, 3)))
return points
@staticmethod
def compute_sigmoid_timing(n_points, total_duration_ms):
"""
Generates a sigmoid-based timing schedule for gesture points.
Produces intervals that are longer at the start and end
(slow acceleration/deceleration) and shorter in the middle
(peak velocity). This matches real human swipe kinematics.
Returns:
List of inter-point delay times in seconds (length n_points)
"""
if n_points <= 1:
return [total_duration_ms / 1000.0]
# Generate sigmoid-spaced t values
raw_intervals = []
for i in range(n_points):
# Normalized position
t = i / (n_points - 1) if n_points > 1 else 0.5
# Inverted sigmoid: fast in middle, slow at edges
# Higher value = longer delay = slower movement
sigmoid = 1.0 / (1.0 + math.exp(-8 * (t - 0.5)))
# U-shaped: slow at start & end, fast in middle
speed_factor = 0.4 + 1.2 * (4 * (t - 0.5) ** 2)
raw_intervals.append(speed_factor)
# Normalize to total duration
total_raw = sum(raw_intervals)
total_sec = total_duration_ms / 1000.0
intervals = [(r / total_raw) * total_sec for r in raw_intervals]
# Add micro-jitter to timing (humans are never perfectly rhythmic)
intervals = [
max(0.002, i + random.uniform(-0.003, 0.003)) for i in intervals
]
return intervals
@staticmethod
def compute_fling_timing(n_points, total_duration_ms):
"""
Generates a J-curve timing schedule for flick/swipe gestures.
Unlike the sigmoid (which slows down at the end), this curve
accelerates through the middle and maintains high velocity
until the very last point to simulate a sudden 'liftoff' flick.
This allows Android's ScrollView to register a high fling velocity.
Returns:
List of inter-point delay times in seconds (length n_points)
"""
if n_points <= 1:
return [total_duration_ms / 1000.0]
raw_intervals = []
for i in range(n_points):
t = i / (n_points - 1)
# Starts slow (larger delay), speeds up continuously (smaller delay)
speed_factor = 1.0 - (0.8 * t)
raw_intervals.append(speed_factor)
total_raw = sum(raw_intervals)
total_sec = total_duration_ms / 1000.0
intervals = [(r / total_raw) * total_sec for r in raw_intervals]
# Add micro-jitter to timing
intervals = [max(0.002, i + random.uniform(-0.003, 0.003)) for i in intervals]
return intervals

View File

@@ -0,0 +1,232 @@
"""
Physics — Humanized Input Simulation.
All low-level device interaction functions that simulate human touch behavior:
scroll, click, swipe, horizontal swipe.
Uses Biomechanical Bézier curve generation for organic, non-linear touch paths
with sigmoid velocity profiles and Gaussian pressure variation.
Falls back to linear `input swipe` when sendevent is unavailable.
Extracted from bot_flow.py to enable isolated testing and reuse.
"""
import logging
import random
from time import sleep
from GramAddict.core.physics.biomechanics import PhysicsBody, BezierGesture
from GramAddict.core.physics.sendevent_injector import SendEventInjector
logger = logging.getLogger(__name__)
def humanized_scroll(device, is_skip=False, resonance_score=None):
"""
Simulates a human thumb flick to trigger native scroll-snapping.
Uses Bézier curves for non-linear path generation and sigmoid timing
for organic acceleration/deceleration. The PhysicsBody provides
session-persistent anchor drift and fatigue modeling.
resonance_score: Optional. If high, increases chance of 'Correction' (Reverse scroll).
"""
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
body = PhysicsBody.get_session_instance(device)
injector = SendEventInjector.get_instance(device)
# 1. Calculate Base Probability for Correction (Reverse Flick)
# Default 15% for doomscroll corrections.
# If resonance is high, we scale up to 45% chance to "Look back" at what we just passed.
correction_prob = 0.15
if resonance_score is not None and resonance_score > 0.7:
correction_prob = 0.15 + (resonance_score - 0.7) * 1.0 # 0.7=0.15, 1.0=0.45
# Start position from PhysicsBody (session-aware, drifting)
start_x, start_y = body.get_scroll_start()
end_x = start_x + random.gauss(0, w * 0.008) # Slight horizontal drift
do_correction = random.random() < correction_prob
if is_skip:
# Aggressive fast fling to skip quickly
if do_correction:
logger.debug(f"🪀 [Doomscroll] Correction (Prob: {correction_prob:.2f}) — Wait, what was that?")
distance = int(h * random.uniform(0.3, 0.5))
duration = random.uniform(100, 150)
end_y = min(start_y + distance, h - 10) # Move down to pull UI up
else:
distance = int(h * random.uniform(0.6, 0.75))
duration = random.uniform(100, 150)
end_y = start_y - distance
else:
# Playful, organic human scrolling
play_choice = random.random()
if play_choice > (1.0 - (correction_prob / 3.0)) or play_choice > 0.95:
# "Go back" / Scroll UP
start_y = int(h * random.uniform(0.20, 0.40))
distance = int(h * random.uniform(0.30, 0.50))
duration = random.uniform(100, 180)
end_y = min(start_y + distance, h - 10)
logger.info(f"🪀 [Playful Scroll] Correction (Prob: {correction_prob:.2f}) — Flicking back up...")
elif play_choice > 0.85:
# "Reading Jitter" / Playing around (10% chance)
distance = int(h * random.uniform(0.05, 0.15))
duration = random.uniform(300, 600)
if random.random() > 0.5:
end_y = start_y - distance
else:
start_y = int(h * random.uniform(0.30, 0.50))
end_y = start_y + distance
logger.info("🪀 [Playful Scroll] Micro-jitter...")
elif play_choice > 0.25:
# "Lazy Flick" - Post to Post Snap (60% chance)
distance = int(h * random.uniform(0.15, 0.25))
duration = random.uniform(150, 350)
end_y = start_y - distance
else:
# Medium classic swipe (25% chance)
distance = int(h * random.uniform(0.30, 0.45))
duration = random.uniform(250, 500)
end_y = start_y - distance
# --- Behavioral Micro-Patterns (new human behaviors) ---
behavior = _select_scroll_behavior()
if behavior == "pre_touch_dwell":
# Finger lands on glass before swiping (50-200ms dwell)
logger.debug("🦴 [Biomechanics] Pre-touch dwell...")
if behavior == "overshoot_correction":
# Scroll too far, then micro-correct back
logger.debug("🦴 [Biomechanics] Overshoot + Correction pattern")
# Extend original distance, then we'll add a correction swipe after
original_end_y = end_y
overshoot = int(h * random.uniform(0.08, 0.15))
if end_y < start_y:
end_y -= overshoot # Scroll further down
else:
end_y += overshoot # Scroll further up
if behavior == "reading_pause":
logger.debug("🦴 [Biomechanics] Mid-scroll reading pause")
# --- Generate Bézier Curve ---
points = BezierGesture.scroll_curve(
(start_x, start_y), (int(end_x), end_y), body
)
timing = BezierGesture.compute_sigmoid_timing(len(points), duration)
# Pre-touch dwell: hold finger on glass before moving
if behavior == "pre_touch_dwell":
pre_dwell_ms = random.uniform(0.05, 0.2)
# Insert a stationary point at the beginning
points.insert(0, points[0])
timing.insert(0, pre_dwell_ms)
# Reading pause: insert a long dwell mid-gesture
if behavior == "reading_pause":
mid = len(points) // 2
pause_point = points[mid]
pause_duration = random.uniform(0.5, 2.0)
points.insert(mid + 1, pause_point)
timing.insert(mid, pause_duration)
# --- Inject Gesture ---
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
# Post-gesture: overshoot correction
if behavior == "overshoot_correction":
sleep(random.uniform(0.3, 0.6))
# Small corrective scroll back
corr_start_x, corr_start_y = body.get_scroll_start()
corr_distance = int(h * random.uniform(0.05, 0.1))
if original_end_y < start_y:
corr_end_y = corr_start_y + corr_distance # Scroll back up
else:
corr_end_y = corr_start_y - corr_distance # Scroll back down
corr_points = BezierGesture.scroll_curve(
(corr_start_x, corr_start_y), (corr_start_x, corr_end_y), body,
n_points=6
)
corr_timing = BezierGesture.compute_sigmoid_timing(len(corr_points), 200)
injector.inject_gesture(corr_points, corr_timing, touch_major=body.get_touch_major())
def humanized_click(device, x, y, double=False, sleep_mod=1.0):
"""Simulates a human tap with biomechanical jitter and micro-drift."""
body = PhysicsBody.get_session_instance(device)
injector = SendEventInjector.get_instance(device)
def single_tap():
points = BezierGesture.tap_curve(x, y, body)
# Tap timing: 40-90ms contact time
tap_duration = random.uniform(40, 90)
timing = BezierGesture.compute_sigmoid_timing(len(points), tap_duration)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
if double:
# For double tap, the timing is extremely critical (<300ms between taps).
# We bypass sendevent overhead and batch two input taps directly in the shell.
device.shell(f"input tap {int(x)} {int(y)} && input tap {int(x)} {int(y)}")
else:
single_tap()
def humanized_horizontal_swipe(device, start_x, end_x, y, duration_ms):
"""Simulates a human horizontal swipe with thumb arc simulation."""
body = PhysicsBody.get_session_instance(device)
injector = SendEventInjector.get_instance(device)
# Apply jitter to start/end positions
noise_y = random.randint(-15, 15)
actual_start_x = int(start_x) + random.randint(-10, 10)
actual_end_x = int(end_x) + random.randint(-20, 20)
actual_y = int(y) + noise_y
# Timing wobble (+/- 30%)
actual_duration = int(duration_ms * random.uniform(0.7, 1.3))
points = BezierGesture.horizontal_swipe_curve(
(actual_start_x, actual_y), (actual_end_x, actual_y), body
)
timing = BezierGesture.compute_sigmoid_timing(len(points), actual_duration)
direction = "" if end_x > start_x else ""
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
def _select_scroll_behavior():
"""
Selects a micro-behavior pattern for the current scroll gesture.
Returns one of:
- None: standard scroll (most common)
- "pre_touch_dwell": finger lands on glass before swiping
- "overshoot_correction": scrolls too far, then corrects back
- "reading_pause": finger pauses mid-scroll
"""
roll = random.random()
if roll < 0.08:
return "pre_touch_dwell"
elif roll < 0.20:
return "overshoot_correction"
elif roll < 0.35:
return "reading_pause"
return None

View File

@@ -0,0 +1,267 @@
"""
SendEvent Injector — Kernel-Level Touch Event Injection via ADB.
Injects raw MotionEvent sequences through `adb shell sendevent` to produce
touch events that are indistinguishable from real finger input at the kernel level.
Key advantages over `input swipe`:
- Supports pressure variation (ABS_MT_PRESSURE)
- Supports touch contact area (ABS_MT_TOUCH_MAJOR)
- Produces SOURCE_TOUCHSCREEN events (not SOURCE_UNKNOWN)
- Multi-point non-linear paths
Falls back to `input swipe` if sendevent device detection fails.
Note: sendevent codes are device-specific. This module auto-detects the
correct /dev/input/eventX and the axis ranges on first use.
"""
import logging
import re
import time
from time import sleep
logger = logging.getLogger(__name__)
class SendEventInjector:
"""
Injects touch events via adb shell sendevent for organic gesture simulation.
Uses a batched shell command approach: all events for one gesture are piped
into a single `adb shell` invocation to minimize latency.
"""
_instance = None
# Standard Linux input event types/codes
EV_ABS = 3
EV_SYN = 0
EV_KEY = 1
# Multitouch protocol B codes (most modern Android devices)
ABS_MT_TRACKING_ID = 0x39 # 57
ABS_MT_POSITION_X = 0x35 # 53
ABS_MT_POSITION_Y = 0x36 # 54
ABS_MT_PRESSURE = 0x3A # 58
ABS_MT_TOUCH_MAJOR = 0x30 # 48
SYN_REPORT = 0
BTN_TOUCH = 0x14A # 330
def __init__(self, device):
self.device = device
self.event_device = None
self.x_max = 1080
self.y_max = 2400
self.pressure_max = 255
self.touch_major_max = 30
self._fallback_mode = False
self._detected = False
@classmethod
def get_instance(cls, device):
"""Returns a singleton injector for the device."""
if cls._instance is None:
cls._instance = cls(device)
cls._instance._detect_touch_device()
return cls._instance
@classmethod
def reset(cls):
"""Reset for testing / device change."""
cls._instance = None
def _detect_touch_device(self):
"""
Auto-detects the touchscreen input device and its axis ranges
by parsing `getevent -pl` output.
"""
try:
result = self.device.shell("getevent -pl")
if not isinstance(result, str):
result = str(result)
# Find device with ABS_MT_POSITION_X
current_device = None
for line in result.split("\n"):
line = line.strip()
# Device header: /dev/input/eventX
dev_match = re.match(r'add device \d+:\s*(/dev/input/event\d+)', line)
if dev_match:
current_device = dev_match.group(1)
# Check for multitouch capability
if current_device and "ABS_MT_POSITION_X" in line:
self.event_device = current_device
logger.info(
f"🖐️ [SendEvent] Touch device detected: {self.event_device}"
)
# Parse axis ranges from the same section
self._parse_axis_ranges(result, current_device)
self._detected = True
return
# If no MT device found, try fallback pattern
logger.debug(
"⚠️ [SendEvent] No multitouch device found. "
"Falling back to `input swipe` mode."
)
self._fallback_mode = True
except Exception as e:
logger.warning(
f"⚠️ [SendEvent] Device detection failed: {e}. "
f"Falling back to `input swipe` mode."
)
self._fallback_mode = True
def _parse_axis_ranges(self, getevent_output, device_path):
"""
Parses axis max values from getevent output.
Lines look like: ABS_MT_POSITION_X : value 0, min 0, max 1079, ...
"""
try:
in_device = False
for line in getevent_output.split("\n"):
if device_path in line:
in_device = True
continue
if in_device and line.strip().startswith("add device"):
break # Next device
if in_device:
if "ABS_MT_POSITION_X" in line:
m = re.search(r'max\s+(\d+)', line)
if m:
self.x_max = int(m.group(1))
elif "ABS_MT_POSITION_Y" in line:
m = re.search(r'max\s+(\d+)', line)
if m:
self.y_max = int(m.group(1))
elif "ABS_MT_PRESSURE" in line:
m = re.search(r'max\s+(\d+)', line)
if m:
self.pressure_max = int(m.group(1))
elif "ABS_MT_TOUCH_MAJOR" in line:
m = re.search(r'max\s+(\d+)', line)
if m:
self.touch_major_max = int(m.group(1))
logger.debug(
f"🖐️ [SendEvent] Axis ranges: X=0-{self.x_max}, "
f"Y=0-{self.y_max}, P=0-{self.pressure_max}, "
f"TM=0-{self.touch_major_max}"
)
except Exception as e:
logger.debug(f"[SendEvent] Axis parsing error: {e}")
def inject_gesture(self, points, timing_intervals, touch_major=6):
"""
Injects a complete gesture (touch-down → move → touch-up) using sendevent.
Args:
points: List of (x, y, pressure) tuples from BezierGesture
timing_intervals: List of inter-point delays in seconds
touch_major: Contact area size
Falls back to `input swipe` if sendevent is unavailable.
"""
if self._fallback_mode or not self.event_device:
return self._fallback_input_swipe(points, timing_intervals)
if len(points) < 2:
return
try:
dev = self.event_device
# Scale coordinates from display space to input device space
try:
info = self.device.get_info()
display_w = int(info.get("displayWidth", 1080)) if isinstance(info, dict) else 1080
display_h = int(info.get("displayHeight", 2400)) if isinstance(info, dict) else 2400
except (TypeError, ValueError):
display_w, display_h = 1080, 2400
scale_x = self.x_max / display_w
scale_y = self.y_max / display_h
# --- Touch Down (first point) ---
x, y, pressure = points[0]
ix = int(x * scale_x)
iy = int(y * scale_y)
ip = int(pressure * self.pressure_max)
itm = min(touch_major, self.touch_major_max)
# Build batch command for touch-down
cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_TRACKING_ID} 0")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_PRESSURE} {ip}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_TOUCH_MAJOR} {itm}")
cmds.append(f"sendevent {dev} {self.EV_KEY} {self.BTN_TOUCH} 1")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
# Execute touch-down
self.device.shell(" && ".join(cmds))
# --- Move through intermediate points ---
for i in range(1, len(points) - 1):
if i - 1 < len(timing_intervals):
sleep(timing_intervals[i - 1])
x, y, pressure = points[i]
ix = int(x * scale_x)
iy = int(y * scale_y)
ip = int(pressure * self.pressure_max)
cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_PRESSURE} {ip}")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
self.device.shell(" && ".join(cmds))
# --- Touch Up (last point) ---
if len(timing_intervals) >= len(points) - 1:
sleep(timing_intervals[-1])
else:
sleep(0.01)
x, y, pressure = points[-1]
ix = int(x * scale_x)
iy = int(y * scale_y)
cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_PRESSURE} 0")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_TRACKING_ID} -1")
cmds.append(f"sendevent {dev} {self.EV_KEY} {self.BTN_TOUCH} 0")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
self.device.shell(" && ".join(cmds))
except Exception as e:
logger.warning(f"⚠️ [SendEvent] Injection failed: {e}. Falling back.")
self._fallback_input_swipe(points, timing_intervals)
def _fallback_input_swipe(self, points, timing_intervals):
"""
Fallback: Uses adb `input swipe` with first and last point.
Loses pressure and curvature but maintains timing.
"""
if len(points) < 2:
return
sx, sy, _ = points[0]
ex, ey, _ = points[-1]
total_ms = int(sum(timing_intervals) * 1000) if timing_intervals else 300
self.device.shell(
f"input swipe {int(sx)} {int(sy)} {int(ex)} {int(ey)} {total_ms}"
)

View File

@@ -0,0 +1,186 @@
"""
Physics — Timing & Wait Utilities.
UI readiness polling, post alignment, and adaptive snap recovery.
These functions wait for the Android UI to reach a known state before
the bot proceeds with interactions.
Extracted from bot_flow.py to enable isolated testing.
"""
import logging
import re
import time
from time import sleep
from GramAddict.core.perception.feed_analysis import FEED_MARKERS
from GramAddict.core.diagnostic_dump import dump_ui_state
logger = logging.getLogger(__name__)
def wait_for_post_loaded(device, timeout=5, nav_graph=None):
"""
Polls the UI hierarchy until feed markers appear, confirming a post is on screen.
If timeout is reached, attempts Adaptive Snap recovery:
1. Detects trap states (Story/Reel viewer, Profile)
2. Presses BACK to escape
3. Micro-wobbles to force render
"""
start = time.time()
xml = ""
while time.time() - start < timeout:
try:
xml = device.dump_hierarchy()
if any(marker in xml for marker in FEED_MARKERS):
logger.debug("📱 Post loaded successfully.")
return True
except Exception:
pass
sleep(0.5)
logger.warning("⚠️ Post did not load within timeout. Attempting Adaptive Snap.")
dump_ui_state(device, "post_load_timeout", {"timeout_sec": timeout})
try:
xml_lower = xml.lower()
# 1. Trapped in a Story or Reel viewer? Press back.
if "reel_viewer_root" in xml_lower or "clips_viewer" in xml_lower:
logger.warning("🧗 [Adaptive Snap] Trapped in Story/Reel viewer. Pressing BACK.")
device.press("back")
sleep(1.5)
# Give it one more chance to load the feed
xml = device.dump_hierarchy()
if any(marker in xml for marker in FEED_MARKERS):
logger.info("✅ Recovered to Feed.")
return True
# 2. Trapped in Profile?
if "profile_header" in xml_lower and "row_feed_photo_profile_name" not in xml_lower:
logger.warning("🧗 [Adaptive Snap] Trapped in Profile. Pressing BACK.")
device.press("back")
sleep(1.5)
# 3. Stuck between posts (Feed markers not fully visible)? Micro-wobble.
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
logger.warning("🧗 [Adaptive Snap] Wobbling to force render.")
device.swipe(int(w/2), int(h/2), int(w/2), int(h/2) - 100, 0.1)
sleep(0.5)
device.swipe(int(w/2), int(h/2) - 100, int(w/2), int(h/2), 0.1)
except Exception as e:
logger.error(f"❌ [Adaptive Snap] Failed: {e}")
return False
def wait_for_story_loaded(device, timeout=5):
"""Polls the UI hierarchy until story markers appear, confirming a story is on screen."""
start = time.time()
while time.time() - start < timeout:
try:
xml_lower = device.dump_hierarchy().lower()
if "reel_viewer_root" in xml_lower or "story_viewer" in xml_lower:
logger.debug("📱 Story loaded successfully.")
return True
except Exception:
pass
sleep(0.5)
logger.warning("⚠️ Story did not load within timeout.")
return False
def wait_for_profile_loaded(device, timeout=5):
"""Polls the UI hierarchy until profile markers appear."""
import time
start = time.time()
PROFILE_MARKERS = ["profile_header", "action_bar_title", "profile_tabs_container"]
while time.time() - start < timeout:
try:
xml_lower = device.dump_hierarchy().lower()
if any(marker in xml_lower for marker in PROFILE_MARKERS):
logger.debug("📱 Profile loaded successfully.")
return True
except Exception:
pass
sleep(0.5)
logger.warning("⚠️ Profile did not load within timeout.")
return False
def align_active_post(device):
"""
Programmatic snapping correction. Finds the nearest post header and perfectly
snaps it to the top margin. Fixes inverted scroll mapping that pushed content away.
Loops to ensure absolute alignment if stuck deeply between posts.
"""
aligned = False
attempts = 0
max_attempts = 3
while not aligned and attempts < max_attempts:
attempts += 1
try:
xml = device.dump_hierarchy()
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
target_node = telepath.find_best_node(
xml, "post author header profile",
min_confidence=0.4, device=device
)
if target_node:
original_attribs = target_node.get('original_attribs', {})
bounds = original_attribs.get('bounds', '')
if not bounds:
bounds = target_node.get('bounds', '')
m = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds)
if m:
l, t, r, b = map(int, m.groups())
header_y = (t + b) // 2
# Instagram's optimal top margin for a snapped post is ~200-280px
target_y = 250
diff = header_y - target_y
# If target is off-center (> 100px), execute precise correction swipe
if abs(diff) > 100:
info = device.get_info()
w, h = info.get("displayWidth", 1080), info.get("displayHeight", 2400)
cx = w // 2
max_safe_swipe = int(h * 0.4)
if diff > 0:
# Content is too LOW. Move it UP.
dist = min(diff, max_safe_swipe)
start_y = int(h * 0.7)
end_y = start_y - dist
else:
# Content is too HIGH. Move it DOWN.
dist = min(abs(diff), max_safe_swipe)
start_y = int(h * 0.3)
end_y = start_y + dist
# Duration 1.0s = precise mechanical drag with ZERO momentum
device.swipe(cx, start_y, cx, end_y, duration=1.0)
sleep(1.0)
logger.debug(f"📐 [Alignment] Snapping attempt {attempts}: Shifted {diff}px.")
else:
aligned = True
else:
break # No header found, cannot align
except Exception as e:
logger.debug(f"📐 [Alignment] Snapping correction failed: {e}")
break
if aligned and attempts > 1:
logger.debug(f"📐 [Alignment] Snapped post cleanly into view after {attempts} attempts.")
return True
return aligned

View File

@@ -314,7 +314,7 @@ class UIMemoryDB(QdrantBase):
"""
return hashlib.sha256(intent.encode("utf-8")).hexdigest()[:32]
def retrieve_memory(self, intent: str, xml_context: str, similarity_threshold: float = None) -> Optional[dict]:
def retrieve_memory(self, intent: str, xml_context: str, similarity_threshold: float = None, exact_only: bool = False) -> Optional[dict]:
"""
Queries Qdrant for a known resolution representing the given intent.
Returns the cached intent result (e.g. is_ad boolean, or bounding box) if found.
@@ -326,6 +326,59 @@ class UIMemoryDB(QdrantBase):
if similarity_threshold is None:
similarity_threshold = 0.85
def _evaluate_payload(payload: dict, score: float = 1.0, point_id: str = None) -> Optional[dict]:
solution = payload.get("solution")
confidence = payload.get("confidence", self.DEFAULT_CONFIDENCE)
# ── TTL-based confidence decay ──
stored_at = payload.get("stored_at", 0)
if stored_at > 0:
import time as _time
age_hours = (_time.time() - stored_at) / 3600
time_decay = min(age_hours * 0.05, 0.4) # Max 40% decay over 8 hours
effective_confidence = confidence - time_decay
else:
effective_confidence = confidence
# Filter 1: Never return NOT_FOUND/error entries from cache
if isinstance(solution, dict) and solution.get("type") == "error":
logger.debug(f"Purging poisoned NOT_FOUND entry for '{intent}' from Qdrant Memory.")
if point_id:
self._delete_point(point_id)
return None
# Filter 2: Skip low-confidence entries
if effective_confidence < self.MIN_CONFIDENCE:
logger.debug(f"Skipping expired/low-confidence ({effective_confidence:.2f}, raw: {confidence:.2f}) entry for '{intent}'.")
if effective_confidence < 0.1 and point_id:
self._delete_point(point_id)
return None
return {"solution": solution, "effective_confidence": effective_confidence}
# 0. Zero-cost deterministic ID exact match (bypasses LLM embedding API)
try:
point_id = self._deterministic_id(intent)
exact_points = self.client.retrieve(
collection_name=self.collection_name,
ids=[point_id],
with_payload=True,
with_vectors=False,
)
if exact_points:
eval_result = _evaluate_payload(exact_points[0].payload, score=1.0, point_id=point_id)
if eval_result:
logger.debug(f"Resolved intent '{intent}' from Qdrant Memory via EXACT ID MATCH! (Confidence: {eval_result['effective_confidence']:.2f})")
return eval_result["solution"]
# If exact match failed evaluation (e.g. decayed), we shouldn't fall back to vector search because it's the exact intent!
return None
except Exception as e:
logger.debug(f"Qdrant exact match retrieval error: {e}")
if exact_only:
return None
# 1. Fallback to vector similarity (requires embedding API call)
# Embed ONLY the intent. We learn elements globally to drastically reduce LLM calls.
text_to_embed = f"Intent: {intent}"
vector = self._get_embedding(text_to_embed)
@@ -341,38 +394,11 @@ class UIMemoryDB(QdrantBase):
).points
if results and results[0].score >= similarity_threshold:
payload = results[0].payload
solution = payload.get("solution")
confidence = payload.get("confidence", self.DEFAULT_CONFIDENCE)
# ── TTL-based confidence decay ──
# Entries lose ~5% confidence per hour if not re-validated.
# This prevents stale, poisoned entries from persisting forever.
stored_at = payload.get("stored_at", 0)
if stored_at > 0:
import time as _time
age_hours = (_time.time() - stored_at) / 3600
time_decay = min(age_hours * 0.05, 0.4) # Max 40% decay over 8 hours
effective_confidence = confidence - time_decay
else:
effective_confidence = confidence
# Filter 1: Never return NOT_FOUND/error entries from cache
if isinstance(solution, dict) and solution.get("type") == "error":
logger.debug(f"Purging poisoned NOT_FOUND entry for '{intent}' from Qdrant Memory.")
self._delete_point(results[0].id)
return None
# Filter 2: Skip low-confidence entries (using effective confidence with TTL decay)
if effective_confidence < self.MIN_CONFIDENCE:
logger.debug(f"Skipping expired/low-confidence ({effective_confidence:.2f}, raw: {confidence:.2f}) entry for '{intent}'.")
# If it's really old and decayed, just delete it
if effective_confidence < 0.1:
self._delete_point(results[0].id)
return None
logger.debug(f"Resolved intent '{intent}' from Qdrant Memory! (Score: {results[0].score:.3f}, Confidence: {effective_confidence:.2f})")
return solution
eval_result = _evaluate_payload(results[0].payload, score=results[0].score, point_id=results[0].id)
if eval_result:
logger.debug(f"Resolved intent '{intent}' from Qdrant Memory via vector search! (Score: {results[0].score:.3f}, Confidence: {eval_result['effective_confidence']:.2f})")
return eval_result["solution"]
return None
else:
score = results[0].score if results else 0.0
logger.debug(f"No high-confidence match found in Qdrant (Best score: {score:.3f})")

View File

@@ -256,28 +256,56 @@ class SituationalAwarenessEngine:
xml_lower = xml_dump.lower()
# ── DANGER: Action Blocked (must remain hardcoded — account safety) ──
blocked_markers = [
"try again later", "action blocked", "restrict certain activity",
"help us confirm you own", "confirm it's you",
"später erneut versuchen", "bestätige, dass du es bist",
"handlung blockiert", "eingeschränkt",
]
if any(m in xml_lower for m in blocked_markers):
return SituationType.DANGER_ACTION_BLOCKED
# Guard: Check if the text matches are relatively isolated (e.g. short strings).
# If the string is buried inside a 200-character caption, it's a false positive.
# We can regex match text="..." attributes that are less than 60 characters total,
# OR just use the compressed string where text is capped at 60 chars anyway.
compressed_lower = self._compress_xml(xml_dump).lower()
if any(re.search(rf"(?:text|desc)='[^']*?{m}[^']*?'", compressed_lower) for m in blocked_markers):
# To be extra safe against false positives, check if there's a dialog/modal container
if "dialog" in compressed_lower or "bottom_sheet" in compressed_lower or "alert" in compressed_lower:
return SituationType.DANGER_ACTION_BLOCKED
# ── Hardware Guard: Screen Off / Locked ──
if not getattr(self.device.deviceV2, 'info', {}).get("screenOn", True):
logger.info("📱 [SAE Perceive] Screen is physically OFF.")
return SituationType.OBSTACLE_LOCKED_SCREEN
# ── Foreign Environment Detection (package-based) ──
# Extract ALL packages present in the dump
# ── System Dialog / Permission Detect (Fast Path) ──
packages = set(re.findall(r'package=["\']([^"\']+)["\']', xml_dump))
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
# If Instagram package is not present AT ALL, we are outside the app.
if app_id not in packages:
system_dialog_pkgs = {
'com.google.android.permissioncontroller',
'com.android.permissioncontroller',
'com.samsung.android.permissioncontroller'
}
if any(pkg in system_dialog_pkgs for pkg in packages):
logger.info("📱 [SAE Perceive] System permission dialog explicitly detected.")
return SituationType.OBSTACLE_SYSTEM
# ── Foreign Environment Detection (package-based) ──
# If there are ANY packages that are not our app and not safe system packages,
# we have a foreign overlay/app open!
safe_system_pkgs = {
app_id,
'com.android.systemui',
'android',
'com.google.android.inputmethod.latin', # Gboard
'com.samsung.android.honeyboard', # Samsung Keyboard
'com.sec.android.inputmethod', # Old Samsung Keyboard
'com.touchtype.swiftkey' # SwiftKey
}
has_foreign_package = any(pkg not in safe_system_pkgs for pkg in packages)
if has_foreign_package or app_id not in packages:
# We explicitly ask the TelepathicEngine to classify this to avoid writing brittle substring hacks
# for Android System UI variations across different device manufacturers.
try:
@@ -320,37 +348,81 @@ class SituationalAwarenessEngine:
logger.warning(f"⚠️ [Smart Perceive] LLM Classification failed ({e}). Defaulting to FOREIGN_APP.")
return SituationType.OBSTACLE_FOREIGN_APP
# ── Modal/Obstacle Detection (structural, not ID-based) ──
# Instead of checking specific IDs, we check STRUCTURAL patterns:
# A modal is any overlay that has clickable elements AND covers main content
# ── Modal/Obstacle Detection (Autonomous LLM + Memory) ──
# We explicitly query ScreenMemoryDB. If unknown, we ask the LLM.
# This replaces ALL brittle string/ID matching for modals.
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
compressed = self._compress_xml(xml_dump)
# ── Structural Fast-Check: Content-Creation Overlays ──
# These full-screen overlays live INSIDE Instagram's package but block
# all normal navigation. They are invisible to the foreign-app detector
# and frequently fool the LLM into thinking they are "normal" browsing.
# Detecting them structurally is O(1) and requires ZERO LLM calls.
creation_flow_markers = (
'quick_capture', # Camera / story capture overlay
'gallery_cancel_button', # Story gallery "Back to Home" button
'creation_flow', # Post creation wizard
'reel_camera', # Reel recording interface
)
# Guard: Check against compressed string to ensure these markers ONLY appear
# as resource IDs (e.g. "id=quick_capture_...") and not as plain text in
# user comments/bios (which would look like "text='... creation_flow ...'")
if any(re.search(rf'id=[^\s|]*{marker}', compressed, re.IGNORECASE) for marker in creation_flow_markers):
logger.info("🧠 [SAE Perceive] Content-creation overlay detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "NORMAL":
return SituationType.NORMAL
# If not cached, query LLM for autonomous structural classification
try:
clean = re.sub(r'<\?xml.*?\?>', '', xml_dump).strip()
root = ET.fromstring(clean)
# Find all top-level FrameLayouts from the Instagram package
ig_frames = []
for elem in root.iter('node'):
if elem.get('package') == app_id:
ig_frames.append(elem)
break # Get the root Instagram frame
if ig_frames:
ig_root = ig_frames[0]
# Walk the tree and look for overlay patterns:
# - FrameLayout children that overlap with the main content
# - Nodes with "dialog", "sheet", "modal", "overlay" in their resource-id
# - Nodes with dismiss/close/cancel/not now in text
for elem in ig_root.iter('node'):
rid = elem.get('resource-id', '').lower()
# Structural modal detection — ANY container with these words
if any(kw in rid for kw in ['dialog', 'sheet', 'modal', 'overlay', 'survey', 'interstitial', 'popup']):
# Check if it has actual content (not an empty container)
children = list(elem)
if len(children) > 0:
logger.info(f"🔍 [SAE Perceive] Modal/overlay detected via structural scan: {rid}")
return SituationType.OBSTACLE_MODAL
except Exception:
pass
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.config import Config
prompt = (
"You are a Situation Classifier for a mobile automation agent.\n"
"Analyze the given Android UI XML dump. Is there a blocking MODAL, DIALOG, or POPUP "
"covering the screen that needs to be dismissed, or is this a NORMAL usable screen?\n"
"A 'clean_sheet_container' with standard Instagram feed content is NORMAL.\n"
"A survey, rating prompt, 'not now' prompt, or permission dialog is an OBSTACLE_MODAL.\n"
"An 'Add to story' screen, camera interface, 'quick_capture' layout, gallery picker, "
"or ANY content-creation flow (reel recording, post editor, live mode) is an OBSTACLE_MODAL — "
"it blocks normal navigation and must be dismissed.\n"
"Respond ONLY with a valid JSON object strictly matching this schema: "
"{\"situation\": \"OBSTACLE_MODAL\" | \"NORMAL\"}\n\n"
f"XML:\n{compressed[:2500]}"
)
args = {}
try: args = Config().args
except Exception: pass
model = getattr(args, "ai_telepathic_model", "qwen3.5:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
res = query_telepathic_llm(model=model, url=url, system_prompt="Strict JSON classifier.", user_prompt=prompt, use_local_edge=True)
import json
data = json.loads(res)
situ_str = data.get("situation", "NORMAL")
if situ_str == "OBSTACLE_MODAL":
logger.info("🧠 [Smart Perceive] Screen classified as: OBSTACLE_MODAL.")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
else:
logger.info("🧠 [Smart Perceive] Screen classified as: NORMAL.")
screen_memory.store_screen(compressed, "NORMAL")
return SituationType.NORMAL
except Exception as e:
logger.warning(f"⚠️ [Smart Perceive] Modal classification failed ({e}). Defaulting to NORMAL.")
return SituationType.NORMAL
@@ -358,118 +430,9 @@ class SituationalAwarenessEngine:
# 2. PLAN: AI-driven escape strategy
# ──────────────────────────────────────────────
def _plan_escape_via_structure(self, xml_dump: str, situation_type: SituationType, failed_this_session: set = None) -> EscapeAction:
"""
Unified structural escape planning.
Scans for ALL potential dismissal candidates (buttons or BACK) and scores them.
"""
if failed_this_session is None:
failed_this_session = set()
try:
clean = re.sub(r'<\?xml.*?\?>', '', xml_dump).strip()
root = ET.fromstring(clean)
except Exception:
return EscapeAction("back", reason="XML parse failed, pressing BACK as fallback")
# ── EXTREME PRIORITY: Force foreground if foreign app ──
if situation_type == SituationType.OBSTACLE_FOREIGN_APP:
return EscapeAction("app_start", reason="Foreign app detected, forcing Instagram to foreground")
# ── EXTREME PRIORITY: Unlock if locked ──
if situation_type == SituationType.OBSTACLE_LOCKED_SCREEN:
return EscapeAction("unlock", reason="Hardware condition: Device lock screen detected")
# ── UNIFIED SCAN ──
dismiss_text_keywords = [
'not now', 'cancel', 'dismiss', 'skip', 'deny', 'don\'t allow',
'no thanks', 'nicht jetzt', 'abbrechen', 'schließen',
'überspringen', 'ablehnen', 'nein danke', 'später',
'maybe later', 'vielleicht später', 'close', 'done',
]
dangerous_id_patterns = [
'follow', 'unfollow', 'mute', 'block', 'report',
'restrict', 'hide', 'favorite', 'close_friend',
'share', 'send', 'delete', 'archive', 'confirm',
]
candidates = []
# 1. Add "BACK" as a baseline candidate
back_key = "back:0,0"
if back_key not in failed_this_session:
# We don't add BACK for SYSTEM popups initially as it's often ignored
if situation_type == SituationType.OBSTACLE_MODAL:
candidates.append(EscapeAction("back", reason="Safest method: Try BACK first for modals"))
elif situation_type != SituationType.OBSTACLE_SYSTEM:
candidates.append(EscapeAction("back", reason="Generic BACK fallback candidate"))
# 2. Add clickable buttons from the XML
for elem in root.iter('node'):
if elem.get('clickable', 'false') != 'true':
continue
text = elem.get('text', '').strip().lower()
desc = elem.get('content-desc', '').strip().lower()
rid = elem.get('resource-id', '').strip().lower()
bounds = elem.get('bounds', '')
# Safety Guards
if any(dp in rid for dp in dangerous_id_patterns):
continue
match = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds)
if not match:
continue
l, t, r, b = map(int, match.groups())
cx, cy = (l + r) // 2, (t + b) // 2
if f"click:{cx},{cy}" in failed_this_session:
continue
# Check for dismissal intent in text/desc
searchable = f"{text} {desc}"
if any(kw in searchable for kw in dismiss_text_keywords):
candidates.append(EscapeAction(
"click", cx, cy,
reason=f"Found explicit dismiss button: '{text or desc}'",
resource_id=rid
))
if not candidates:
logger.warning("🔍 [SAE Plan] No candidates found in structural scan!")
return EscapeAction("back", reason="No candidates found, attempting final BACK press as a last resort")
# 3. Score candidates to pick the best move
# Priority: -1 (Highest) -> 5 (Lowest)
def get_priority(ea):
# BACK fallback
if ea.action_type == "back":
# For modals, BACK is the standard "safe" way out and must be tried first
if situation_type == SituationType.OBSTACLE_MODAL:
return -1
return 4
if ea.action_type == "click":
r = ea.reason.lower()
# Explicit denials (High priority)
if any(k in r for k in ['not now', 'cancel', 'deny', 'don\'t allow', 'dismiss']):
return 0
# Generic close (Medium-High)
if any(k in r for k in ['close', 'schließen', 'skip', 'done']):
return 1
return 3
return 5
for c in candidates:
logger.debug(f"🔍 [SAE Plan] Candidate: type={c.action_type} reason='{c.reason}' priority={get_priority(c)}")
candidates.sort(key=get_priority)
return candidates[0]
def _plan_escape_via_llm(self, xml_dump: str, compressed: str, situation_type: SituationType) -> Optional[EscapeAction]:
def _plan_escape_via_llm(self, xml_dump: str, compressed: str, situation_type: SituationType, failed_actions: set = None) -> Optional[EscapeAction]:
"""
LLM-powered escape planning for situations where structural scan fails.
Called ONLY when recall AND structural planning both miss.
@@ -491,21 +454,26 @@ class SituationalAwarenessEngine:
"Analyze the screen content and return a JSON escape action.\n\n"
"Rules:\n"
"- If you see a dismiss/close/cancel/skip/not now button, click it\n"
"- If you see a foreign app (not Instagram), press BACK\n"
"- If the Situation type is OBSTACLE_LOCKED_SCREEN, action must be 'unlock'\n"
"- If the Situation type is OBSTACLE_FOREIGN_APP, action must be 'kill_foreign_apps'\n"
"- If there is NO obstacle and the screen is a normal Instagram view (false positive), action must be 'false_positive'\n"
"- If nothing else works, suggest 'app_start' to force-reopen Instagram\n"
"- NEVER click 'OK'/'Confirm'/'Accept' on surveys or prompts\n"
"- Return ONLY valid JSON: {\"action\": \"click\"|\"back\"|\"app_start\", \"x\": N, \"y\": N, \"reason\": \"...\"}"
"- Return ONLY valid JSON: {\"action\": \"click\"|\"back\"|\"app_start\"|\"unlock\"|\"kill_foreign_apps\"|\"false_positive\", \"x\": N, \"y\": N, \"reason\": \"...\"}"
)
user_prompt = (
f"Situation type: {situation_type.value}\n\n"
f"Screen content:\n{compressed}\n\n"
f"What action should I take to clear this obstacle and return to Instagram? Return JSON only."
)
if failed_actions:
user_prompt += f"Failed actions this session (DO NOT REPEAT): {list(failed_actions)}\n\n"
user_prompt += "What action should I take to clear this obstacle and return to Instagram? Return JSON only."
try:
resp = query_llm(url=url, model=model, prompt=user_prompt, system=system_prompt,
format_json=True, timeout=30, max_tokens=100, temperature=0.0)
format_json=True, timeout=30, max_tokens=300, temperature=0.0)
if resp and "response" in resp:
import json
data = json.loads(resp["response"])
@@ -550,6 +518,23 @@ class SituationalAwarenessEngine:
self.device.app_start(app_id, use_monkey=True)
random_sleep(2.0, 3.5)
elif action.action_type == "kill_foreign_apps":
logger.info(f"🔪 [SAE Act] Killing foreign apps: {action.reason}")
# The reason string will contain the package name or 'all'
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
try:
# We can dump current package again, or just get it from device
current_pkg = self.device.deviceV2.app_current().get("package")
if current_pkg and current_pkg != app_id and current_pkg not in ('com.android.systemui', 'android'):
logger.info(f"🔪 Stopping {current_pkg}")
self.device.app_stop(current_pkg)
random_sleep(1.0, 2.0)
# Ensure we are back
self.device.app_start(app_id, use_monkey=True)
random_sleep(1.5, 2.5)
except Exception as e:
logger.debug(f"Failed to kill foreign app: {e}")
elif action.action_type == "home_then_app":
logger.info(f"🏠 [SAE Act] HOME → App Start: {action.reason}")
self.device.press("home")
@@ -571,6 +556,9 @@ class SituationalAwarenessEngine:
from GramAddict.core.exceptions import ActionBlockedError
failed_this_session = set()
cleared_something = False
last_situation = None
situation_attempts = 0
for attempt in range(max_attempts):
# ── PERCEIVE ──
@@ -580,6 +568,11 @@ class SituationalAwarenessEngine:
xml_dump = self.device.dump_hierarchy()
situation = self.perceive(xml_dump)
if last_situation != situation:
situation_attempts = 0
last_situation = situation
failed_this_session.clear()
# ── CLEAR: Nothing to do ──
if situation == SituationType.NORMAL:
@@ -608,43 +601,57 @@ class SituationalAwarenessEngine:
action = EscapeAction.from_dict(recalled)
logger.info(f"🧠 [SAE] Using recalled strategy: {action.reason}")
else:
logger.info(f"🧠 [SAE] Recalled strategy already failed this session. Using structural planning.")
logger.info(f"🧠 [SAE] Recalled strategy already failed this session. Using LLM planning.")
recalled = None
if not recalled:
if attempt < 3:
action = self._plan_escape_via_structure(xml_dump, situation, failed_this_session)
elif attempt < 5:
logger.info("🧠 [SAE] Escalating to LLM-assisted escape planning...")
action = self._plan_escape_via_llm(xml_dump, compressed, situation)
elif attempt == 5:
action = EscapeAction("app_start", reason=f"Escalation level 6: force app restart after {attempt} failed attempts")
if situation_attempts < 3:
logger.info("🧠 [SAE] Autonomous Blank Start: Escalating to LLM-assisted escape planning...")
action = self._plan_escape_via_llm(xml_dump, compressed, situation, failed_this_session)
elif situation_attempts == 3:
action = EscapeAction("app_start", reason=f"Escalation level 4: force app restart after {situation_attempts} failed attempts on this situation")
else:
action = EscapeAction("home_then_app", reason=f"Nuclear escalation: HOME + app_start after {attempt} failed attempts")
action = EscapeAction("home_then_app", reason=f"Nuclear escalation: HOME + app_start after {situation_attempts} failed attempts on this situation")
# ── EXECUTE ──
self._execute_escape(action)
if action.action_type == "false_positive":
logger.warning(f"🧠 [SAE Unlearn] LLM identified false positive obstacle. Overwriting Qdrant memory to NORMAL.")
from GramAddict.core.qdrant_memory import ScreenMemoryDB
ScreenMemoryDB().store_screen(compressed, "NORMAL")
self._consecutive_failures = 0
return True
else:
self._execute_escape(action)
cleared_something = True
# ── VERIFY ──
post_xml = self.device.dump_hierarchy()
post_situation = self.perceive(post_xml)
success = (post_situation == SituationType.NORMAL)
reached_normal = (post_situation == SituationType.NORMAL)
situation_changed = (post_situation != situation)
# ── LEARN ──
self.episodes.learn(compressed, action, success)
if success:
if reached_normal:
# ── LEARN FULL SUCCESS ──
self.episodes.learn(compressed, action, True)
logger.info(f"✅ [SAE] Obstacle cleared! Strategy '{action.reason}' worked. Stored for future recall.")
self._consecutive_failures = 0
return True
elif situation_changed:
# ── LEARN PARTIAL SUCCESS ──
self.episodes.learn(compressed, action, True)
logger.info(f"🔄 [SAE] Situation changed from {situation.value} to {post_situation.value}. Continuing recovery...")
# We do not increment consecutive_failures or situation_attempts because we made progress
# The next loop iteration will clear failed_this_session since last_situation != situation
else:
# ── LEARN FAILURE ──
self.episodes.learn(compressed, action, False)
action_key = f"{action.action_type}:{action.x},{action.y}"
failed_this_session.add(action_key)
logger.warning(
f"⚠️ [SAE] Escape attempt {attempt + 1} failed ('{action.reason}'). Trying next strategy..."
)
self._consecutive_failures += 1
situation_attempts += 1
logger.error(f"❌ [SAE] UNRECOVERABLE: Failed to clear screen after {max_attempts} attempts")
return False

View File

@@ -7,6 +7,7 @@ import json
import os
import time
from typing import Optional, Tuple, Dict, Any
from colorama import Fore
from GramAddict.core.qdrant_memory import QdrantBase
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.diagnostic_dump import dump_ui_state
@@ -282,6 +283,7 @@ class TelepathicEngine:
'creation_tab', 'live_button', 'go_live',
'close_friend', 'close_friends',
'reel_empty_badge', # This is the "Add to story" badge that caused the real-world bug
'quick_capture', # Camera/story capture overlay — absolute navigation trap
]
def _is_forbidden_action(self, node: dict) -> bool:
@@ -379,6 +381,14 @@ class TelepathicEngine:
# Silently filter non-bottom elements for navigation intents to prevent log spam
return False
# 3.5. Reject Action Bars for Content Intents
# If we are looking for user content (posts, grid items, media), never click an action bar or tab bar.
is_content_intent = any(k in low_intent for k in ["post", "grid", "media", "photo", "video", "reel", "item"])
res_id_lower = node.get("resource_id", "").lower()
is_generic_bar = any(k in res_id_lower for k in ["action_bar", "tab_bar", "navigation_bar", "tab_layout"])
if is_content_intent and is_generic_bar:
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.
@@ -511,7 +521,8 @@ class TelepathicEngine:
"y": best_match["y"],
"score": 1.0,
"semantic": best_match.get("semantic_string", "Escape Hatch"),
"source": "keyword_fast_path"
"source": "keyword_fast_path",
"original_attribs": best_match.get("original_attribs", {})
}
# Expand known Instagram aliases to avoid sending UI basics to the LLM mappings
@@ -522,12 +533,21 @@ class TelepathicEngine:
"like": ["heart"],
"comment": ["reply"],
"profile": ["user", "account"],
# Structural feed content aliases for autonomous extraction
"author": ["name", "profile_name", "owner"],
"username": ["name"],
"header": ["header"],
"post": ["feed", "photo", "carousel", "clips"],
"media": ["imageview", "video", "clips", "media_group", "carousel"],
"image": ["imageview"],
"content": ["media", "group", "imageview", "video", "caption"],
"description": ["content-desc", "caption"],
}
scored = []
for node in nodes:
sem = node.get("semantic_string", "").lower()
rid = node.get("resource_id", "").lower().replace("_", " ")
rid = node.get("resource_id", "").lower().replace("_", " ").replace("/", " ")
desc_text = node.get("original_attribs", {}).get("desc", "").lower()
node_text = node.get("original_attribs", {}).get("text", "").lower()
@@ -604,7 +624,8 @@ class TelepathicEngine:
"score": 0.95, # High confidence — deterministic match
"semantic": best_node["semantic_string"],
"area": best_node.get("area", 0),
"source": "keyword"
"source": "keyword",
"original_attribs": best_node.get("original_attribs", {})
}
# ──────────────────────────────────────────────
@@ -776,11 +797,16 @@ class TelepathicEngine:
# ── Modal Guard ──
# If a bottom sheet or dialog is active, it likely obscures the main navigation tabs.
# We scan the raw XML to catch non-interactable modals (like those in Reels traps).
# We rely strictly on the SAE (Situational Awareness Engine) for 100% autonomous detection
# (via Qdrant cache or LLM reasoning) instead of hardcoded resource IDs.
is_nav_intent = any(k in intent_lower for k in ["tab", "navigation", "reels tab", "profile tab", "home tab", "explore tab", "message tab"])
if is_nav_intent and self._is_modal_active(interactive_nodes, raw_xml_string=xml_hierarchy):
logger.warning(f"🛡️ [Modal Guard] A bottom sheet or dialog is blocking the screen. Refusing to seek nav-intent '{intent_description}'.")
return {"blocked_by_modal": True}
if is_nav_intent and device is not None:
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
sae = SituationalAwarenessEngine(device)
situation = sae.perceive(xml_hierarchy)
if situation == SituationType.OBSTACLE_MODAL:
logger.warning(f"🛡️ [Modal Guard] SAE detected a modal blocking the screen. Refusing to seek nav-intent '{intent_description}'.")
return {"blocked_by_modal": True}
# ── DM Thread (Forbidden Zone) Guard ──
# If the bot is trapped inside a DM thread, it should NOT attempt to find
@@ -857,8 +883,20 @@ class TelepathicEngine:
self._memory = self._load_json(MEMORY_FILE) # Reload for freshness
if intent_description in self._memory:
known_semantics = self._memory[intent_description]
# Anti-poisoning: for dynamic intents, strip text/desc to match structural signatures
dynamic_intents = {
"tap post username", "tap post author", "tap story ring avatar",
"tap feed item", "explore grid", "grid item", "first image", "profile grid",
"tap user profile", "post media content"
}
is_dynamic = any(d in intent_description.lower() for d in dynamic_intents)
for n in viable_nodes:
sem_str = n["semantic_string"]
if is_dynamic:
sem_str = re.sub(r"(text|description):\s*'[^']*',?\s*", "", sem_str).strip()
sem_str = re.sub(r",\s*id context:", "id context:", sem_str).strip()
# Check if this semantic string is in memory (handles both old list format and new dict format)
if isinstance(known_semantics, dict) and sem_str in known_semantics:
@@ -889,7 +927,8 @@ class TelepathicEngine:
"y": n["y"],
"score": confidence,
"semantic": f"Memory Match: {sem_str}",
"source": "memory"
"source": "memory",
"original_attribs": n.get("original_attribs", {})
}
elif isinstance(known_semantics, list) and sem_str in known_semantics:
# Legacy fallback logic for old JSON list format
@@ -904,7 +943,8 @@ class TelepathicEngine:
"y": n["y"],
"score": 1.0,
"semantic": f"Memory Match: {sem_str}",
"source": "memory"
"source": "memory",
"original_attribs": n.get("original_attribs", {})
}
# ── Stage 1.5: Deterministic Keyword Fast Path ──
@@ -962,7 +1002,8 @@ class TelepathicEngine:
"y": best_node["y"],
"score": best_score,
"semantic": best_node["semantic_string"],
"source": "vector"
"source": "vector",
"original_attribs": best_node.get("original_attribs", {})
}
elif scored_nodes:
logger.warning(f"⚠️ [Telepathic] Low confidence ({scored_nodes[0][1]:.3f} < {min_confidence}) for '{intent_description}'.")
@@ -970,7 +1011,8 @@ class TelepathicEngine:
# ── Stage 3: Telepathic LLM Fallback (Text-Based XML Reasoning) ──
if device:
logger.info(f"🧠 [Agentic Fallback] Activating structural LLM reasoning for: '{intent_description}'")
return self._vision_cortex_fallback(intent_description, viable_nodes, device, screen_height)
goal = kwargs.get('goal', None)
return self._vision_cortex_fallback(intent_description, viable_nodes, device, screen_height, goal=goal)
return None
@@ -989,8 +1031,18 @@ class TelepathicEngine:
xml = device.dump_hierarchy()
nodes = self._extract_semantic_nodes(xml)
# Identify grid nodes (posts)
grid_nodes = [n for n in nodes if any(k in n.get("original_attribs", {}).get("resource-id", "") for k in ["image_button", "grid_card_layout_container", "imageview", "button"])]
# Identify grid nodes (posts) dynamically without hardcoded IDs
# Look for interactive elements that are primarily visual (no text) or have photo/video descriptions
grid_nodes = []
for n in nodes:
orig = n.get("original_attribs", {})
desc = orig.get("content-desc", "").lower()
has_text = bool(orig.get("text", ""))
# A grid post is typically a visual block without raw UI text,
# or explicitly labeled as photo/video by accessibility layers.
if not has_text and ("photo" in desc or "video" in desc or "image" in desc or "post" in desc or desc == ""):
grid_nodes.append(n)
if not grid_nodes:
logger.warning("👁️ [Vision Core] No grid items found to evaluate. Falling back to default navigation.")
@@ -1062,20 +1114,85 @@ class TelepathicEngine:
idx = data.get("best_index")
if idx is not None and 0 <= idx < len(grid_nodes):
chosen = grid_nodes[idx]
logger.info(f"✅ [Vision Match] Cell {idx} chosen: {data.get('reason')}", extra={"color": f"{Fore.GREEN}"})
logger.info(f"✅ [Vision Match] Cell {idx} chosen: {data.get('reason')}", extra={"color": Fore.GREEN})
self._track_click(f"Visual Grid Selection ({idx})", chosen)
return {
"x": chosen["x"],
"y": chosen["y"],
"score": 0.99,
"semantic": f"Visual match {idx}: {data.get('reason')}",
"source": "vlm_grid"
"source": "vlm_grid",
"original_attribs": chosen.get("original_attribs", {})
}
except Exception as e:
logger.error(f"👁️ [Vision Core] Grid evaluation failed: {e}")
return None
def evaluate_post_vibe(self, device, persona_interests: list[str]) -> Optional[dict]:
"""
Takes a screenshot of the current post/reel and asks the VLM to evaluate its
aesthetic quality and niche alignment. Returns a dict with quality_score and matches_niche.
"""
logger.info(f"👁️ [Vision Core] Capturing post screenshot for Content Vibe Check...", extra={"color": "\033[36m"})
try:
screenshot_b64 = device.get_screenshot_b64()
except Exception as e:
logger.error(f"👁️ [Vision Core] Failed to capture post screenshot: {e}")
return None
system_prompt = (
"You are a strict aesthetic evaluator for an Instagram growth agent. "
"You are looking at a screenshot of a single Instagram post or Reel. "
"Evaluate the visual content, subject matter, and aesthetic quality. "
"Return a JSON object: {\"quality_score\": number (1-10), \"matches_niche\": boolean, \"reason\": \"string\"}. "
"Extremely generic, spammy, or unrelated content should get a low score (< 5). "
"High quality, aesthetic, or highly niche-aligned content gets >= 7."
)
user_prompt = (
f"Niche/Interests: {', '.join(persona_interests) if persona_interests else 'Aesthetic / General Quality'}\n\n"
"Evaluate the provided post screenshot strictly."
)
try:
from GramAddict.core.llm_provider import query_llm
args = getattr(device, "args", None)
model = getattr(args, "ai_telepathic_model", "llama3.2-vision") if args else "llama3.2-vision"
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate") if args else "http://localhost:11434/api/generate"
resp_dict = query_llm(
url=url,
model=model,
prompt=user_prompt,
system=system_prompt,
format_json=True,
images_b64=[screenshot_b64],
max_tokens=200,
temperature=0.2,
timeout=45
)
if resp_dict and "response" in resp_dict:
import json
try:
res_text = resp_dict["response"]
if res_text.startswith("```json"):
res_text = res_text[7:]
if res_text.endswith("```"):
res_text = res_text[:-3]
data = json.loads(res_text.strip())
return data
except Exception as e:
logger.error(f"👁️ [Vision Core] Failed to parse JSON from VLM: {e}\nResponse: {resp_dict['response']}")
except Exception as e:
logger.error(f"👁️ [Vision Core] Error calling VLM for post check: {e}")
return None
def evaluate_profile_vibe(self, device, persona_interests: list[str]) -> Optional[dict]:
"""
[Phase 1] High-fidelity Target Profile Vibe Check.
@@ -1141,7 +1258,7 @@ class TelepathicEngine:
def _grid_fast_path(self, intent_description: str, viable_nodes: list, skip_positions: set = None) -> Optional[dict]:
"""
Deterministic grid navigation: filters for image_button nodes,
Deterministic grid navigation: structurally filters for likely grid items (images, no text),
sorts by (y, x), and returns the topmost-leftmost candidate.
skip_positions: set of (x, y) tuples to skip on retry, ensuring
@@ -1150,7 +1267,16 @@ class TelepathicEngine:
if skip_positions is None:
skip_positions = set()
grid_nodes = [n for n in viable_nodes if any(k in n.get("resource_id", "") for k in ["image_button", "grid_card_layout_container"])]
# Structure-based grid detection: interactive visual elements with no text
grid_nodes = []
for n in viable_nodes:
orig = n.get("original_attribs", {})
desc = orig.get("content-desc", "").lower()
has_text = bool(orig.get("text", ""))
if not has_text and ("photo" in desc or "video" in desc or "image" in desc or "post" in desc or desc == ""):
grid_nodes.append(n)
if not grid_nodes:
return None
@@ -1178,7 +1304,8 @@ class TelepathicEngine:
"y": candidate["y"],
"score": 0.98,
"semantic": candidate["semantic_string"],
"source": "grid_fastpath"
"source": "grid_fastpath",
"original_attribs": candidate.get("original_attribs", {})
}
# All grid nodes were in skip_positions
@@ -1190,12 +1317,13 @@ class TelepathicEngine:
[Phase 2] Hardened Fast Path with Qdrant Self-Learning.
Absolutely deterministic resource-ID targeting for core application navigation
(like direct messages or post usernames). We query Qdrant first. If empty,
we seed it with legacy fallbacks.
we MUST fall back to Telepathic semantic discovery.
This enforces a 100% autonomous Blank Start architecture.
"""
low_intent = intent_description.lower().strip()
# 0. Query Qdrant Memory first!
mem = self.ui_memory.retrieve_memory(intent_description, "", similarity_threshold=0.9)
mem = self.ui_memory.retrieve_memory(intent_description, "", similarity_threshold=0.9, exact_only=True)
if mem and isinstance(mem, dict):
if mem.get("action") == "tap" and mem.get("resource_id"):
learned_res_id = mem.get("resource_id")
@@ -1208,90 +1336,9 @@ class TelepathicEngine:
"y": n["y"],
"score": 1.0,
"semantic": n["semantic_string"],
"source": "qdrant_nav"
"source": "qdrant_nav",
"original_attribs": n.get("original_attribs", {})
}
# 0.5 Enforce Bootstrapper Lifecycle
# If Qdrant memory is sufficiently populated, ignore hardcoded seeds to enforce 100% self-learning
mem_size = self.ui_memory.get_collection_size()
if isinstance(mem_size, int) and mem_size > 25:
logger.debug(f"🌱 [Bootstrapper] Skipping hardcoded seeds for '{intent_description}' (Qdrant memory populated).")
return None
# 1. Post Username (Feed Profile)
if low_intent in ["tap_post_username", "tap post username"]:
for n in viable_nodes:
res_id = n.get("resource_id", "")
if "row_feed_photo_profile_name" in res_id or "media_header_user" in res_id:
logger.info(f"⚡ [Core Nav Fast Path] Found explicit username mapping for '{intent_description}' -> '{res_id}'")
self._track_click(intent_description, n)
return {
"x": n["x"],
"y": n["y"],
"score": 1.0,
"semantic": n["semantic_string"],
"source": "core_nav"
}
# 2. Direct Message Inbox
if "message" in low_intent and "icon" in low_intent:
for n in viable_nodes:
res_id = n.get("resource_id", "")
if "action_bar" in res_id and "direct" in res_id:
logger.info(f"⚡ [Core Nav Fast Path] Found explicit DM Inbox action bar mapping for '{intent_description}' -> '{res_id}'")
self._track_click(intent_description, n)
return {
"x": n["x"],
"y": n["y"],
"score": 1.0,
"semantic": n["semantic_string"],
"source": "core_nav"
}
if "direct_tab" in res_id:
logger.info(f"⚡ [Core Nav Fast Path] Found explicit DM Tab mapping for '{intent_description}' -> '{res_id}'")
self._track_click(intent_description, n)
return {
"x": n["x"],
"y": n["y"],
"score": 1.0,
"semantic": n["semantic_string"],
"source": "core_nav"
}
# 3. Application Navigation Tabs
tab_mappings = {
"tap_home_tab": "feed_tab",
"tap home tab": "feed_tab",
"tap_explore_tab": "search_tab",
"tap explore tab": "search_tab",
"tap_reels_tab": "clips_tab",
"tap reels tab": "clips_tab",
"tap_messages_tab": "direct_tab",
"tap messages tab": "direct_tab",
"tap_profile_tab": "profile_tab",
"tap profile tab": "profile_tab"
}
if low_intent in tab_mappings:
target_res_id = tab_mappings[low_intent]
for n in viable_nodes:
if target_res_id in n.get("resource_id", "").lower():
logger.warning(f"🌱 [TelepathicEngine] Seeding Qdrant Memory with legacy fast-path: '{intent_description}' -> '{target_res_id}'")
self.ui_memory.store_memory(
intent_description,
"",
{
"resource_id": target_res_id,
"action": "tap"
}
)
self._track_click(intent_description, n)
return {
"x": n["x"],
"y": n["y"],
"score": 1.0,
"semantic": n["semantic_string"],
"source": "core_nav"
}
return None
def _track_click(self, intent: str, node: dict):
@@ -1322,6 +1369,35 @@ class TelepathicEngine:
actual_intent = intent or ctx["intent"]
sem = ctx["semantic_string"]
# ── Qdrant Structural Persistence ──
# Extract resource-id from semantic string to learn structural navigation
if hasattr(self, "ui_memory") and self.ui_memory and self.ui_memory.is_connected:
res_id_match = re.search(r"id context:\s*'([^']+)'", sem)
if res_id_match:
learned_res_id = res_id_match.group(1)
self.ui_memory.store_memory(
intent=actual_intent,
xml_context="", # UI Memory uses intent hashes
solution={"action": "tap", "resource_id": learned_res_id}
)
# ── Anti-Poisoning Guard for JSON Memory ──
dynamic_intents = {
"tap post username", "tap post author", "tap story ring avatar",
"tap feed item", "explore grid", "grid item", "first image", "profile grid",
"tap user profile", "post media content"
}
is_dynamic = any(d in actual_intent.lower() for d in dynamic_intents)
if is_dynamic:
# Strip dynamic text and description to prevent overfitting
sem = re.sub(r"(text|description):\s*'[^']*',?\s*", "", sem).strip()
sem = re.sub(r",\s*id context:", "id context:", sem).strip()
if not sem:
logger.debug(f"⚠️ [Confirmed Learning] Semantic string became empty after stripping dynamic content for '{actual_intent}'. Skipping JSON cache.")
TelepathicEngine._last_click_context = None
return
# Add to positive memory with 100% confidence
if actual_intent not in self._memory:
self._memory[actual_intent] = {}
@@ -1489,7 +1565,7 @@ class TelepathicEngine:
# Vision Cortex Fallback (VLM)
# ──────────────────────────────────────────────
def _vision_cortex_fallback(self, intent: str, nodes: list[dict], device, screen_height: int = 2400) -> Optional[dict]:
def _vision_cortex_fallback(self, intent: str, nodes: list[dict], device, screen_height: int = 2400, goal: str = None) -> Optional[dict]:
"""
Uses a Language Model to identify the correct node from parsed screen XML
when embeddings are insufficient. Opt-in Native Vision Processing via Device Screenshots!
@@ -1597,21 +1673,39 @@ class TelepathicEngine:
"6. ACCOUNT SAFETY: NEVER select buttons that modify account state (Favorite, Mute, Block, Unfollow, Restrict) unless specifically commanded."
)
user_prompt = (
f"Which element should I tap to: {intent}\n\n"
f"Elements:\n{json.dumps(simplified_nodes, indent=1)}\n\n"
"Rules:\n"
"- 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, notifications)\n"
"- IGNORE BOTTOM NAVIGATION TABS (Home, Search, Reels, Message, Profile) if the intent is to interact with a post or comment.\n"
"- SPATIAL GUARD: Elements in the TOP 10% (Y < 240) are USUALLY part of the system status bar. ELEMENT MUST BE BELOW THIS UNLESS IT IS A STORY CIRCLE.\n"
"- SPATIAL GUARD: Elements in the BOTTOM 10% (Y > 2160) are USUALLY part of the navigation bar. ELEMENT MUST BE ABOVE THIS UNLESS IT IS A NAVIGATION TAB.\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"
"- NAVIGATION TABS (Home, Explore, Reels, News, Profile) are ALWAYS in the BOTTOM zone (Y coordinates > 0.90 of screen height).\n"
"Return: {\"index\": number, \"reason\": \"...\"}"
)
if goal:
user_prompt = (
f"Your overarching ultimate GOAL is to: '{goal}'.\n\n"
f"Which element should I tap to progress towards this goal? The specific short-term intent requested was: {intent}\n\n"
f"Elements:\n{json.dumps(simplified_nodes, indent=1)}\n\n"
"Rules:\n"
"- 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, notifications)\n"
"- IGNORE BOTTOM NAVIGATION TABS (Home, Search, Reels, Message, Profile) if the intent is to interact with a post or comment.\n"
"- SPATIAL GUARD: Elements in the TOP 10% (Y < 240) are USUALLY part of the system status bar. ELEMENT MUST BE BELOW THIS UNLESS IT IS A STORY CIRCLE.\n"
"- SPATIAL GUARD: Elements in the BOTTOM 10% (Y > 2160) are USUALLY part of the navigation bar. ELEMENT MUST BE ABOVE THIS UNLESS IT IS A NAVIGATION TAB.\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"
"- NAVIGATION TABS (Home, Explore, Reels, News, Profile) are ALWAYS in the BOTTOM zone (Y coordinates > 0.90 of screen height).\n"
"Return: {\"index\": number, \"reason\": \"...\"}"
)
else:
user_prompt = (
f"Which element should I tap to: {intent}\n\n"
f"Elements:\n{json.dumps(simplified_nodes, indent=1)}\n\n"
"Rules:\n"
"- 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, notifications)\n"
"- IGNORE BOTTOM NAVIGATION TABS (Home, Search, Reels, Message, Profile) if the intent is to interact with a post or comment.\n"
"- SPATIAL GUARD: Elements in the TOP 10% (Y < 240) are USUALLY part of the system status bar. ELEMENT MUST BE BELOW THIS UNLESS IT IS A STORY CIRCLE.\n"
"- SPATIAL GUARD: Elements in the BOTTOM 10% (Y > 2160) are USUALLY part of the navigation bar. ELEMENT MUST BE ABOVE THIS UNLESS IT IS A NAVIGATION TAB.\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"
"- NAVIGATION TABS (Home, Explore, Reels, News, Profile) are ALWAYS in the BOTTOM zone (Y coordinates > 0.90 of screen height).\n"
"Return: {\"index\": number, \"reason\": \"...\"}"
)
resp_str = query_telepathic_llm(model, url, system_prompt, user_prompt, images_b64=images_payload)
data = json.loads(resp_str)
@@ -1652,7 +1746,7 @@ class TelepathicEngine:
logger.warning(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", "reels tab", "profile tab", "home tab", "message tab"])
is_nav_intent = any(k in intent.lower() for k in ["tab", "navigation", "reels tab", "profile tab", "home tab", "message tab", "following", "follower", "followers"])
# NAVIGATION TAB ENFORCEMENT:
# Real navigation tabs (Home, Search, Reels, Store, Profile) are ALWAYS in the bottom zone.
@@ -1691,7 +1785,8 @@ class TelepathicEngine:
"y": match["y"],
"score": 0.85, # Not 1.0 — VLM is provisional, not ground truth
"semantic": f"VLM Match: {match['semantic_string']}",
"source": "agentic_fallback"
"source": "agentic_fallback",
"original_attribs": match.get("original_attribs", {})
}
except Exception as e:
@@ -1699,56 +1794,7 @@ class TelepathicEngine:
return None
def _is_modal_active(self, nodes: list, raw_xml_string: str = "") -> bool:
"""
Detects if a dominant bottom sheet, dialog, or modal is covering the screen.
Scans both semantic nodes and the raw XML hierarchy for markers.
"""
import re
modal_res_ids = {
"com.instagram.android:id/bottom_sheet_container",
"com.instagram.android:id/modal_container",
"com.instagram.android:id/dialog_root",
"com.instagram.android:id/message_box_container",
"com.instagram.android:id/bottom_sheet_drag_handle",
"com.instagram.android:id/bottom_sheet_container_view",
"com.instagram.android:id/comment_composer_text_view"
}
# 1. Structural Regex Check (Fastest and catches 'empty' or non-interactable modals)
if raw_xml_string:
# We must be careful: some containers are always present but empty (e.g. bottom_sheet_camera_container)
# A modal is only "active" if it has actual content or is NOT a self-closing tag.
import re
for rid in modal_res_ids:
if "bottom_sheet_camera_container" in rid:
continue
# Search for the resource-id. If found, check if it's a self-closing node (/>)
# This is a heuristic: if we find the ID and it's immediately followed by "/>", it's empty.
# If it's followed by ">" then more tags, it likely has children.
pattern = rf'resource-id="{re.escape(rid)}"[^>]*>'
match = re.search(pattern, raw_xml_string)
if match:
# Check if the node is self-closing
if not match.group(0).endswith("/>"):
# It might have children. Verify if it actually contains nodes.
# For robustness, we'll verify if there is at least one child node before the next sibling.
logger.debug(f"🛡️ [Modal Guard] Detected potential active modal container: {rid}")
return True
def _ensure_telepathic_agent_context(self):
# Empty placeholder if needed by legacy hooks, or remove if unused.
pass
# 2. Semantic Node Check (Iterative fallback)
for n in nodes:
# Direct resource-ID check
rid = n.get("resource-id", "")
if rid in modal_res_ids:
# Double check that it's actually visible/large
if n.get("visible", True) and n.get("area", 0) > 100000:
return True
# Semantic check (e.g., "Comments" title in a sheet)
if "bottom_sheet" in rid.lower() or "dialog" in rid.lower():
return True
return False