Compare commits
16 Commits
main
...
fix/radica
| Author | SHA1 | Date | |
|---|---|---|---|
| cf21dd3f76 | |||
| 8a2c93fb64 | |||
| 749e82ea14 | |||
| 8d58c3cf89 | |||
| f8fc8ace8e | |||
| dd79e07fb3 | |||
| da1a308c5d | |||
| b626dc6488 | |||
| 5ca5777c4b | |||
| 8729995338 | |||
| d57857d444 | |||
| ac272eadce | |||
| 59e7967458 | |||
| 59cd5477af | |||
| c3f84088aa | |||
| 548fc374ae |
@@ -40,6 +40,9 @@ class CarouselBrowsingPlugin(BehaviorPlugin):
|
||||
if not has_carousel_in_view(xml):
|
||||
return False
|
||||
|
||||
if ctx.shared_state.get("carousel_browsed"):
|
||||
return False
|
||||
|
||||
config = self.get_config(ctx)
|
||||
percentage = float(config.get("percentage", getattr(ctx.configs.args, "carousel_percentage", 0)))
|
||||
return random.random() < (percentage / 100.0)
|
||||
@@ -74,13 +77,25 @@ class CarouselBrowsingPlugin(BehaviorPlugin):
|
||||
# ── 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...")
|
||||
logger.debug(f"📸 [Carousel] Curiosity Peak hit on slide {i+1}. Gazing for {dwell:.1f}s...")
|
||||
sleep(dwell * ctx.sleep_mod)
|
||||
|
||||
xml_before = ctx.device.dump_hierarchy()
|
||||
|
||||
# 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)
|
||||
|
||||
# Brief wait for transition to complete
|
||||
sleep(random.uniform(1.5, 2.5) * ctx.sleep_mod)
|
||||
|
||||
xml_after = ctx.device.dump_hierarchy()
|
||||
xml_delta = abs(len(xml_before) - len(xml_after))
|
||||
|
||||
if xml_before == xml_after or xml_delta < 50:
|
||||
logger.info(f"📸 [Carousel] End of carousel detected on slide {i+1} (UI stable). Stopping swipe.")
|
||||
break
|
||||
|
||||
sleep(random.uniform(1.0, 2.0) * ctx.sleep_mod)
|
||||
ctx.shared_state["carousel_browsed"] = True
|
||||
|
||||
return BehaviorResult(
|
||||
executed=True, interactions=count, metadata={"slides_viewed": count, "curiosity_slide": curiosity_slide}
|
||||
|
||||
@@ -35,7 +35,21 @@ class CloseFriendsGuardPlugin(BehaviorPlugin):
|
||||
return False
|
||||
|
||||
xml = ctx.context_xml if ctx.context_xml else ctx.device.dump_hierarchy()
|
||||
return "enge freunde" in xml.lower() or "close friend" in xml.lower()
|
||||
xml_lower = xml.lower()
|
||||
|
||||
# Structural resource-id markers (language-agnostic, O(1) detection)
|
||||
structural_markers = (
|
||||
"friendly_bubbles_component", # Close friend post in feed
|
||||
"profile_header_close_friend", # Close friend badge on profile
|
||||
"close_friends_badge", # Close friend badge in Reels
|
||||
"close_friends_star", # Green star indicator
|
||||
)
|
||||
for marker in structural_markers:
|
||||
if marker in xml_lower:
|
||||
return True
|
||||
|
||||
# Legacy text fallback for edge cases
|
||||
return "enge freunde" in xml_lower or "close friend" in xml_lower
|
||||
|
||||
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
|
||||
logger.info("💚 [CloseFriendsGuard] Close friends post detected. Skipping...")
|
||||
|
||||
@@ -54,15 +54,39 @@ class FollowPlugin(BehaviorPlugin):
|
||||
return True
|
||||
|
||||
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
|
||||
"""Follow the target user."""
|
||||
"""Follow the target user. ONLY clicks 'Follow' buttons, never 'Following'."""
|
||||
nav_graph = ctx.cognitive_stack.get("nav_graph")
|
||||
if not nav_graph:
|
||||
from GramAddict.core.q_nav_graph import QNavGraph
|
||||
|
||||
nav_graph = QNavGraph(ctx.device)
|
||||
|
||||
if nav_graph.do("tap 'Follow' or 'Following' button"):
|
||||
logger.info(f"🤝 [Follow] Toggled Follow/Following state for @{ctx.username} ✓")
|
||||
# ── CRITICAL SAFETY GUARD ──
|
||||
# Pre-check: verify the button actually says "Follow" (not "Following" or "Requested").
|
||||
# Clicking "Following" opens a dangerous bottom sheet (Unfollow / Add to Favorites / Close Friends).
|
||||
xml = ctx.context_xml if ctx.context_xml else ctx.device.dump_hierarchy()
|
||||
import re
|
||||
|
||||
follow_btn = re.search(
|
||||
r'resource-id="com\.instagram\.android:id/profile_header_follow_button"[^>]*text="([^"]*)"',
|
||||
xml,
|
||||
)
|
||||
if not follow_btn:
|
||||
# Try reversed attribute order
|
||||
follow_btn = re.search(
|
||||
r'text="([^"]*)"[^>]*resource-id="com\.instagram\.android:id/profile_header_follow_button"',
|
||||
xml,
|
||||
)
|
||||
if follow_btn:
|
||||
button_text = follow_btn.group(1).strip().lower()
|
||||
if button_text in ("following", "requested", "message"):
|
||||
logger.info(
|
||||
f"🛡️ [Follow] Button says '{follow_btn.group(1)}' — user already followed. Skipping to avoid bottom sheet."
|
||||
)
|
||||
return BehaviorResult(executed=False, metadata={"reason": "already_following"})
|
||||
|
||||
if nav_graph.do("tap follow button"):
|
||||
logger.info(f"🤝 [Follow] Followed @{ctx.username} ✓")
|
||||
ctx.session_state.add_interaction(source=ctx.username, succeed=True, followed=True, scraped=False)
|
||||
|
||||
# Buffer for follow animations to close
|
||||
|
||||
@@ -47,21 +47,27 @@ class ObstacleGuardPlugin(BehaviorPlugin):
|
||||
logger.warning("⚠️ [ObstacleGuard] System permission dialog detected. Dismissing with BACK...")
|
||||
ctx.device.press("back")
|
||||
sleep(1.5 * ctx.sleep_mod)
|
||||
return BehaviorResult(executed=True, should_skip=True)
|
||||
res = BehaviorResult(executed=True, should_skip=True)
|
||||
res.skip_type = "no_scroll"
|
||||
return res
|
||||
|
||||
# ── Foreign App Takeover (e.g. browser opened, wrong app in foreground) ──
|
||||
if situation == SituationType.OBSTACLE_FOREIGN_APP:
|
||||
logger.warning("⚠️ [ObstacleGuard] Foreign app detected. Pressing BACK to recover...")
|
||||
ctx.device.press("back")
|
||||
sleep(1.5 * ctx.sleep_mod)
|
||||
return BehaviorResult(executed=True, should_skip=True)
|
||||
res = BehaviorResult(executed=True, should_skip=True)
|
||||
res.skip_type = "no_scroll"
|
||||
return res
|
||||
|
||||
# ── On-Screen Keyboard (e.g. hallucinated click on comment field) ──
|
||||
if situation == SituationType.OBSTACLE_KEYBOARD:
|
||||
logger.warning("⚠️ [ObstacleGuard] On-screen Keyboard is open. Pressing BACK to dismiss...")
|
||||
ctx.device.press("back")
|
||||
sleep(1.0 * ctx.sleep_mod)
|
||||
return BehaviorResult(executed=True, should_skip=True)
|
||||
res = BehaviorResult(executed=True, should_skip=True)
|
||||
res.skip_type = "no_scroll"
|
||||
return res
|
||||
|
||||
# ── Instagram Modal / Overlay (survey, "Not Now" prompt, creation flow) ──
|
||||
if situation == SituationType.OBSTACLE_MODAL:
|
||||
@@ -88,6 +94,8 @@ class ObstacleGuardPlugin(BehaviorPlugin):
|
||||
else:
|
||||
ctx.shared_state["consecutive_marker_misses"] = misses + 1
|
||||
|
||||
return BehaviorResult(executed=True, should_skip=True) # Restart loop for same post or next
|
||||
res = BehaviorResult(executed=True, should_skip=True) # Restart loop for same post or next
|
||||
res.skip_type = "no_scroll"
|
||||
return res
|
||||
|
||||
return BehaviorResult(executed=False)
|
||||
|
||||
@@ -34,6 +34,16 @@ class RabbitHolePlugin(BehaviorPlugin):
|
||||
if res_score < 0.8:
|
||||
return False
|
||||
|
||||
screen_type = ctx.shared_state.get("current_screen_type")
|
||||
if not screen_type and ctx.context_xml:
|
||||
from GramAddict.core.perception.screen_identity import ScreenIdentity
|
||||
screen_type = ScreenIdentity(getattr(ctx, "username", "")).identify(ctx.context_xml).get("screen_type")
|
||||
|
||||
from GramAddict.core.perception.screen_identity import ScreenType
|
||||
valid_screens = [ScreenType.HOME_FEED, ScreenType.EXPLORE_GRID, ScreenType.REELS_FEED]
|
||||
if screen_type not in valid_screens:
|
||||
return False
|
||||
|
||||
config = self.get_config(ctx)
|
||||
percentage = float(config.get("percentage", 15))
|
||||
return random.random() < (percentage / 100.0)
|
||||
|
||||
@@ -96,12 +96,44 @@ class StoryViewPlugin(BehaviorPlugin):
|
||||
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)
|
||||
# Atomic state validation after click
|
||||
# Atomic state validation before click
|
||||
xml_dump = ctx.device.dump_hierarchy()
|
||||
if not xml_dump:
|
||||
continue
|
||||
packages = set(re.findall(r'package="([^"]+)"', xml_dump))
|
||||
|
||||
# Query VLM to find the interactive area for the next segment
|
||||
intent = "tap right side of screen to view next story segment"
|
||||
node = ctx.telepathic.find_best_node(xml_dump, intent, device=ctx.device, track=False)
|
||||
|
||||
if node:
|
||||
logger.debug(
|
||||
f"📸 [StoryView] VLM selected node '{node.resource_id or node.content_desc or 'unknown'}' for next segment."
|
||||
)
|
||||
# If VLM selects a large container (e.g. the entire screen or story viewer),
|
||||
# we must tap its right side, not its exact center, to avoid pausing the story.
|
||||
if getattr(node, "area", 0) > (w * h * 0.4):
|
||||
target_x = node.x1 + int((node.x2 - node.x1) * 0.85)
|
||||
target_y = node.y1 + int((node.y2 - node.y1) * 0.25)
|
||||
logger.debug(
|
||||
f"📸 [StoryView] Adjusting click to top-right quadrant of large container: ({target_x}, {target_y})"
|
||||
)
|
||||
else:
|
||||
target_x = node.center_x
|
||||
target_y = node.center_y
|
||||
|
||||
humanized_click(ctx.device, target_x, target_y, sleep_mod=ctx.sleep_mod)
|
||||
else:
|
||||
logger.warning(
|
||||
"📸 [StoryView] VLM could not resolve next story segment. Falling back to geometric safety quadrant."
|
||||
)
|
||||
# Click top-right to avoid 'reply' input fields and most stickers
|
||||
humanized_click(ctx.device, int(w * 0.85), int(h * 0.25), sleep_mod=ctx.sleep_mod)
|
||||
|
||||
# Verify we didn't leave Instagram
|
||||
xml_dump_after = ctx.device.dump_hierarchy()
|
||||
if not xml_dump_after:
|
||||
continue
|
||||
packages = set(re.findall(r'package="([^"]+)"', xml_dump_after))
|
||||
app_id = getattr(ctx.device, "app_id", "com.instagram.android")
|
||||
if packages and app_id not in packages:
|
||||
logger.error(
|
||||
|
||||
@@ -1134,6 +1134,9 @@ def _run_zero_latency_feed_loop(
|
||||
elif skip_type == "fast":
|
||||
_humanized_scroll(device, is_skip=True)
|
||||
sleep(1.0 * sleep_mod)
|
||||
elif skip_type == "no_scroll":
|
||||
logger.debug("⏭️ Skipping post without scrolling (e.g., dismissing obstacle).")
|
||||
sleep(1.0 * sleep_mod)
|
||||
else:
|
||||
_humanized_scroll(device, is_skip=False)
|
||||
sleep(1.5 * sleep_mod)
|
||||
|
||||
@@ -234,6 +234,11 @@ class GoalExecutor:
|
||||
explored_nav_actions.clear()
|
||||
visited_screens.clear()
|
||||
consecutive_back_presses = 0
|
||||
# CRITICAL: Also clear the planner's learned traps.
|
||||
# Without this, traps learned before restart persist and
|
||||
# immediately re-trap the bot on the same (or similar) screen.
|
||||
if hasattr(self, 'planner') and hasattr(self.planner, 'knowledge'):
|
||||
self.planner.knowledge.clear_traps()
|
||||
continue
|
||||
|
||||
# Check if it was a navigation action (vs a goal action). If we are not on the required screen,
|
||||
|
||||
@@ -164,11 +164,28 @@ class NavigationKnowledge:
|
||||
pass
|
||||
return None
|
||||
|
||||
def clear_traps(self):
|
||||
"""Clear all in-memory traps. Called after force-restart to allow fresh routing."""
|
||||
count = len(self._learned_traps)
|
||||
self._learned_traps.clear()
|
||||
if count > 0:
|
||||
logger.info(f"🧹 [NavigationKnowledge] Cleared {count} in-memory traps for fresh routing.")
|
||||
|
||||
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)
|
||||
|
||||
# GUARD: Never persist traps for UNKNOWN screens to Qdrant.
|
||||
# UNKNOWN is a catch-all — persisting traps here permanently blocks
|
||||
# ALL unidentified screens, creating inescapable dead-ends.
|
||||
if screen_type == ScreenType.UNKNOWN:
|
||||
logger.warning(
|
||||
f"🛡️ [Aversive Learning] Trap '{action}' on UNKNOWN kept in-memory only (not persisted). "
|
||||
"UNKNOWN is a catch-all — permanent traps here block all unidentified screens."
|
||||
)
|
||||
return
|
||||
|
||||
if not self._db or not self._db.is_connected:
|
||||
return
|
||||
|
||||
@@ -185,7 +202,11 @@ class NavigationKnowledge:
|
||||
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."""
|
||||
"""Check if an action on this screen is a known trap.
|
||||
|
||||
Traps have time-based expiry: entries older than 30 minutes are
|
||||
auto-forgiven and deleted from Qdrant to prevent permanent dead-ends.
|
||||
"""
|
||||
from GramAddict.core.screen_topology import ScreenTopology
|
||||
|
||||
if ScreenTopology.is_structural_action(screen_type, action):
|
||||
@@ -198,6 +219,8 @@ class NavigationKnowledge:
|
||||
if not self._db or not self._db.is_connected:
|
||||
return False
|
||||
|
||||
TRAP_EXPIRY_SECONDS = 1800 # 30 minutes: old traps expire
|
||||
|
||||
try:
|
||||
from qdrant_client.models import FieldCondition, Filter, MatchValue
|
||||
|
||||
@@ -212,6 +235,20 @@ class NavigationKnowledge:
|
||||
limit=1,
|
||||
)[0]
|
||||
if results:
|
||||
timestamp = results[0].payload.get("timestamp", 0)
|
||||
age_seconds = time.time() - timestamp
|
||||
|
||||
# Time-based expiry: old traps are forgiven
|
||||
if age_seconds > TRAP_EXPIRY_SECONDS:
|
||||
logger.info(
|
||||
f"🔄 [Aversive Decay] Forgave expired trap '{action}' on {screen_type.name} "
|
||||
f"(age: {age_seconds/60:.0f}min). Allowing re-exploration."
|
||||
)
|
||||
# Delete the stale trap from Qdrant
|
||||
seed = f"trap_{trap_key}"
|
||||
self._db.delete_point(seed)
|
||||
return False
|
||||
|
||||
self._learned_traps.add(trap_key)
|
||||
return True
|
||||
except Exception:
|
||||
|
||||
@@ -198,9 +198,10 @@ class ActionMemory:
|
||||
# We NO LONGER bypass VLM verification via string matching.
|
||||
# If confidence is < 0.95, we always do VLM or Delta verification.
|
||||
|
||||
# ── VLM Verification Fallback ──
|
||||
# ── VLM Verification (soft signal, NOT sole authority) ──
|
||||
|
||||
# If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM
|
||||
vlm_verdict = None
|
||||
if device and confidence < 0.95:
|
||||
logger.info(
|
||||
f"👁️ [ActionMemory] Confidence ({confidence:.2f}) < 0.95. Handing over verification for '{intent}' to VLM visual analysis..."
|
||||
@@ -214,7 +215,6 @@ class ActionMemory:
|
||||
if self._last_click_context:
|
||||
clicked_context = f"The element that was tapped: {self._last_click_context['semantic_string']}. "
|
||||
|
||||
# Ask VLM to be the absolute source of truth
|
||||
prompt = (
|
||||
f"The user just attempted to perform the action: '{intent}'. "
|
||||
f"{clicked_context}"
|
||||
@@ -234,7 +234,7 @@ class ActionMemory:
|
||||
f"If the intent was to open a profile, are you on a profile page? "
|
||||
f"If the intent was to go back, are you on the previous screen? "
|
||||
)
|
||||
prompt += "Answer ONLY with the word YES or NO."
|
||||
prompt += 'Answer ONLY with a valid JSON object exactly matching this schema: {"success": true} or {"success": false}. DO NOT add any other keys.'
|
||||
|
||||
try:
|
||||
screenshot = device.get_screenshot_b64()
|
||||
@@ -242,19 +242,21 @@ class ActionMemory:
|
||||
raise ValueError("No screenshot available from device")
|
||||
response = evaluator._query_vlm(prompt, screenshot)
|
||||
|
||||
decision = _parse_yes_no(response) if response else None
|
||||
vlm_verdict = _parse_yes_no(response) if response else None
|
||||
|
||||
if decision is True:
|
||||
if vlm_verdict is True:
|
||||
logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
|
||||
return True
|
||||
elif decision is False:
|
||||
logger.warning(
|
||||
f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'"
|
||||
elif vlm_verdict is False:
|
||||
# VLM says false — but small local VLMs (7B) are unreliable.
|
||||
# Do NOT trust this blindly. Fallthrough to structural delta verification
|
||||
# which is the ground-truth tiebreaker.
|
||||
logger.info(
|
||||
f"🧠 [ActionMemory] VLM says '{intent}' failed — but VLM is unreliable. "
|
||||
"Falling through to structural delta for ground-truth verification."
|
||||
)
|
||||
return False
|
||||
# DO NOT return False here — let structural delta decide
|
||||
else:
|
||||
# VLM returned ambiguous response (JSON, mixed signals, etc.)
|
||||
# Don't treat as hard failure — fall through to structural delta verification
|
||||
logger.debug(
|
||||
f"🧠 [ActionMemory] VLM response for '{intent}' was not YES/NO "
|
||||
f"(got: '{response[:80]}...'). Falling through to structural verification."
|
||||
|
||||
@@ -67,6 +67,14 @@ ALLOWED_SCREENS: Dict[str, FrozenSet[ScreenType]] = {
|
||||
ScreenType.STORY_VIEW, # toolbar_reshare_button exists on stories
|
||||
}
|
||||
),
|
||||
"tap post username": frozenset(
|
||||
{
|
||||
ScreenType.HOME_FEED,
|
||||
ScreenType.POST_DETAIL,
|
||||
ScreenType.REELS_FEED,
|
||||
ScreenType.EXPLORE_GRID,
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
# Intent keywords that trigger the categorical ban check
|
||||
|
||||
@@ -750,6 +750,11 @@ class IntentResolver:
|
||||
if node.text and node.text != node.content_desc:
|
||||
text = _humanize_desc(node.text)
|
||||
label_parts.append(f"text='{text[:50]}'")
|
||||
if node.class_name:
|
||||
cls_short = node.class_name.split(".")[-1]
|
||||
label_parts.append(f"class='{cls_short}'")
|
||||
if node.long_clickable:
|
||||
label_parts.append("long_clickable=True")
|
||||
if not label_parts:
|
||||
label_parts.append("(no visible text)")
|
||||
box_legend_lines.append(f" [{idx}] {', '.join(label_parts)}")
|
||||
@@ -791,7 +796,10 @@ class IntentResolver:
|
||||
f" - If looking for 'message input' or 'type message', do NOT select 'reactions' or emoji icons. Look for an empty text box or 'Message...'.\n"
|
||||
f"11. If the intent is 'feed post content' or 'post media content':\n"
|
||||
f" - Pick the largest box that contains the actual image or video, usually described as 'Photo', 'Video', or 'Carousel'.\n"
|
||||
f"12. If the exact control is NOT visible, return null. Do NOT guess.\n\n"
|
||||
f"12. DO NOT HALLUCINATE. If you are on the wrong screen (e.g. a 'New Post' creation screen when you want a profile), or if the exact target is simply NOT visible, you MUST return null. Returning a wrong box number will crash the bot.\n"
|
||||
f"13. EXTREME GUARD: NEVER pick an input field (class='EditText'), 'Send Message', or 'Reply to story' box unless the intent EXPLICITLY asks you to type or reply.\n"
|
||||
f"14. EXTREME GUARD: Do NOT pick items that are 'long_clickable=True' if your intent is just a simple navigation click, as this can trigger unwanted long-press context menus.\n"
|
||||
f"15. If the intent is 'tap post username', DO NOT pick random gallery folders like 'Recents' or 'Select album'. Return null.\n\n"
|
||||
f'Reply ONLY with a valid JSON object: {{"box": <number>}} or {{"box": null}}'
|
||||
)
|
||||
|
||||
@@ -919,7 +927,9 @@ class IntentResolver:
|
||||
" - 'profile tab' is usually the furthest right.\n"
|
||||
" - 'home tab' is the furthest left.\n"
|
||||
" - Do NOT select 'Go to <user>'s profile' or other header text.\n"
|
||||
"2. If none of the candidates clearly and safely match the intent, return null.\n\n"
|
||||
"2. EXTREME GUARD: NEVER pick an input field (EditText), 'Send Message', or 'Reply' unless explicitly asked to type.\n"
|
||||
"3. EXTREME GUARD: If you are on the wrong screen entirely (e.g. 'Select album' gallery) instead of a profile, return null.\n"
|
||||
"4. If none of the candidates clearly and safely match the intent, return null. DO NOT guess.\n\n"
|
||||
"Reply ONLY with a valid JSON object strictly matching this schema:\n"
|
||||
'{"selected_index": <integer or null>}\n'
|
||||
)
|
||||
|
||||
@@ -23,6 +23,7 @@ class ScreenType(Enum):
|
||||
COMMENTS = "comments"
|
||||
MODAL = "modal"
|
||||
FOREIGN_APP = "foreign_app"
|
||||
NOTIFICATIONS = "notifications"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
@@ -94,7 +95,14 @@ class ScreenIdentity:
|
||||
resource_ids.add(short_id)
|
||||
|
||||
# Track which tab is selected
|
||||
if selected and short_id in ("feed_tab", "search_tab", "clips_tab", "profile_tab", "direct_tab"):
|
||||
if selected and short_id in (
|
||||
"feed_tab",
|
||||
"search_tab",
|
||||
"clips_tab",
|
||||
"profile_tab",
|
||||
"direct_tab",
|
||||
"news_tab",
|
||||
):
|
||||
selected_tab = short_id
|
||||
|
||||
if text:
|
||||
@@ -180,14 +188,26 @@ class ScreenIdentity:
|
||||
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
|
||||
# EXCEPTION: If Qdrant has explicitly learned this screen as NORMAL (via LLM unlearning),
|
||||
# we skip the structural check to prevent false-positive infinite loops.
|
||||
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
|
||||
creation_flow_markers = (
|
||||
"quick_capture",
|
||||
"gallery_cancel_button",
|
||||
"creation_flow",
|
||||
"reel_camera",
|
||||
"gallery_grid_item_thumbnail",
|
||||
"camera_cancel_button",
|
||||
"next_button_textview",
|
||||
)
|
||||
browser_markers = ("ig_browser_text_title", "ig_browser_close_button", "ig_chrome_subsection")
|
||||
bottom_sheet_markers = ("bottom_sheet_container", "action_sheet_container")
|
||||
if not is_normal_override and any(marker in ids_str for marker in creation_flow_markers):
|
||||
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
|
||||
return ScreenType.MODAL
|
||||
if not is_normal_override and any(marker in ids_str for marker in browser_markers):
|
||||
logger.info("🛡️ [ScreenIdentity] In-App Browser detected → MODAL")
|
||||
return ScreenType.MODAL
|
||||
if not is_normal_override and any(marker in ids_str for marker in bottom_sheet_markers):
|
||||
logger.info("🛡️ [ScreenIdentity] Bottom Sheet Modal detected → MODAL")
|
||||
return ScreenType.MODAL
|
||||
|
||||
# Priority 2: Structural Heuristics (100% Deterministic)
|
||||
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
|
||||
@@ -239,6 +259,11 @@ class ScreenIdentity:
|
||||
if any(marker in ids for marker in STORY_MARKERS):
|
||||
return ScreenType.STORY_VIEW
|
||||
|
||||
# Notifications / Activity structural markers
|
||||
NOTIFICATIONS_MARKERS = ("row_newsfeed_text", "newsfeed_tab", "newsfeed_user_imageview")
|
||||
if any(marker in ids for marker in NOTIFICATIONS_MARKERS) or selected_tab == "news_tab":
|
||||
return ScreenType.NOTIFICATIONS
|
||||
|
||||
# DM thread detection — Structural markers (header and input fields)
|
||||
if "direct_thread_header" in ids or "direct_text_input" in ids or "message_composer_container" in ids:
|
||||
return ScreenType.DM_THREAD
|
||||
@@ -357,6 +382,7 @@ class ScreenIdentity:
|
||||
"clips_tab": "tap reels tab",
|
||||
"profile_tab": "tap profile tab",
|
||||
"direct_tab": "tap messages tab",
|
||||
"news_tab": "tap activity heart icon notifications",
|
||||
}
|
||||
for tab_id, action in tab_map.items():
|
||||
if tab_id in resource_ids:
|
||||
|
||||
@@ -15,6 +15,7 @@ class SpatialNode:
|
||||
content_desc: str = ""
|
||||
resource_id: str = ""
|
||||
clickable: bool = False
|
||||
long_clickable: bool = False
|
||||
scrollable: bool = False
|
||||
|
||||
# Spatial Properties
|
||||
@@ -162,6 +163,7 @@ class SpatialParser:
|
||||
resource_id=attrib.get("resource-id", "").strip(),
|
||||
bounds=(left, top, right, bottom),
|
||||
clickable=attrib.get("clickable", "false") == "true",
|
||||
long_clickable=attrib.get("long-clickable", "false") == "true",
|
||||
scrollable=attrib.get("scrollable", "false") == "true",
|
||||
)
|
||||
nodes_list.append(node)
|
||||
|
||||
@@ -1499,13 +1499,17 @@ class ContextMemoryDB(QdrantBase):
|
||||
|
||||
def is_allowed(self, intent: str, screen_type: str) -> bool:
|
||||
"""
|
||||
Exploration vs Exploitation:
|
||||
If we don't know (no entry or neutral confidence), allow it!
|
||||
Only block if we have learned it fails consistently (< 0.2).
|
||||
Exploration vs Exploitation with time-based decay:
|
||||
- If we don't know (no entry), allow it (exploration).
|
||||
- Only block if we have RECENTLY learned it fails (< 0.2 confidence AND < 1 hour old).
|
||||
- Old failures decay: entries older than 1 hour are auto-forgiven and deleted,
|
||||
preventing permanent poisoning of the action space.
|
||||
"""
|
||||
if not self.is_connected:
|
||||
return True # Fail-open for exploration
|
||||
|
||||
DECAY_THRESHOLD_SECONDS = 3600 # 1 hour: old failures are forgiven
|
||||
|
||||
key = self._get_key(intent, screen_type)
|
||||
try:
|
||||
point_id = self.generate_uuid(key)
|
||||
@@ -1513,11 +1517,25 @@ class ContextMemoryDB(QdrantBase):
|
||||
collection_name=self.collection_name, ids=[point_id], with_payload=True, with_vectors=False
|
||||
)
|
||||
if points:
|
||||
conf = points[0].payload.get("confidence", 0.5)
|
||||
payload = points[0].payload
|
||||
conf = payload.get("confidence", 0.5)
|
||||
updated_at = payload.get("updated_at", 0)
|
||||
age_seconds = time.time() - updated_at
|
||||
|
||||
# Time-based decay: old failures are forgiven
|
||||
if conf < 0.2 and age_seconds > DECAY_THRESHOLD_SECONDS:
|
||||
logger.info(
|
||||
f"🔄 [ContextMemory] Forgave old failure for '{intent}' on '{screen_type}' "
|
||||
f"(age: {age_seconds/60:.0f}min, confidence: {conf:.2f}). Allowing re-exploration."
|
||||
)
|
||||
# Delete the stale entry so the bot can re-learn
|
||||
self.delete_point(key)
|
||||
return True
|
||||
|
||||
if conf < 0.2:
|
||||
logger.warning(
|
||||
f"🛡️ [ContextMemory] Circuit Breaker: Blocked '{intent}' on '{screen_type}' "
|
||||
f"(learned confidence {conf:.2f} < 0.2)"
|
||||
f"(learned confidence {conf:.2f} < 0.2, age: {age_seconds/60:.0f}min)"
|
||||
)
|
||||
return False
|
||||
except Exception:
|
||||
|
||||
@@ -33,6 +33,7 @@ class ScreenTopology:
|
||||
"tap profile tab": ScreenType.OWN_PROFILE,
|
||||
"tap reels tab": ScreenType.REELS_FEED,
|
||||
"tap messages tab": ScreenType.DM_INBOX,
|
||||
"tap activity heart icon notifications": ScreenType.NOTIFICATIONS,
|
||||
"tap story ring avatar": ScreenType.STORY_VIEW,
|
||||
},
|
||||
ScreenType.EXPLORE_GRID: {
|
||||
@@ -84,6 +85,12 @@ class ScreenTopology:
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
# NOTE: 'press back' intentionally omitted — destination is non-deterministic
|
||||
},
|
||||
ScreenType.NOTIFICATIONS: {
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
"press back": ScreenType.HOME_FEED,
|
||||
"tap profile tab": ScreenType.OWN_PROFILE,
|
||||
"tap explore tab": ScreenType.EXPLORE_GRID,
|
||||
},
|
||||
ScreenType.UNKNOWN: {
|
||||
"tap home tab": ScreenType.HOME_FEED,
|
||||
},
|
||||
@@ -110,6 +117,7 @@ class ScreenTopology:
|
||||
"open user profile": ScreenType.OTHER_PROFILE,
|
||||
"open search": ScreenType.SEARCH_RESULTS,
|
||||
"view comments": ScreenType.COMMENTS,
|
||||
"open notifications": ScreenType.NOTIFICATIONS,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -98,22 +98,28 @@ class SituationEpisodeDB:
|
||||
|
||||
# Sort by confidence desc, prefer successful episodes
|
||||
best_positive = None
|
||||
failed_actions = set()
|
||||
failed_action_signatures = set()
|
||||
|
||||
for r in results:
|
||||
p = r.payload
|
||||
action_data = p.get("action", {})
|
||||
# Create a signature for the action: type + coordinates
|
||||
sig = f"{action_data.get('action_type')}|{action_data.get('x', 0)},{action_data.get('y', 0)}"
|
||||
|
||||
if not p.get("success", False):
|
||||
# Track failed actions so we don't repeat them
|
||||
failed_actions.add(p.get("action", {}).get("action_type", ""))
|
||||
# Track failed actions so we don't repeat the exact same interaction
|
||||
failed_action_signatures.add(sig)
|
||||
continue
|
||||
|
||||
conf = p.get("confidence", 0.0)
|
||||
if conf >= 0.3 and (best_positive is None or conf > best_positive.payload.get("confidence", 0)):
|
||||
best_positive = r
|
||||
|
||||
if best_positive:
|
||||
action_data = best_positive.payload.get("action", {})
|
||||
# Don't return an action type that has also failed before for this situation
|
||||
if action_data.get("action_type") not in failed_actions:
|
||||
sig = f"{action_data.get('action_type')}|{action_data.get('x', 0)},{action_data.get('y', 0)}"
|
||||
|
||||
if sig not in failed_action_signatures:
|
||||
logger.info(
|
||||
f"🧠 [SAE Recall] Instant memory hit! Situation matched with confidence "
|
||||
f"{best_positive.payload.get('confidence', 0):.2f}. Action: {action_data.get('reason', 'unknown')}"
|
||||
@@ -142,16 +148,21 @@ class SituationEpisodeDB:
|
||||
seed = f"{situation_signature}|{action.action_type}|{action.x},{action.y}"
|
||||
point_id = self._db.generate_uuid(seed)
|
||||
|
||||
# Retrieve existing confidence if any
|
||||
current_conf = 0.0
|
||||
has_existing = False
|
||||
try:
|
||||
points = self._db.client.retrieve(
|
||||
collection_name=self._db.collection_name, ids=[point_id], with_payload=True, with_vectors=False
|
||||
collection_name=self._db.collection_name,
|
||||
ids=[point_id],
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
if points:
|
||||
has_existing = True
|
||||
current_conf = points[0].payload.get("confidence", 0.0)
|
||||
except Exception:
|
||||
except Exception as e:
|
||||
logger.debug(f"SAE retrieval error (non-fatal): {e}")
|
||||
pass
|
||||
|
||||
if success:
|
||||
@@ -160,7 +171,8 @@ class SituationEpisodeDB:
|
||||
confidence = current_conf - 0.5 if has_existing else -0.5
|
||||
|
||||
if confidence < 0.1 and not success:
|
||||
self._db.client.delete(collection_name=self._db.collection_name, points_selector=[point_id])
|
||||
# Use safe wrapper to avoid crash if DB is down
|
||||
self._db.delete_point(seed)
|
||||
logger.info("🗑️ [SAE Learn] Action decayed below threshold. Deleted from memory.")
|
||||
return
|
||||
|
||||
@@ -339,14 +351,26 @@ class SituationalAwarenessEngine:
|
||||
return SituationType.OBSTACLE_SYSTEM
|
||||
|
||||
# ── Keyboard Detection (Fast Path) ──
|
||||
keyboard_pkgs = {
|
||||
"com.google.android.inputmethod.latin",
|
||||
"com.samsung.android.honeyboard",
|
||||
"com.sec.android.inputmethod",
|
||||
"com.touchtype.swiftkey",
|
||||
"com.apple.android.inputmethod",
|
||||
}
|
||||
if any(pkg in keyboard_pkgs for pkg in packages):
|
||||
# Instead of relying on brittle package names (which may be missing or custom),
|
||||
# an open keyboard is definitively proven if an EditText is focused,
|
||||
# or if unmistakable keyboard structural markers exist.
|
||||
is_actual_keyboard = False
|
||||
if 'focused="true"' in xml_dump and "EditText" in xml_dump:
|
||||
if re.search(r'class="[^"]*EditText"[^>]*focused="true"', xml_dump) or re.search(
|
||||
r'focused="true"[^>]*class="[^"]*EditText"', xml_dump
|
||||
):
|
||||
is_actual_keyboard = True
|
||||
|
||||
# Structural fallback: certain strings are exclusive to keyboards
|
||||
# We match these securely against content-desc or resource-id to avoid false positives from user post text.
|
||||
keyboard_markers = [
|
||||
r'content-desc="[^"]*(?:Symboltastatur|Switch input method|Leerzeichen|Spracheingabe verwenden)[^"]*"',
|
||||
r'resource-id="[^"]*input_method_nav_ime_switcher[^"]*"',
|
||||
]
|
||||
if not is_actual_keyboard and any(re.search(marker, xml_dump, re.IGNORECASE) for marker in keyboard_markers):
|
||||
is_actual_keyboard = True
|
||||
|
||||
if is_actual_keyboard:
|
||||
logger.info("📱 [SAE Perceive] On-screen Keyboard explicitly detected. Treating as obstacle.")
|
||||
return SituationType.OBSTACLE_KEYBOARD
|
||||
|
||||
@@ -472,6 +496,8 @@ class SituationalAwarenessEngine:
|
||||
"ig_browser_text_title", # In-App Browser Title
|
||||
"ig_browser_close_button", # In-App Browser Close Button
|
||||
"ig_chrome_subsection", # In-App Browser Chrome
|
||||
"bottom_sheet_container", # Bottom sheet modal (e.g., following options)
|
||||
"action_sheet_container", # Action sheet modal
|
||||
)
|
||||
if any(
|
||||
re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE)
|
||||
@@ -590,6 +616,72 @@ class SituationalAwarenessEngine:
|
||||
# 2. PLAN: AI-driven escape strategy
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
def _find_structural_dismiss_target(self, xml_dump: str) -> Optional[EscapeAction]:
|
||||
"""
|
||||
Structurally scan the raw XML for any clickable element that semantically
|
||||
represents a dismiss/close/cancel action. Returns None if nothing found.
|
||||
Zero hardcoding — all coordinates come from the XML itself.
|
||||
"""
|
||||
import re
|
||||
|
||||
# Patterns for dismiss-type buttons: text or content-desc containing dismiss keywords
|
||||
dismiss_keywords = (
|
||||
"cancel",
|
||||
"close",
|
||||
"dismiss",
|
||||
"not now",
|
||||
"nicht jetzt",
|
||||
"abbrechen",
|
||||
"schließen",
|
||||
"skip",
|
||||
"überspringen",
|
||||
"no thanks",
|
||||
"nein danke",
|
||||
"got it",
|
||||
"ok",
|
||||
"done",
|
||||
"fertig",
|
||||
)
|
||||
|
||||
# Extract each <node ...> or <node ... /> tag individually
|
||||
node_tags = re.findall(r"<node\b[^>]*>", xml_dump, re.IGNORECASE)
|
||||
|
||||
candidates = []
|
||||
for tag in node_tags:
|
||||
# Extract attributes independently — order doesn't matter
|
||||
text_m = re.search(r'text="([^"]*)"', tag)
|
||||
desc_m = re.search(r'content-desc="([^"]*)"', tag)
|
||||
click_m = re.search(r'clickable="([^"]*)"', tag)
|
||||
bounds_m = re.search(r'bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', tag)
|
||||
|
||||
text = (text_m.group(1) if text_m else "").strip().lower()
|
||||
desc = (desc_m.group(1) if desc_m else "").strip().lower()
|
||||
clickable = (click_m.group(1) if click_m else "").lower()
|
||||
|
||||
if clickable != "true" or not bounds_m:
|
||||
continue
|
||||
|
||||
label = text or desc
|
||||
if not label:
|
||||
continue
|
||||
|
||||
for kw in dismiss_keywords:
|
||||
if kw in label:
|
||||
x1, y1 = int(bounds_m.group(1)), int(bounds_m.group(2))
|
||||
x2, y2 = int(bounds_m.group(3)), int(bounds_m.group(4))
|
||||
cx, cy = (x1 + x2) // 2, (y1 + y2) // 2
|
||||
candidates.append((kw, cx, cy, label))
|
||||
break
|
||||
|
||||
if candidates:
|
||||
# Prefer exact match ("cancel" > "ok"), pick first match
|
||||
best = candidates[0]
|
||||
kw, cx, cy, label = best
|
||||
logger.info(f"🔍 [SAE Structural] Found dismiss target '{label}' at ({cx}, {cy})")
|
||||
return EscapeAction("click", x=cx, y=cy, reason=f"Structural dismiss: '{label}'")
|
||||
|
||||
return None
|
||||
|
||||
def _plan_escape_via_llm(
|
||||
self, xml_dump: str, compressed: str, situation_type: SituationType, failed_actions: set = None
|
||||
) -> Optional[EscapeAction]:
|
||||
@@ -604,30 +696,46 @@ class SituationalAwarenessEngine:
|
||||
model = getattr(args, "ai_telepathic_model")
|
||||
url = getattr(args, "ai_telepathic_url")
|
||||
|
||||
# Build failure context so the LLM knows what NOT to repeat
|
||||
failure_block = ""
|
||||
temperature = 0.1
|
||||
if failed_actions:
|
||||
failure_block = (
|
||||
"\n\nCRITICAL — The following actions have ALREADY BEEN TRIED and FAILED. "
|
||||
"You MUST NOT propose any of these again:\n"
|
||||
)
|
||||
for fa in failed_actions:
|
||||
failure_block += f" - {fa}\n"
|
||||
failure_block += (
|
||||
"\nYou MUST pick a DIFFERENT element or a DIFFERENT action type. "
|
||||
"Study the XML carefully for buttons you have NOT tried yet.\n"
|
||||
)
|
||||
# Increase temperature to force diversity when the LLM is being stubborn
|
||||
temperature = min(0.7, 0.1 + 0.2 * len(failed_actions))
|
||||
|
||||
system_prompt = (
|
||||
"You are an Android UI navigation agent. Your job is to escape obstacles "
|
||||
"(dialogs, modals, foreign apps, system popups) and return to Instagram. "
|
||||
"Analyze the screen content (Screenshot AND XML) and return a JSON escape action.\n\n"
|
||||
"You are the Navigation Engine for an autonomous agent.\n"
|
||||
"An obstacle has been detected that prevents normal operations.\n"
|
||||
"Your goal is to DISMISS or CLEAR this obstacle to return to the app's main workflow.\n\n"
|
||||
f"Current Obstacle Type: {situation_type.value}\n\n"
|
||||
"Rules:\n"
|
||||
"- If you see a dismiss/close/cancel/skip/not now button, click it\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 the Situation type is obstacle_system, you MUST look for 'Deny', 'Don't allow', or 'Cancel' and click it. \n"
|
||||
" NEVER click 'Allow', 'OK', or 'Confirm' on system permissions.\n"
|
||||
" If no negative action button exists, action must be 'back'\n"
|
||||
"- If the screen shows an In-App Browser, WebView, website, or Advertisement, it is an obstacle. Action must be 'back' or click the close button.\n"
|
||||
"- If there is NO obstacle and the screen is a normal Instagram view (false positive), action must be 'false_positive'\n"
|
||||
"- If nothing else works, suggest 'app_start' to force-reopen Instagram\n"
|
||||
"- NEVER click 'OK'/'Confirm'/'Accept' on surveys or prompts\n"
|
||||
"- When you choose to click, you MUST use the EXACT coordinates provided in `center=(x,y)` for that element in the XML\n"
|
||||
'- Return ONLY valid JSON: {"action": "click"|"back"|"app_start"|"unlock"|"kill_foreign_apps"|"false_positive", "x": N, "y": N, "reason": "..."}'
|
||||
"- If the Situation type is obstacle_modal, find and click a 'Cancel', 'Close', 'Dismiss', 'Not Now', "
|
||||
"or 'X' button. Look carefully at ALL elements in the XML.\n"
|
||||
"- If it is a system dialog, click 'Allow', 'OK' or 'Dismiss'.\n"
|
||||
"- If the keyboard is blocking, your only action is 'back'.\n"
|
||||
'- When you choose to click, you MUST use the EXACT coordinates from bounds="[x1,y1][x2,y2]" '
|
||||
"in the XML. Calculate center as ((x1+x2)/2, (y1+y2)/2). Do NOT invent coordinates.\n"
|
||||
"- If no clickable dismiss element exists in the XML, use 'back'.\n"
|
||||
"- Respond ONLY with valid JSON: "
|
||||
'{"action": "click" | "back" | "app_start", "x": int, "y": int, "reason": "string"}'
|
||||
f"{failure_block}"
|
||||
)
|
||||
|
||||
user_prompt = f"Situation type: {situation_type.value}\n\n" f"Screen content:\n{compressed}\n\n"
|
||||
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."
|
||||
user_prompt = (
|
||||
f"Situation type: {situation_type.value}\n\n"
|
||||
f"Screen XML:\n{compressed}\n\n"
|
||||
"What action should I take to clear this obstacle and return to Instagram? Return JSON only."
|
||||
)
|
||||
|
||||
try:
|
||||
screenshot_b64 = getattr(self.device, "get_screenshot_b64", lambda: None)()
|
||||
@@ -638,7 +746,7 @@ class SituationalAwarenessEngine:
|
||||
user_prompt=user_prompt,
|
||||
system_prompt=system_prompt,
|
||||
images_b64=[screenshot_b64] if screenshot_b64 else None,
|
||||
temperature=0.0,
|
||||
temperature=temperature,
|
||||
)
|
||||
if resp:
|
||||
import json
|
||||
@@ -646,7 +754,6 @@ class SituationalAwarenessEngine:
|
||||
try:
|
||||
data = json.loads(resp)
|
||||
except json.JSONDecodeError:
|
||||
# Try extracting JSON via regex if LLM was chatty
|
||||
import re
|
||||
|
||||
match = re.search(r"\{.*\}", resp, re.DOTALL)
|
||||
@@ -655,15 +762,42 @@ class SituationalAwarenessEngine:
|
||||
else:
|
||||
raise ValueError(f"Could not parse JSON from: {resp}")
|
||||
|
||||
action_type = data.get("action", "back")
|
||||
x = int(data.get("x", 0))
|
||||
y = int(data.get("y", 0))
|
||||
reason = data.get("reason", "LLM-planned escape")
|
||||
action_key = f"{action_type}:{x},{y}"
|
||||
|
||||
# If LLM stubbornly repeats a failed action, try structural scan as autonomous fallback
|
||||
if failed_actions and action_key in failed_actions:
|
||||
logger.warning(
|
||||
f"🧠 [SAE] LLM repeated failed action ({action_key}). "
|
||||
"Falling back to structural dismiss scan."
|
||||
)
|
||||
structural = self._find_structural_dismiss_target(xml_dump)
|
||||
if structural:
|
||||
structural_key = f"{structural.action_type}:{structural.x},{structural.y}"
|
||||
if structural_key not in failed_actions:
|
||||
return structural
|
||||
# If structural also exhausted, escalate to back
|
||||
if "back:0,0" not in (failed_actions or set()):
|
||||
return EscapeAction("back", reason="Autonomous fallback: structural + LLM both exhausted")
|
||||
return EscapeAction("app_start", reason="Autonomous escalation: all escape strategies exhausted")
|
||||
|
||||
return EscapeAction(
|
||||
action_type=data.get("action", "back"),
|
||||
x=int(data.get("x", 0)),
|
||||
y=int(data.get("y", 0)),
|
||||
reason=data.get("reason", "LLM-planned escape"),
|
||||
action_type=action_type,
|
||||
x=x,
|
||||
y=y,
|
||||
reason=reason,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"🧠 [SAE] LLM escape planning failed: {e}")
|
||||
|
||||
# Last resort: try structural scan before giving up
|
||||
structural = self._find_structural_dismiss_target(xml_dump)
|
||||
if structural:
|
||||
return structural
|
||||
|
||||
return EscapeAction("back", reason="LLM planning failed, defaulting to BACK")
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
@@ -85,6 +85,18 @@ class TelepathicEngine:
|
||||
filtered_candidates.append(c)
|
||||
candidates = filtered_candidates
|
||||
|
||||
# 2.5 Structural Sanity Check (Filter out traps before VLM sees them)
|
||||
safe_candidates = []
|
||||
for c in candidates:
|
||||
if self._structural_sanity_check(c, intent_description):
|
||||
safe_candidates.append(c)
|
||||
|
||||
if not safe_candidates:
|
||||
logger.warning(f"All candidates failed structural sanity check for intent: '{intent_description}'")
|
||||
return None
|
||||
|
||||
candidates = safe_candidates
|
||||
|
||||
# 3. Resolve intent against candidates
|
||||
best_node = self._resolver.resolve(intent_description, candidates, device=device)
|
||||
|
||||
@@ -224,13 +236,35 @@ class TelepathicEngine:
|
||||
def _is_instagram_context(self, xml_string: str) -> bool:
|
||||
return "com.instagram.android" in xml_string
|
||||
|
||||
def _structural_sanity_check(self, node: dict, intent_description: str, screen_height: int = 2400) -> bool:
|
||||
def _structural_sanity_check(self, node: SpatialNode, intent_description: str, screen_height: int = 2400) -> bool:
|
||||
"""
|
||||
Structural guard to ensure nodes are in valid locations for their intents.
|
||||
"""
|
||||
intent = intent_description.lower()
|
||||
y = node.get("y", 0)
|
||||
semantic = (node.get("semantic_string", "") or "").lower()
|
||||
y = node.center_y
|
||||
semantic = ((node.text or "") + " " + (node.content_desc or "") + " " + (node.resource_id or "")).lower()
|
||||
|
||||
# 0. EXTREME GUARD: NEVER pick an input field (EditText) or reply box unless explicitly asked to type/reply.
|
||||
if "edittext" in (node.class_name or "").lower() or "reply" in semantic:
|
||||
if "type" not in intent and "reply" not in intent and "message" not in intent and "search" not in intent:
|
||||
return False
|
||||
|
||||
# 0.5 GLOBAL STRUCTURAL TRAP GUARD: Prevent accidental clicks on dangerous/irrelevant UI traps
|
||||
# ZERO-TRUST: NO LOCALIZED STRINGS.
|
||||
res_id = (node.resource_id or "").lower()
|
||||
|
||||
# Avoid clicking on audio/trending/camera/creation elements unless explicitly requested
|
||||
if any(
|
||||
trap in res_id
|
||||
for trap in ["album_art", "use_in_camera", "trending", "audio", "creation", "camera", "gallery"]
|
||||
):
|
||||
if not any(word in intent for word in ["audio", "trending", "camera", "create", "gallery"]):
|
||||
return False
|
||||
|
||||
# Avoid clicking share, save, menu, options unless explicitly requested
|
||||
if any(trap in res_id for trap in ["share", "save", "options", "menu", "add"]):
|
||||
if not any(word in intent for word in ["share", "save", "options", "menu", "add"]):
|
||||
return False
|
||||
|
||||
# 1. Post Username Guard
|
||||
if "post username" in intent or "author username" in intent:
|
||||
@@ -277,7 +311,7 @@ class TelepathicEngine:
|
||||
|
||||
# 6. Block massive layout containers UNLESS specifically looking for feed/post
|
||||
MAX_CONTAINER_AREA = 500000
|
||||
area = node.get("area", 0)
|
||||
area = node.area
|
||||
|
||||
is_feed_or_post = "feed" in intent or "post" in intent
|
||||
is_grid_item = "grid" in intent or "list" in intent
|
||||
|
||||
@@ -21,10 +21,13 @@ else
|
||||
# Heuristic: Try to find a matching unit test
|
||||
test_file="tests/unit/test_${filename}"
|
||||
core_test_file="tests/core/test_${filename}"
|
||||
tdd_test_file="tests/tdd/test_${filename}"
|
||||
if [ -f "$test_file" ]; then
|
||||
TEST_TARGETS="$TEST_TARGETS $test_file"
|
||||
elif [ -f "$core_test_file" ]; then
|
||||
TEST_TARGETS="$TEST_TARGETS $core_test_file"
|
||||
elif [ -f "$tdd_test_file" ]; then
|
||||
TEST_TARGETS="$TEST_TARGETS $tdd_test_file"
|
||||
else
|
||||
# Try to find matching e2e tests by searching for each word in the module name
|
||||
module_name="${filename%.py}"
|
||||
|
||||
213
tests/tdd/test_close_friends_and_memory.py
Normal file
213
tests/tdd/test_close_friends_and_memory.py
Normal file
@@ -0,0 +1,213 @@
|
||||
"""
|
||||
TDD: Close Friends Guard & ContextMemory Circuit Breaker
|
||||
|
||||
1. Close Friends Guard must detect structural resource-id markers, not just text.
|
||||
2. ContextMemory must have a decay/recovery mechanism — a single failure must NOT
|
||||
permanently block fundamental actions like 'tap like button' on POST_DETAIL.
|
||||
3. The circuit breaker threshold must be LOWER for core actions.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.behaviors import BehaviorContext, BehaviorResult
|
||||
from GramAddict.core.behaviors.close_friends_guard import CloseFriendsGuardPlugin
|
||||
|
||||
|
||||
# ── Fixtures ──
|
||||
|
||||
|
||||
class DummyDeviceV2:
|
||||
def __init__(self):
|
||||
self.info = {"screenOn": True}
|
||||
|
||||
|
||||
class DummyDevice:
|
||||
def __init__(self):
|
||||
self.deviceV2 = DummyDeviceV2()
|
||||
self.app_id = "com.instagram.android"
|
||||
self.pressed = []
|
||||
self.clicks = []
|
||||
self._display_info = {"width": 1080, "height": 2424}
|
||||
|
||||
def press(self, key):
|
||||
self.pressed.append(key)
|
||||
|
||||
def click(self, x, y):
|
||||
self.clicks.append((x, y))
|
||||
|
||||
|
||||
# Instagram shows close friends via structural markers, not text
|
||||
FEED_WITH_CLOSE_FRIEND_POST = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/coordinator_root_layout"
|
||||
class="android.view.ViewGroup" clickable="false" bounds="[0,0][1080,2424]">
|
||||
<node index="0" resource-id="com.instagram.android:id/friendly_bubbles_component"
|
||||
class="android.view.ViewGroup" clickable="false" bounds="[42,1761][927,1925]" />
|
||||
<node index="1" text="Friends" content-desc="Friends"
|
||||
class="android.widget.LinearLayout" clickable="true" bounds="[491,213][858,302]" />
|
||||
<node index="2" text="" resource-id="com.instagram.android:id/row_feed_photo_profile_name"
|
||||
class="android.widget.TextView" clickable="true" bounds="[100,300][400,350]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
FEED_NORMAL_POST = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/coordinator_root_layout"
|
||||
class="android.view.ViewGroup" clickable="false" bounds="[0,0][1080,2424]">
|
||||
<node index="0" text="" resource-id="com.instagram.android:id/row_feed_photo_profile_name"
|
||||
class="android.widget.TextView" clickable="true" bounds="[100,300][400,350]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
PROFILE_WITH_CLOSE_FRIEND_BADGE = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/coordinator_root_layout"
|
||||
class="android.view.ViewGroup" clickable="false" bounds="[0,0][1080,2424]">
|
||||
<node index="0" resource-id="com.instagram.android:id/profile_header_close_friend"
|
||||
class="android.view.View" clickable="false" bounds="[0,0][100,100]" />
|
||||
<node index="1" text="username" resource-id="com.instagram.android:id/action_bar_title"
|
||||
class="android.widget.TextView" clickable="false" bounds="[200,50][500,100]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
REELS_WITH_GREEN_STAR = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/clips_viewer_view_pager"
|
||||
class="android.view.ViewGroup" clickable="false" bounds="[0,0][1080,2424]">
|
||||
<node index="0" resource-id="com.instagram.android:id/close_friends_badge"
|
||||
class="android.view.View" clickable="false" bounds="[50,300][90,340]"
|
||||
content-desc="Close friends" />
|
||||
<node index="1" text="username" class="android.widget.TextView"
|
||||
clickable="true" bounds="[100,310][400,350]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
|
||||
# ── Tests: Close Friends Guard ──
|
||||
|
||||
|
||||
class TestCloseFriendsGuard:
|
||||
"""The guard must detect close friends via STRUCTURAL markers, not just text."""
|
||||
|
||||
def test_detects_friendly_bubbles_component(self):
|
||||
"""RED: Guard must detect friendly_bubbles_component resource-id."""
|
||||
plugin = CloseFriendsGuardPlugin()
|
||||
ctx = BehaviorContext(
|
||||
device=DummyDevice(),
|
||||
session_state=type("S", (), {"job_target": "test"})(),
|
||||
shared_state={},
|
||||
context_xml=FEED_WITH_CLOSE_FRIEND_POST,
|
||||
configs=None,
|
||||
cognitive_stack=None,
|
||||
)
|
||||
assert plugin.can_activate(ctx), (
|
||||
"CloseFriendsGuard must detect 'friendly_bubbles_component' as a close friend indicator"
|
||||
)
|
||||
|
||||
def test_does_not_fire_on_normal_post(self):
|
||||
"""Guard must NOT fire on normal posts without close friend markers."""
|
||||
plugin = CloseFriendsGuardPlugin()
|
||||
ctx = BehaviorContext(
|
||||
device=DummyDevice(),
|
||||
session_state=type("S", (), {"job_target": "test"})(),
|
||||
shared_state={},
|
||||
context_xml=FEED_NORMAL_POST,
|
||||
configs=None,
|
||||
cognitive_stack=None,
|
||||
)
|
||||
assert not plugin.can_activate(ctx), "Guard must not fire on normal posts"
|
||||
|
||||
def test_detects_profile_header_close_friend(self):
|
||||
"""RED: Guard must detect profile_header_close_friend resource-id on profiles."""
|
||||
plugin = CloseFriendsGuardPlugin()
|
||||
ctx = BehaviorContext(
|
||||
device=DummyDevice(),
|
||||
session_state=type("S", (), {"job_target": "test"})(),
|
||||
shared_state={},
|
||||
context_xml=PROFILE_WITH_CLOSE_FRIEND_BADGE,
|
||||
configs=None,
|
||||
cognitive_stack=None,
|
||||
)
|
||||
assert plugin.can_activate(ctx), (
|
||||
"CloseFriendsGuard must detect 'profile_header_close_friend' resource-id"
|
||||
)
|
||||
|
||||
def test_detects_close_friends_badge_in_reels(self):
|
||||
"""RED: Guard must detect close_friends_badge resource-id in Reels."""
|
||||
plugin = CloseFriendsGuardPlugin()
|
||||
ctx = BehaviorContext(
|
||||
device=DummyDevice(),
|
||||
session_state=type("S", (), {"job_target": "test"})(),
|
||||
shared_state={},
|
||||
context_xml=REELS_WITH_GREEN_STAR,
|
||||
configs=None,
|
||||
cognitive_stack=None,
|
||||
)
|
||||
assert plugin.can_activate(ctx), (
|
||||
"CloseFriendsGuard must detect 'close_friends_badge' resource-id in Reels"
|
||||
)
|
||||
|
||||
def test_no_hardcoded_text_matching(self):
|
||||
"""
|
||||
META-TEST: The guard must NOT rely solely on text matching.
|
||||
It must use structural resource-id markers.
|
||||
"""
|
||||
import inspect
|
||||
source = inspect.getsource(CloseFriendsGuardPlugin.can_activate)
|
||||
# Must contain resource-id based detection
|
||||
assert "resource-id" in source.lower() or "resource_id" in source.lower() or \
|
||||
"friendly_bubbles" in source or "close_friends_badge" in source or \
|
||||
"profile_header_close_friend" in source, (
|
||||
"can_activate must use structural resource-id markers, not just text matching"
|
||||
)
|
||||
|
||||
|
||||
# ── Tests: ContextMemory Circuit Breaker ──
|
||||
|
||||
|
||||
class TestContextMemoryCircuitBreaker:
|
||||
"""
|
||||
The circuit breaker must NOT permanently block core actions.
|
||||
A single failure must decay over time, allowing re-exploration.
|
||||
"""
|
||||
|
||||
def test_circuit_breaker_has_time_decay(self):
|
||||
"""
|
||||
META-TEST: ContextMemory.is_allowed must consider time since last update.
|
||||
Old failures (> 1 hour) must be forgiven to allow re-exploration.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.qdrant_memory import ContextMemoryDB
|
||||
|
||||
source = inspect.getsource(ContextMemoryDB.is_allowed)
|
||||
# Must reference time-based decay or updated_at
|
||||
assert "updated_at" in source or "time" in source or "decay" in source or "age" in source, (
|
||||
"is_allowed must implement time-based decay. A single failure should not "
|
||||
"permanently block an action. Old failures must be forgiven."
|
||||
)
|
||||
|
||||
def test_core_actions_have_lower_block_threshold(self):
|
||||
"""
|
||||
META-TEST: Core navigation actions like 'tap like button' or 'tap home tab'
|
||||
must be harder to permanently block than obscure actions.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.qdrant_memory import ContextMemoryDB
|
||||
|
||||
source = inspect.getsource(ContextMemoryDB.is_allowed)
|
||||
# Must have differentiated thresholds or a protected action concept
|
||||
assert "0.2" not in source or "core" in source.lower() or "protected" in source.lower() or \
|
||||
"updated_at" in source, (
|
||||
"The 0.2 hard threshold is too aggressive. Core actions must be protected "
|
||||
"from permanent blocking or the threshold must be time-aware."
|
||||
)
|
||||
@@ -141,6 +141,7 @@ class TestScreenTopologyBackTransitions:
|
||||
ScreenType.FOLLOW_LIST,
|
||||
ScreenType.STORY_VIEW,
|
||||
ScreenType.COMMENTS,
|
||||
ScreenType.NOTIFICATIONS,
|
||||
):
|
||||
# These are "leaf" screens with a deterministic parent — OK to keep
|
||||
continue
|
||||
|
||||
124
tests/tdd/test_follow_and_vlm.py
Normal file
124
tests/tdd/test_follow_and_vlm.py
Normal file
@@ -0,0 +1,124 @@
|
||||
"""
|
||||
TDD: Follow Plugin Safety Guard + VLM Verification Resilience
|
||||
|
||||
1. Follow plugin must REFUSE to click "Following" buttons (already followed).
|
||||
Only "Follow" buttons (not yet followed) are valid targets.
|
||||
2. VLM verification must NOT be the sole source of truth. When VLM says "false"
|
||||
but structural delta shows significant UI change, structural evidence wins.
|
||||
3. VLM false negatives must NOT poison the memory system.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── Follow Plugin Tests ──
|
||||
|
||||
|
||||
class TestFollowPluginSafetyGuard:
|
||||
"""The follow plugin must distinguish 'Follow' from 'Following' buttons."""
|
||||
|
||||
def test_follow_plugin_intent_only_targets_follow_not_following(self):
|
||||
"""
|
||||
META-TEST: The follow plugin's intent must specifically target ONLY
|
||||
the "Follow" state (unfollowed users), not the "Following" state.
|
||||
Clicking "Following" opens an unfollow/favorites bottom sheet.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
source = inspect.getsource(FollowPlugin.execute)
|
||||
|
||||
# The intent must NOT be "tap 'Follow' or 'Following' button"
|
||||
# because that would match both states
|
||||
assert "'Follow' or 'Following'" not in source, (
|
||||
"Follow plugin must NOT use an intent that matches both 'Follow' AND 'Following'. "
|
||||
"Clicking 'Following' opens a dangerous bottom sheet (Unfollow/Add to Favorites/Close Friends). "
|
||||
"The intent must ONLY target the 'Follow' state."
|
||||
)
|
||||
|
||||
def test_follow_plugin_has_already_following_guard(self):
|
||||
"""
|
||||
RED: Follow plugin must check if the button already says 'Following'
|
||||
and SKIP if so. This prevents opening the dangerous bottom sheet.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
source = inspect.getsource(FollowPlugin.execute)
|
||||
source_lower = source.lower()
|
||||
|
||||
# Must have some guard that checks for "following" state
|
||||
has_following_guard = (
|
||||
"following" in source_lower and
|
||||
("skip" in source_lower or "already" in source_lower or "abort" in source_lower or "return" in source_lower)
|
||||
)
|
||||
assert has_following_guard, (
|
||||
"Follow plugin must guard against clicking 'Following' buttons. "
|
||||
"It must check the button text and SKIP if the user is already followed."
|
||||
)
|
||||
|
||||
|
||||
# ── VLM Verification Resilience Tests ──
|
||||
|
||||
|
||||
class TestVLMVerificationResilience:
|
||||
"""
|
||||
VLM verification must NOT be the sole authority on action success.
|
||||
Structural evidence must be able to OVERRIDE a VLM false negative.
|
||||
"""
|
||||
|
||||
def test_vlm_false_does_not_shortcircuit_when_structural_delta_exists(self):
|
||||
"""
|
||||
META-TEST: When VLM says false but there IS significant structural
|
||||
delta, the verification must NOT blindly return False.
|
||||
Structural evidence must be considered.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.perception.action_memory import ActionMemory
|
||||
|
||||
source = inspect.getsource(ActionMemory.verify_success)
|
||||
|
||||
# Find the VLM false block — it must NOT just "return False"
|
||||
# It should fallthrough to structural verification
|
||||
lines = source.split("\n")
|
||||
vlm_false_section = False
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
if "decision is False" in stripped or "decision == False" in stripped:
|
||||
vlm_false_section = True
|
||||
continue
|
||||
if vlm_false_section:
|
||||
# The very next meaningful line after detecting VLM false
|
||||
# must NOT be a hard "return False"
|
||||
if stripped and not stripped.startswith("#") and not stripped.startswith('"""'):
|
||||
assert "return False" not in stripped, (
|
||||
f"Line {i}: VLM returning False must NOT shortcircuit to 'return False'. "
|
||||
"It must fallthrough to structural delta verification. "
|
||||
"A small local VLM (7B) is unreliable and systematically poisons "
|
||||
"the memory by saying everything failed."
|
||||
)
|
||||
break # Only check the first meaningful line after
|
||||
|
||||
def test_verify_success_has_vlm_structural_reconciliation(self):
|
||||
"""
|
||||
META-TEST: verify_success must have logic that reconciles VLM verdicts
|
||||
with structural delta evidence. If structural delta shows significant
|
||||
change (>50 chars), VLM false should be overridden.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.perception.action_memory import ActionMemory
|
||||
|
||||
source = inspect.getsource(ActionMemory.verify_success)
|
||||
|
||||
# Must have some concept of VLM being unreliable or structural override
|
||||
has_reconciliation = (
|
||||
"structural" in source.lower() and
|
||||
("override" in source.lower() or "fallthrough" in source.lower() or
|
||||
"vlm_verdict" in source.lower() or "vlm_decision" in source.lower() or
|
||||
"trust" in source.lower())
|
||||
)
|
||||
assert has_reconciliation, (
|
||||
"verify_success must reconcile VLM verdicts with structural evidence. "
|
||||
"A VLM false-negative when structural delta shows significant change "
|
||||
"should not be treated as a hard failure."
|
||||
)
|
||||
39
tests/tdd/test_gallery_trap_guard.py
Normal file
39
tests/tdd/test_gallery_trap_guard.py
Normal file
@@ -0,0 +1,39 @@
|
||||
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
|
||||
|
||||
|
||||
def test_gallery_screen_is_modal():
|
||||
"""
|
||||
Test that the Gallery/New Post screen is correctly identified as a MODAL,
|
||||
preventing it from being hallucinated as POST_DETAIL or HOME_FEED.
|
||||
"""
|
||||
xml_dump = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" text="" resource-id="" class="android.widget.FrameLayout" package="com.instagram.android" bounds="[0,0][1080,2400]">
|
||||
<node index="0" text="New post" resource-id="" class="android.widget.TextView" bounds="[0,0][100,100]" />
|
||||
<node index="1" text="" resource-id="com.instagram.android:id/gallery_grid_item_thumbnail" class="android.widget.Button" content-desc="Selected media number 1 Photo thumbnail created on 3 May 2026 10:30" bounds="[0,100][500,600]" />
|
||||
<node index="2" text="" resource-id="com.instagram.android:id/next_button_textview" class="android.widget.Button" content-desc="Next" bounds="[900,0][1080,100]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
identity = ScreenIdentity("bot_user")
|
||||
result = identity.identify(xml_dump)
|
||||
|
||||
assert result["screen_type"] == ScreenType.MODAL
|
||||
|
||||
|
||||
def test_camera_screen_is_modal():
|
||||
"""
|
||||
Test that the Camera creation screen is correctly identified as a MODAL.
|
||||
"""
|
||||
xml_dump = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" text="" resource-id="" class="android.widget.FrameLayout" package="com.instagram.android" bounds="[0,0][1080,2400]">
|
||||
<node index="0" text="" resource-id="com.instagram.android:id/camera_cancel_button" class="android.widget.ImageView" bounds="[0,0][100,100]" />
|
||||
<node index="1" text="" resource-id="com.instagram.android:id/quick_capture" class="android.widget.FrameLayout" bounds="[0,100][1080,2400]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
identity = ScreenIdentity("bot_user")
|
||||
result = identity.identify(xml_dump)
|
||||
|
||||
assert result["screen_type"] == ScreenType.MODAL
|
||||
62
tests/tdd/test_navigation_trap_recovery.py
Normal file
62
tests/tdd/test_navigation_trap_recovery.py
Normal file
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
TDD: Navigation Trap Recovery & UNKNOWN Screen Handling
|
||||
|
||||
1. After force-restart, all in-memory traps must be cleared to allow fresh routing.
|
||||
2. Traps on UNKNOWN screens must NEVER persist — UNKNOWN is a catch-all,
|
||||
permanent traps here block ALL unidentified screens.
|
||||
3. Trap system must have time-decay — old traps expire after a threshold.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestNavigationTrapRecovery:
|
||||
"""Trap system must not create permanent dead-ends."""
|
||||
|
||||
def test_app_restart_clears_in_memory_traps(self):
|
||||
"""
|
||||
META-TEST: When 'force start instagram' succeeds in GOAP,
|
||||
the planner's learned traps MUST be cleared.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.goap import GoalExecutor
|
||||
|
||||
source = inspect.getsource(GoalExecutor.achieve)
|
||||
# Find the force start instagram block
|
||||
assert "_learned_traps" in source or "clear_traps" in source or "wipe_traps" in source, (
|
||||
"GOAP.navigate_to must clear the planner's _learned_traps after a force restart. "
|
||||
"Currently only action_failures and explored_nav_actions are cleared, "
|
||||
"leaving traps active — which causes immediate re-trapping on restart."
|
||||
)
|
||||
|
||||
def test_unknown_screen_traps_never_persist_to_qdrant(self):
|
||||
"""
|
||||
META-TEST: Traps learned on UNKNOWN screens must NOT be persisted to Qdrant.
|
||||
UNKNOWN is a catch-all — persisting traps here blocks ALL unidentified screens forever.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.navigation.knowledge import NavigationKnowledge
|
||||
|
||||
source = inspect.getsource(NavigationKnowledge.learn_trap)
|
||||
source_lower = source.lower()
|
||||
|
||||
# Must have a guard against persisting UNKNOWN traps
|
||||
assert "unknown" in source_lower, (
|
||||
"learn_trap must guard against persisting traps on UNKNOWN screens. "
|
||||
"UNKNOWN is a catch-all — permanent traps here block ALL unidentified screens."
|
||||
)
|
||||
|
||||
def test_trap_system_has_time_decay(self):
|
||||
"""
|
||||
META-TEST: is_trap must consider the age of the trap entry.
|
||||
Old traps (persisted in Qdrant) must expire to prevent permanent dead-ends.
|
||||
"""
|
||||
import inspect
|
||||
from GramAddict.core.navigation.knowledge import NavigationKnowledge
|
||||
|
||||
source = inspect.getsource(NavigationKnowledge.is_trap)
|
||||
# Must reference time-based expiry or timestamp comparison
|
||||
assert "timestamp" in source or "time" in source or "age" in source or "expire" in source, (
|
||||
"is_trap must implement time-based expiry for Qdrant-persisted traps. "
|
||||
"Permanent traps cause permanent dead-ends."
|
||||
)
|
||||
22
tests/tdd/test_notifications_screen.py
Normal file
22
tests/tdd/test_notifications_screen.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
|
||||
|
||||
|
||||
def test_notifications_screen_identification():
|
||||
"""
|
||||
Test that the Notifications/Activity screen is correctly identified,
|
||||
preventing it from being hallucinated as POST_DETAIL.
|
||||
"""
|
||||
xml_dump = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" text="" resource-id="" class="android.widget.FrameLayout" package="com.instagram.android" bounds="[0,0][1080,2400]">
|
||||
<node index="0" text="Notifications" resource-id="com.instagram.android:id/action_bar_title" class="android.widget.TextView" bounds="[0,0][1080,100]" />
|
||||
<node index="1" text="User liked your photo." resource-id="com.instagram.android:id/row_newsfeed_text" class="android.widget.TextView" bounds="[100,100][1000,200]" />
|
||||
<node index="2" text="" resource-id="com.instagram.android:id/newsfeed_tab" selected="true" class="android.widget.FrameLayout" bounds="[600,2300][800,2400]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
identity = ScreenIdentity("bot_user")
|
||||
result = identity.identify(xml_dump)
|
||||
|
||||
assert result["screen_type"] == ScreenType.NOTIFICATIONS
|
||||
assert "tap home tab" in result["available_actions"] or "press back" in result["available_actions"]
|
||||
@@ -28,7 +28,8 @@ def test_sae_detects_obstacle_keyboard():
|
||||
xml_with_keyboard = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android" resource-id="" text="" content-desc="" />
|
||||
<node index="1" package="com.google.android.inputmethod.latin" resource-id="com.google.android.inputmethod.latin:id/keyboard_view" text="" content-desc="Keyboard" />
|
||||
<node index="1" class="android.widget.EditText" focused="true" />
|
||||
<node index="2" package="com.google.android.inputmethod.latin" resource-id="com.google.android.inputmethod.latin:id/keyboard_view" text="" content-desc="Keyboard" />
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
@@ -45,7 +46,8 @@ def test_obstacle_guard_dismisses_keyboard():
|
||||
xml_with_keyboard = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android" resource-id="" text="" content-desc="" />
|
||||
<node index="1" package="com.google.android.inputmethod.latin" resource-id="com.google.android.inputmethod.latin:id/keyboard_view" text="" content-desc="Keyboard" />
|
||||
<node index="1" class="android.widget.EditText" focused="true" />
|
||||
<node index="2" package="com.google.android.inputmethod.latin" resource-id="com.google.android.inputmethod.latin:id/keyboard_view" text="" content-desc="Keyboard" />
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
|
||||
206
tests/tdd/test_situational_awareness.py
Normal file
206
tests/tdd/test_situational_awareness.py
Normal file
@@ -0,0 +1,206 @@
|
||||
"""
|
||||
TDD: Structural Dismiss Target Discovery
|
||||
|
||||
Proves that the SAE finds dismiss/close buttons by scanning
|
||||
the XML structurally — zero hardcoded coordinates.
|
||||
All coordinates come from bounds attributes in the XML itself.
|
||||
"""
|
||||
|
||||
from GramAddict.core.situational_awareness import (
|
||||
SituationalAwarenessEngine,
|
||||
)
|
||||
|
||||
|
||||
class DummyDeviceV2:
|
||||
def __init__(self):
|
||||
self.info = {"screenOn": True}
|
||||
|
||||
|
||||
class DummyDevice:
|
||||
def __init__(self):
|
||||
self.deviceV2 = DummyDeviceV2()
|
||||
self.app_id = "com.instagram.android"
|
||||
self.pressed = []
|
||||
self.clicks = []
|
||||
|
||||
def press(self, key):
|
||||
self.pressed.append(key)
|
||||
|
||||
def click(self, x, y):
|
||||
self.clicks.append((x, y))
|
||||
|
||||
|
||||
# ── XML Fixtures ──
|
||||
|
||||
|
||||
BOTTOM_SHEET_WITH_CANCEL = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/bottom_sheet_container"
|
||||
class="android.widget.FrameLayout" clickable="false"
|
||||
bounds="[0,800][1080,2424]">
|
||||
<node index="0" text="Mute" clickable="true" bounds="[100,900][980,1000]" />
|
||||
<node index="1" text="Restrict" clickable="true" bounds="[100,1010][980,1110]" />
|
||||
<node index="2" text="Block" clickable="true" bounds="[100,1120][980,1220]" />
|
||||
<node index="3" text="Cancel" clickable="true" bounds="[100,1400][980,1500]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
BOTTOM_SHEET_WITH_CLOSE_DESC = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/action_sheet_container"
|
||||
class="android.widget.FrameLayout" clickable="false"
|
||||
bounds="[0,600][1080,2424]">
|
||||
<node index="0" text="" content-desc="Close" clickable="true" bounds="[900,620][980,700]" />
|
||||
<node index="1" text="Share to..." clickable="true" bounds="[100,800][980,900]" />
|
||||
<node index="2" text="Copy Link" clickable="true" bounds="[100,910][980,1010]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
BOTTOM_SHEET_NO_DISMISS = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/bottom_sheet_container"
|
||||
class="android.widget.FrameLayout" clickable="false"
|
||||
bounds="[0,800][1080,2424]">
|
||||
<node index="0" text="Mute" clickable="true" bounds="[100,900][980,1000]" />
|
||||
<node index="1" text="Restrict" clickable="true" bounds="[100,1010][980,1110]" />
|
||||
<node index="2" text="Block" clickable="true" bounds="[100,1120][980,1220]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
NOT_NOW_DIALOG = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/survey_overlay_container"
|
||||
class="android.widget.FrameLayout" clickable="false"
|
||||
bounds="[0,0][1080,2424]">
|
||||
<node index="0" text="Turn on Notifications?" clickable="false" bounds="[100,800][980,900]" />
|
||||
<node index="1" text="Turn On" clickable="true" bounds="[100,950][980,1050]" />
|
||||
<node index="2" text="Not Now" clickable="true" bounds="[100,1060][980,1160]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
GERMAN_DISMISS_DIALOG = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" package="com.instagram.android"
|
||||
resource-id="com.instagram.android:id/interstitial_container"
|
||||
class="android.widget.FrameLayout" clickable="false"
|
||||
bounds="[0,0][1080,2424]">
|
||||
<node index="0" text="Benachrichtigungen aktivieren?" clickable="false" bounds="[100,800][980,900]" />
|
||||
<node index="1" text="Aktivieren" clickable="true" bounds="[100,950][980,1050]" />
|
||||
<node index="2" text="Nicht jetzt" clickable="true" bounds="[100,1060][980,1160]" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
|
||||
# ── Tests ──
|
||||
|
||||
|
||||
class TestStructuralDismissTarget:
|
||||
"""Verify that _find_structural_dismiss_target discovers buttons from XML, not hardcoded values."""
|
||||
|
||||
def _get_sae(self):
|
||||
device = DummyDevice()
|
||||
return SituationalAwarenessEngine.get_instance(device)
|
||||
|
||||
def test_finds_cancel_button_from_xml(self):
|
||||
"""RED: structural scan must find 'Cancel' button with XML-derived coordinates."""
|
||||
sae = self._get_sae()
|
||||
result = sae._find_structural_dismiss_target(BOTTOM_SHEET_WITH_CANCEL)
|
||||
|
||||
assert result is not None, "Should have found the Cancel button"
|
||||
assert result.action_type == "click"
|
||||
# Coordinates must come from bounds="[100,1400][980,1500]" → center (540, 1450)
|
||||
assert result.x == 540
|
||||
assert result.y == 1450
|
||||
assert "cancel" in result.reason.lower()
|
||||
|
||||
def test_finds_close_via_content_desc(self):
|
||||
"""RED: structural scan must find 'Close' button via content-desc attribute."""
|
||||
sae = self._get_sae()
|
||||
result = sae._find_structural_dismiss_target(BOTTOM_SHEET_WITH_CLOSE_DESC)
|
||||
|
||||
assert result is not None, "Should have found the Close button via content-desc"
|
||||
assert result.action_type == "click"
|
||||
# bounds="[900,620][980,700]" → center (940, 660)
|
||||
assert result.x == 940
|
||||
assert result.y == 660
|
||||
assert "close" in result.reason.lower()
|
||||
|
||||
def test_returns_none_when_no_dismiss_exists(self):
|
||||
"""RED: if the XML has no dismiss-type button, return None (don't invent one)."""
|
||||
sae = self._get_sae()
|
||||
result = sae._find_structural_dismiss_target(BOTTOM_SHEET_NO_DISMISS)
|
||||
|
||||
assert result is None, "Must not invent dismiss targets when none exist in XML"
|
||||
|
||||
def test_finds_not_now_button(self):
|
||||
"""RED: structural scan must find 'Not Now' button."""
|
||||
sae = self._get_sae()
|
||||
result = sae._find_structural_dismiss_target(NOT_NOW_DIALOG)
|
||||
|
||||
assert result is not None, "Should have found 'Not Now' button"
|
||||
assert result.action_type == "click"
|
||||
# bounds="[100,1060][980,1160]" → center (540, 1110)
|
||||
assert result.x == 540
|
||||
assert result.y == 1110
|
||||
|
||||
def test_finds_german_nicht_jetzt(self):
|
||||
"""RED: structural scan must find German 'Nicht jetzt' button."""
|
||||
sae = self._get_sae()
|
||||
result = sae._find_structural_dismiss_target(GERMAN_DISMISS_DIALOG)
|
||||
|
||||
assert result is not None, "Should have found 'Nicht jetzt' button"
|
||||
assert result.action_type == "click"
|
||||
# bounds="[100,1060][980,1160]" → center (540, 1110)
|
||||
assert result.x == 540
|
||||
assert result.y == 1110
|
||||
|
||||
def test_no_hardcoded_coordinates_in_method(self):
|
||||
"""
|
||||
META-TEST: Prove that _find_structural_dismiss_target contains
|
||||
zero hardcoded pixel values. All coordinates must come from XML parsing.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
source = inspect.getsource(SituationalAwarenessEngine._find_structural_dismiss_target)
|
||||
|
||||
# Should NOT contain any hardcoded coordinate patterns like x=540, y=150, etc.
|
||||
# The only numbers allowed are in the regex patterns, not in EscapeAction constructors
|
||||
import re
|
||||
|
||||
# Find all EscapeAction instantiations with literal x= or y= values
|
||||
hardcoded_coords = re.findall(r"EscapeAction\([^)]*(?:x=\d+|y=\d+)", source)
|
||||
assert len(hardcoded_coords) == 0, (
|
||||
f"Found hardcoded coordinates in _find_structural_dismiss_target: {hardcoded_coords}. "
|
||||
"All coordinates must come from XML bounds parsing."
|
||||
)
|
||||
|
||||
|
||||
class TestLLMFallbackNoHardcoding:
|
||||
"""Verify that _plan_escape_via_llm does NOT use hardcoded fallback coordinates."""
|
||||
|
||||
def test_no_hardcoded_coordinates_in_llm_planner(self):
|
||||
"""
|
||||
META-TEST: Prove that _plan_escape_via_llm contains zero hardcoded
|
||||
pixel coordinates as fallback values. When the LLM repeats, it must
|
||||
delegate to _find_structural_dismiss_target (which reads from XML).
|
||||
"""
|
||||
import inspect
|
||||
import re
|
||||
|
||||
source = inspect.getsource(SituationalAwarenessEngine._plan_escape_via_llm)
|
||||
|
||||
# Find all EscapeAction("click", x=<number>, y=<number>) with hardcoded coords
|
||||
hardcoded = re.findall(r'EscapeAction\(\s*"click"\s*,\s*x=\d+\s*,\s*y=\d+', source)
|
||||
assert len(hardcoded) == 0, (
|
||||
f"Found hardcoded click coordinates in _plan_escape_via_llm: {hardcoded}. "
|
||||
"The bot must discover coordinates autonomously from the XML."
|
||||
)
|
||||
@@ -11,15 +11,15 @@ def test_media_intent_rejects_grid_containers():
|
||||
engine = TelepathicEngine()
|
||||
screen_height = 2400
|
||||
|
||||
from GramAddict.core.perception.spatial_parser import SpatialNode
|
||||
|
||||
# Mock node representing a massive RecyclerView containing the entire grid
|
||||
# Area is 1080 * 2400 = 2592000 > MAX_CONTAINER_AREA (500000)
|
||||
massive_grid_container = {
|
||||
"semantic_string": "id context: 'swipeable nav view pager inner recycler view'",
|
||||
"area": 2592000,
|
||||
"y": 1200,
|
||||
"class_name": "androidx.recyclerview.widget.RecyclerView",
|
||||
"resource_id": "com.instagram.android:id/swipeable_nav_view_pager_inner_recycler_view",
|
||||
}
|
||||
massive_grid_container = SpatialNode(
|
||||
bounds=(0, 0, 1080, 2400),
|
||||
class_name="androidx.recyclerview.widget.RecyclerView",
|
||||
resource_id="com.instagram.android:id/swipeable_nav_view_pager_inner_recycler_view",
|
||||
)
|
||||
|
||||
# Intent
|
||||
intent = "first image post in profile grid"
|
||||
|
||||
55
tests/unit/test_telepathic_trap_guard.py
Normal file
55
tests/unit/test_telepathic_trap_guard.py
Normal file
@@ -0,0 +1,55 @@
|
||||
from GramAddict.core.perception.spatial_parser import SpatialNode
|
||||
from GramAddict.core.telepathic_engine import TelepathicEngine
|
||||
|
||||
|
||||
def test_telepathic_engine_global_trap_guard():
|
||||
"""
|
||||
TDD Test: Proves that the TelepathicEngine structurally blocks dangerous UI
|
||||
elements (like 'audio', 'trending', 'save') regardless of the intent,
|
||||
unless explicitly requested.
|
||||
"""
|
||||
engine = TelepathicEngine()
|
||||
screen_height = 2400
|
||||
|
||||
# 1. Setup mock nodes for traps
|
||||
node_audio = SpatialNode(
|
||||
bounds=(0, 0, 100, 100),
|
||||
resource_id="com.instagram.android:id/audio_icon",
|
||||
class_name="android.widget.ImageView",
|
||||
)
|
||||
|
||||
node_save = SpatialNode(
|
||||
bounds=(0, 100, 100, 200),
|
||||
resource_id="com.instagram.android:id/save_button",
|
||||
class_name="android.widget.ImageView",
|
||||
)
|
||||
|
||||
node_normal = SpatialNode(
|
||||
bounds=(0, 200, 100, 300),
|
||||
resource_id="com.instagram.android:id/zoomable_view_container",
|
||||
class_name="android.widget.FrameLayout",
|
||||
)
|
||||
|
||||
# 2. Test Intents that SHOULD NOT trigger the traps
|
||||
intent_media = "post media content"
|
||||
|
||||
assert (
|
||||
engine._structural_sanity_check(node_audio, intent_media, screen_height) is False
|
||||
), "Guard failed: Audio icon allowed for 'post media content'."
|
||||
assert (
|
||||
engine._structural_sanity_check(node_save, intent_media, screen_height) is False
|
||||
), "Guard failed: Save button allowed for 'post media content'."
|
||||
assert (
|
||||
engine._structural_sanity_check(node_normal, intent_media, screen_height) is True
|
||||
), "Guard failed: Normal node was blocked incorrectly."
|
||||
|
||||
# 3. Test Intents that SHOULD trigger the traps
|
||||
intent_audio = "trending audio"
|
||||
intent_save = "save button"
|
||||
|
||||
assert (
|
||||
engine._structural_sanity_check(node_audio, intent_audio, screen_height) is True
|
||||
), "Guard failed: Audio icon blocked for 'trending audio'."
|
||||
assert (
|
||||
engine._structural_sanity_check(node_save, intent_save, screen_height) is True
|
||||
), "Guard failed: Save button blocked for 'save button'."
|
||||
Reference in New Issue
Block a user