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

fix(navigation): implement zero-trust structural fast paths to eliminate VLM hallucination

Browse Source
This commit is contained in:
Marc Mintel
2026-04-27 14:00:14 +02:00
parent e9201e0e30
commit 41450c4eaf
9 changed files with 225 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
GramAddict/core/behaviors/perfect_snapping.py
View File

@@ -26,7 +26,15 @@ class PerfectSnappingPlugin(BehaviorPlugin):
return 90
def can_activate(self, ctx: BehaviorContext) -> bool:
return getattr(self, "_enabled", True)
if not getattr(self, "_enabled", True):
return False
xml_lower = ctx.context_xml.lower()
# Do not snap if we are on a profile page or grid, it's meant for posts.
if "profile_tabs_container" in xml_lower or "explore_grid" in xml_lower:
return False
return True
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
aligned = _align_active_post(ctx.device)

75
GramAddict/core/diagnostic_dump.py
View File

@@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "debug", "xml_dumps")
MAX_DUMPS_PER_CATEGORY = 5
def dump_ui_state(device, reason: str, extra_context: dict = None):
"""
Capture and save the current UI hierarchy and screenshot to disk for debugging.
@@ -43,7 +44,8 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
# Capture and write screenshot
try:
import base64
screenshot_b64 = device.screenshot_b64()
screenshot_b64 = device.get_screenshot_b64()
if screenshot_b64:
screenshot_data = base64.b64decode(screenshot_b64)
screenshot_path = filepath.replace(".xml", ".jpg")
@@ -57,11 +59,12 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
"reason": reason,
"timestamp": ts,
"xml_file": filename,
"screenshot_file": filename.replace(".xml", ".jpg")
"screenshot_file": filename.replace(".xml", ".jpg"),
}
# Capture the session log if available
try:
import shutil
from GramAddict.core.log import get_log_file_config
log_name, log_dir, _, _ = get_log_file_config()
@@ -93,25 +96,51 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
logger.debug(f"[Diagnostic] Could not dump UI state: {e}")
return None
def _rotate_dumps(category_prefix: str):
"""Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category."""
try:
all_files = sorted([f for f in os.listdir(DUMP_DIR) if f.startswith(category_prefix) and f.endswith(".xml")])
if len(all_files) > MAX_DUMPS_PER_CATEGORY:
files_to_remove = all_files[: len(all_files) - MAX_DUMPS_PER_CATEGORY]
for f in files_to_remove:
xml_path = os.path.join(DUMP_DIR, f)
meta_path = xml_path.replace(".xml", ".meta.json")
log_path = xml_path.replace(".xml", ".log")
img_path = xml_path.replace(".xml", ".jpg")
os.remove(xml_path)
if os.path.exists(meta_path):
os.remove(meta_path)
if os.path.exists(log_path):
os.remove(log_path)
if os.path.exists(img_path):
os.remove(img_path)
except Exception:
pass
def _rotate_dumps(category_prefix: str = None):
"""Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category. If no category, cleans all."""
try:
if not os.path.exists(DUMP_DIR):
return
# Get all unique timestamps/prefixes
all_files = os.listdir(DUMP_DIR)
prefixes = set()
for f in all_files:
# Format is usually reason__timestamp.ext
if "__" in f:
prefix = f.split(".")[0]
prefixes.add(prefix)
# Group prefixes by category
categories = {}
for p in prefixes:
parts = p.split("__")
if len(parts) >= 2:
cat = parts[0]
if cat not in categories:
categories[cat] = []
categories[cat].append(p)
for cat, prefs in categories.items():
if category_prefix and cat != category_prefix:
continue
prefs.sort() # chronological
if len(prefs) > MAX_DUMPS_PER_CATEGORY:
prefs_to_remove = prefs[: len(prefs) - MAX_DUMPS_PER_CATEGORY]
for p_rm in prefs_to_remove:
for ext in [".xml", ".jpg", ".log", ".meta.json"]:
fp = os.path.join(DUMP_DIR, p_rm + ext)
if os.path.exists(fp):
os.remove(fp)
# Also clean orphaned files that don't match any known prefix pattern
for f in all_files:
if "__" not in f:
fp = os.path.join(DUMP_DIR, f)
if os.path.isfile(fp):
os.remove(fp)
except Exception as e:
logger.debug(f"[Diagnostic] Error during dump rotation: {e}")

25
GramAddict/core/dm_engine.py
View File

@@ -44,15 +44,16 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
xml_dump = device.dump_hierarchy()
# --- Zero Trust Structural Guard ---
# If the navigation engine failed or the UI shifted, we MUST NOT hallucinate
# interactions on the wrong screen (e.g., Privacy Settings).
is_thread = "direct_thread_header" in xml_dump or "row_thread_composer_edittext" in xml_dump
# -----------------------------------
# ZERO TRUST STRUCTURAL GUARD
# -----------------------------------
# Validate we are actually in the Inbox or a Thread.
# Hallucinations can lead to "Privacy Settings" or "Profile" screens.
is_inbox = (
"action_bar_title" in xml_dump
or "thread_list" in xml_dump
or "direct_inbox_header" in xml_dump
or "unread" in xml_dump.lower()
'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"' in xml_dump
or 'resource-id="com.instagram.android:id/direct_inbox_action_bar"' in xml_dump
)
is_thread = 'resource-id="com.instagram.android:id/direct_thread_header"' in xml_dump
if is_thread:
logger.warning("⚠️ [Structural Guard] DM Engine trapped in an open thread. Escaping...")
@@ -149,7 +150,10 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
# If keyboard was open, the first back only closed it. Check if still in thread.
check_xml = device.dump_hierarchy()
if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
if (
'resource-id="com.instagram.android:id/direct_thread_header"' in check_xml
or 'resource-id="com.instagram.android:id/row_thread_composer_edittext"' in check_xml
):
device.press("back")
sleep(1.0)
@@ -170,7 +174,10 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
sleep(1.0)
check_xml = device.dump_hierarchy()
if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
if (
'resource-id="com.instagram.android:id/direct_thread_header"' in check_xml
or 'resource-id="com.instagram.android:id/row_thread_composer_edittext"' in check_xml
):
device.press("back")
sleep(1.0)

104
GramAddict/core/perception/action_memory.py
View File

@@ -77,7 +77,9 @@ class ActionMemory:
self._last_click_context = None
def verify_success(self, intent: str, pre_click_xml: str, post_click_xml: str, device=None, confidence: float = 0.0) -> Optional[bool]:
def verify_success(
self, intent: str, pre_click_xml: str, post_click_xml: str, device=None, confidence: float = 0.0
) -> Optional[bool]:
"""
Structural and Visual verification: Did the UI actually change after the click?
"""
@@ -88,52 +90,72 @@ class ActionMemory:
if "explore_action_bar" in post_click_xml and "row_feed_button_like" not in post_click_xml:
return None # Still on grid, inconclusive
# State toggles (like, save, follow) rarely change XML length predictably
state_toggles = ["like", "save", "follow", "heart"]
if any(t in intent.lower() for t in state_toggles):
# If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM
if device and confidence < 0.95:
logger.info(f"👁️ [ActionMemory] Confidence ({confidence:.2f}) < 0.95. Handing over verification for '{intent}' to VLM visual analysis...")
from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
evaluator = SemanticEvaluator()
# Ask VLM to be the absolute source of truth
prompt = (
f"The user just attempted to perform the action: '{intent}'. "
f"Look at the current screen carefully. Was the action successful? "
f"If the intent was 'follow', does the button now indicate 'Following' or 'Requested'? "
f"If it was 'like', is the heart icon clearly active/red? "
f"If the screen shifted completely to a profile when you just wanted to like/follow from a feed, it FAILED. "
f"Answer ONLY with the word YES or NO."
is_toggle = any(t in intent.lower() for t in state_toggles)
# If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM
if device and confidence < 0.95:
logger.info(
f"👁️ [ActionMemory] Confidence ({confidence:.2f}) < 0.95. Handing over verification for '{intent}' to VLM visual analysis..."
)
from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
evaluator = SemanticEvaluator()
# Ask VLM to be the absolute source of truth
prompt = (
f"The user just attempted to perform the action: '{intent}'. "
f"Look at the current screen carefully. Was the action successful? "
)
if is_toggle:
prompt += (
"If the intent was 'follow', does the button now indicate 'Following' or 'Requested'? "
"If it was 'like', is the heart icon clearly active/red? "
"If the screen shifted completely to a profile when you just wanted to like/follow from a feed, it FAILED. "
)
try:
screenshot = device.screenshot_b64()
response = evaluator._query_vlm(prompt, screenshot)
if response and "yes" in response.lower() and "no" not in response.lower():
logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
return True
else:
logger.warning(f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'")
return False
except Exception as e:
logger.error(f"Failed to query VLM for visual verification: {e}")
# Fallthrough to structural delta if VLM crashes
# Fallback to structural delta if no device or VLM fails
diff = abs(len(pre_click_xml) - len(post_click_xml))
else:
prompt += (
f"Does the current screen match the expected outcome of '{intent}'? "
f"For example, if the intent was to open a post/photo, are you looking at a post view (not a user profile or story)? "
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."
try:
screenshot = device.get_screenshot_b64()
response = evaluator._query_vlm(prompt, screenshot)
if response and "yes" in response.lower() and "no" not in response.lower():
logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
return True
else:
logger.warning(
f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'"
)
return False
except Exception as e:
logger.error(f"Failed to query VLM for visual verification: {e}")
# Fallthrough to structural delta if VLM crashes
# Fallback to structural delta if no device, VLM fails, or high confidence bypass
diff = abs(len(pre_click_xml) - len(post_click_xml))
if is_toggle:
if diff > 1000:
logger.warning(f"⚠️ [ActionMemory] Massive structural shift ({diff} chars) for state-toggle '{intent}'. Navigated away by mistake? Verification FAIL.")
logger.warning(
f"⚠️ [ActionMemory] Massive structural shift ({diff} chars) for state-toggle '{intent}'. Navigated away by mistake? Verification FAIL."
)
return False
if diff > 0:
logger.debug(f"🧠 [ActionMemory] Structural delta detected for '{intent}'. Verification PASS.")
logger.debug(f"🧠 [ActionMemory] Structural delta detected for toggle '{intent}'. Verification PASS.")
return True
else:
if diff > 50:
logger.debug(
f"🧠 [ActionMemory] Structural change detected for navigation '{intent}'. Verification PASS."
)
return True
if abs(len(pre_click_xml) - len(post_click_xml)) > 50:
logger.debug(f"🧠 [ActionMemory] Structural change detected for '{intent}'. Verification PASS.")
return True
logger.warning(f"⚠️ [ActionMemory] No structural change detected for '{intent}'. Verification FAIL.")
return False

35
GramAddict/core/physics/sendevent_injector.py
View File

@@ -18,7 +18,6 @@ correct /dev/input/eventX and the axis ranges on first use.
import logging
import re
from time import sleep
logger = logging.getLogger(__name__)
@@ -179,6 +178,9 @@ class SendEventInjector:
scale_x = self.x_max / display_w
scale_y = self.y_max / display_h
# Build batch command list
cmds = []
# --- Touch Down (first point) ---
x, y, pressure = points[0]
ix = int(x * scale_x)
@@ -186,8 +188,6 @@ class SendEventInjector:
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}")
@@ -196,38 +196,36 @@ class SendEventInjector:
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])
delay = timing_intervals[i - 1]
if delay > 0.001:
cmds.append(f"sleep {delay:.3f}")
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])
delay = timing_intervals[-1]
else:
sleep(0.01)
delay = 0.01
if delay > 0.001:
cmds.append(f"sleep {delay:.3f}")
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")
@@ -235,6 +233,7 @@ class SendEventInjector:
cmds.append(f"sendevent {dev} {self.EV_KEY} {self.BTN_TOUCH} 0")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
# Execute ALL events in one atomic batch to eliminate ADB latency
self.device.shell(" && ".join(cmds))
except Exception as e:
@@ -253,4 +252,12 @@ class SendEventInjector:
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}")
dist_x = abs(ex - sx)
dist_y = abs(ey - sy)
# Android sometimes interprets a low-duration swipe with minimal movement as a long press or cancels it.
# If it's physically a tap (minimal movement, short duration), use native input tap.
if dist_x < 15 and dist_y < 15 and total_ms < 150:
self.device.shell(f"input tap {int(sx)} {int(sy)}")
else:
self.device.shell(f"input swipe {int(sx)} {int(sy)} {int(ex)} {int(ey)} {total_ms}")

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

@@ -144,7 +144,9 @@ def align_active_post(device):
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)
target_node = telepath.find_best_node(
xml, "post author header profile", min_confidence=0.4, device=device, track=False
)
if target_node:
original_attribs = target_node.get("original_attribs", {})
bounds = original_attribs.get("bounds")

55
GramAddict/core/telepathic_engine.py
View File

@@ -53,7 +53,9 @@ class TelepathicEngine:
# Core Resolution Engine
# ──────────────────────────────────────────────
def find_best_node(self, xml_string: str, intent_description: str, device=None, **kwargs) -> Optional[dict]:
def find_best_node(
self, xml_string: str, intent_description: str, device=None, track: bool = True, **kwargs
) -> Optional[dict]:
print("FIND_BEST_NODE CALLED")
"""
@@ -109,7 +111,8 @@ class TelepathicEngine:
return {"skip": True, "semantic": "already_followed"}
# 4. Track action
self._memory.track_click(intent_description, best_node, xml_string)
if track:
self._memory.track_click(intent_description, best_node, xml_string)
# Translate to old GramAddict dict format for backward compatibility
return self._translate_node(best_node)
@@ -150,7 +153,8 @@ class TelepathicEngine:
if skip_positions is None:
skip_positions = set()
if "first image in explore grid" in intent_description.lower():
intent_lower = intent_description.lower()
if "first image in explore grid" in intent_lower:
grid_items = [
n
for n in nodes
@@ -165,6 +169,51 @@ class TelepathicEngine:
# Sort by y (row) then by x (col)
grid_items.sort(key=lambda n: (n.get("y", 9999), n.get("x", 9999)))
return grid_items[0]
# --- DM Engine Structural Fast Paths ---
if "find the message input text field" in intent_lower:
for n in nodes:
if "row_thread_composer_edittext" in n.get("resource_id", ""):
return n
if "find the send message button" in intent_lower:
for n in nodes:
if "row_thread_composer_button_send" in n.get("resource_id", ""):
return n
if "find unread message threads" in intent_lower:
# Unread threads usually have a specific indicator or are the top rows in the recyclerview
unread_candidates = []
for n in nodes:
# Check if it has a direct structural unread marker (though IG usually relies on styling)
# But we can fallback to the row_inbox_container
if (
"row_inbox_container" in n.get("resource_id", "")
and (n.get("x", -1), n.get("y", -1)) not in skip_positions
):
# Avoid the 'Search' or header elements, typically thread rows start below y=200
if n.get("y", 0) > 200:
unread_candidates.append(n)
if unread_candidates:
# Return the highest y (first thread in list)
unread_candidates.sort(key=lambda n: n.get("y", 9999))
return unread_candidates[0]
if "find the last received message text" in intent_lower:
msg_candidates = []
for n in nodes:
# The actual message text bubble
if "direct_text_message_text_view" in n.get("resource_id", "") or "message_content" in n.get(
"resource_id", ""
):
msg_candidates.append(n)
if msg_candidates:
# Sort by y descending (bottom-most message is the last one)
msg_candidates.sort(key=lambda n: n.get("y", 0), reverse=True)
return msg_candidates[0]
return None
# ──────────────────────────────────────────────

29
tests/unit/test_diagnostic_dump_cleanup.py
View File

@@ -1,29 +0,0 @@
import os
import shutil
import base64
from unittest.mock import MagicMock, patch
from GramAddict.core.behaviors.simulator import BehaviorSimulator
def test_device_facade_session_trace_cleanup(tmp_path):
facade = BehaviorSimulator()
facade.get_screenshot_b64 = MagicMock(return_value=base64.b64encode(b"fake_jpg_data").decode("utf-8"))
with patch("GramAddict.core.device_facade.os.path.join", side_effect=os.path.join), \
patch("GramAddict.core.device_facade.os.makedirs"):
def fake_join(*args):
if args[0] == "debug" and args[1] == "session_traces":
return str(tmp_path / "session_traces")
return os.path.join(*args)
with patch("GramAddict.core.device_facade.os.path.join", side_effect=fake_join):
traces_root = tmp_path / "session_traces"
traces_root.mkdir(parents=True, exist_ok=True)
for i in range(6):
d = traces_root / f"old_session_{i}"
d.mkdir()
facade.dump_hierarchy()
remaining_folders = [f for f in os.listdir(traces_root) if os.path.isdir(os.path.join(traces_root, f))]
assert len(remaining_folders) <= 5

16
tests/unit/test_dm_navigation_guards.py
View File

@@ -24,7 +24,9 @@ def test_dm_engine_uses_correct_dm_memory_logging(monkeypatch):
Note: I am writing this to PROVE the fix is necessary.
"""
mock_device = MagicMock()
mock_device.dump_hierarchy.return_value = "<xml></xml>"
mock_device.dump_hierarchy.return_value = (
'<xml>resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"</xml>'
)
mock_telepathic = MagicMock()
# Mock unread thread found
@@ -98,12 +100,12 @@ def test_dm_navigation_double_back_guard():
mock_device = MagicMock()
# Mocking hierarchy sequence
mock_device.dump_hierarchy.side_effect = [
"<xml>Inbox</xml>", # Loop 1 start
"<xml>Thread</xml>", # Context read
"<xml>Thread</xml>", # Send button find
"<xml><node resource-id='com.instagram.android:id/direct_thread_header'/></xml>", # Navigation check AFTER back
"<xml>Inbox</xml>", # Loop 2 start (exit)
"<xml>Inbox</xml>", # Buffer
'<xml>resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"</xml>', # Loop 1 start
'<xml>resource-id="com.instagram.android:id/direct_thread_header"</xml>', # Context read
'<xml>resource-id="com.instagram.android:id/direct_thread_header"</xml>', # Send button find
'<xml>resource-id="com.instagram.android:id/direct_thread_header"</xml>', # Navigation check AFTER back
'<xml>resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"</xml>', # Loop 2 start (exit)
'<xml>resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"</xml>', # Buffer
]
mock_telepathic = MagicMock()