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

test(e2e): eliminate create_emulator_facade monkeypatching and replace with make_real_device_with_xml where applicable

Browse Source
This commit is contained in:
Marc Mintel
2026-05-03 11:31:26 +02:00
parent f0a54d4e20
commit f85d0a8a76
20 changed files with 472 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
GramAddict/core/bot_flow.py
View File

@@ -925,6 +925,12 @@ def _run_zero_latency_feed_loop(
elif governance_decision == "CHECK_CURIOSITY":
logger.info("👀 [Curiosity] Spontaneously checking DMs / Notifications...")
# 🛡️ Structural Guard: Curiosity targets (DMs, Notifications) are ONLY available on HomeFeed.
# We must navigate there first, breaking current context.
nav_graph.navigate_to("HomeFeed", zero_engine)
sleep(random.uniform(1.0, 2.5))
dm_config = configs.get_plugin_config("dm_reply")
if dm_config.get("enabled", False):
explore_target = random.choice(["MessageInbox", "Notifications"])

3
GramAddict/core/dm_engine.py
View File

@@ -86,6 +86,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
identity_engine = ScreenIdentity(getattr(configs.args, "username", ""))
identity_engine.device = device
screen_info = identity_engine.identify(xml_dump)
screen_type = screen_info["screen_type"]
@@ -220,6 +221,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
check_identity = ScreenIdentity(getattr(configs.args, "username", ""))
check_identity.device = device
check_screen = check_identity.identify(check_xml)
if check_screen["screen_type"] == ScreenType.DM_THREAD:
@@ -246,6 +248,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
check_identity = ScreenIdentity(getattr(configs.args, "username", ""))
check_identity.device = device
check_screen = check_identity.identify(check_xml)
if check_screen["screen_type"] == ScreenType.DM_THREAD:

4
GramAddict/core/goap.py
View File

@@ -63,6 +63,7 @@ class GoalExecutor:
self.device = device
self.username = bot_username
self.screen_id = ScreenIdentity(bot_username)
self.screen_id.device = device
self.planner = GoalPlanner(bot_username)
self.path_memory = PathMemory(bot_username)
self.max_steps = 15 # Safety: never execute more than 15 steps
@@ -239,7 +240,8 @@ class GoalExecutor:
from GramAddict.core.screen_topology import ScreenTopology
keys_to_clear = [
k for k in self.action_failures.keys()
k
for k in self.action_failures.keys()
if k[0] == screen_type and ScreenTopology.is_structural_action(screen_type, k[1])
]
for k in keys_to_clear:

89
GramAddict/core/perception/screen_identity.py
View File

@@ -43,7 +43,7 @@ class ScreenIdentity:
except ImportError:
self.screen_memory = None
def identify(self, xml_dump: str) -> Dict[str, Any]:
def identify(self, xml_dump: str, screenshot_b64: str = None) -> Dict[str, Any]:
"""
Analyzes an XML dump and returns a complete screen description.
@@ -116,6 +116,11 @@ class ScreenIdentity:
}
)
from GramAddict.core.situational_awareness import SituationalAwarenessEngine
sae = SituationalAwarenessEngine.get_instance()
signature = sae._compress_xml(xml_dump) if sae else self._compute_signature(resource_ids, content_descs, texts)
# ── Foreign app check ──
if app_id not in packages:
return {
@@ -123,18 +128,16 @@ class ScreenIdentity:
"available_actions": ["press back", "force start instagram"],
"selected_tab": None,
"context": {"packages": list(packages)},
"signature": self._compute_signature(resource_ids, content_descs, texts),
"signature": signature,
}
desc_lower = " ".join(content_descs).lower()
text_lower = " ".join(texts).lower()
ids_str = " ".join(resource_ids).lower()
signature = self._compute_signature(resource_ids, content_descs, texts)
# ── Identify screen type from structural signals ──
screen_type = self._classify_screen(
resource_ids, content_descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature
resource_ids, content_descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature, screenshot_b64
)
# ── Extract available actions from clickable elements ──
@@ -153,16 +156,36 @@ class ScreenIdentity:
"signature": signature,
}
def _classify_screen(self, ids, descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature=None):
"""Classify screen type using Semantic Memory with LLM fallback — NO hardcoded states."""
def _classify_screen(
self, ids, descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature=None, screenshot_b64=None
):
"""
Classify screen type using Semantic Memory with LLM fallback — NO hardcoded states."""
# Priority 0: Content-creation overlays that block ALL navigation.
# Priority 0: Check Qdrant Semantic Cache (Learned Truth/LLM Overrides)
# This MUST be checked first. If the LLM declared this specific layout a "false positive"
# and cached it as NORMAL, it must override any rigid structural heuristics below to prevent
# infinite loops.
is_normal_override = False
if signature and self.screen_memory and self.screen_memory.is_connected:
cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92)
if cached_type_str:
if cached_type_str == "NORMAL":
is_normal_override = True
else:
try:
return ScreenType[cached_type_str]
except KeyError:
pass
# Priority 1: Content-creation overlays that block ALL navigation.
# These full-screen Instagram UIs have no navigation tabs and trap the bot.
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
if not is_normal_override:
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
# Priority 1: Structural Heuristics (100% Deterministic)
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
@@ -225,50 +248,52 @@ class ScreenIdentity:
if "message_input" in ids:
return ScreenType.DM_INBOX # Fallback for DM thread as inbox
# Priority 2: Check Qdrant Semantic Cache (Fuzzy/VLM derived)
if signature and self.screen_memory and self.screen_memory.is_connected:
cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92)
if cached_type_str:
try:
return ScreenType[cached_type_str]
except KeyError:
pass
# End of structural heuristics
# Priority 3: Semantic VLM Classification Fallback
if not screenshot_b64 and getattr(self, "device", None) is not None:
screenshot_b64 = self.device.get_screenshot_b64()
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_llm
from GramAddict.core.llm_provider import query_telepathic_llm
cfg = Config()
url = (
getattr(cfg.args, "ai_model_url", "http://localhost:11434/api/generate")
getattr(cfg.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
if hasattr(cfg, "args")
else "http://localhost:11434/api/generate"
)
model = getattr(cfg.args, "ai_model", "qwen3.5:latest") if hasattr(cfg, "args") else "qwen3.5:latest"
model = getattr(cfg.args, "ai_telepathic_model", "llava:latest") if hasattr(cfg, "args") else "llava:latest"
layout_context = (
f"Selected Tab: {selected_tab}\nResource IDs: {list(ids)}\nVisible Texts context: {texts[:10]}\n"
)
prompt = (
f"Identify the Instagram screen layout type based on these DOM structural signals.\n"
f"Identify the Instagram screen layout type based on the provided screenshot and structural signals.\n"
f"Valid types: {[t.name for t in ScreenType]}\n"
f"Context:\n{layout_context}\n"
f"Reply ONLY with the exact matching enum Type Name string, or 'UNKNOWN' if no type matches."
)
try:
response = query_llm(
url=url, model=model, prompt="Classify this screen layout.", system=prompt, format_json=False
response = query_telepathic_llm(
model=model,
url=url,
system_prompt=prompt,
user_prompt="Classify this screen layout.",
images_b64=[screenshot_b64] if screenshot_b64 else None,
temperature=0.0,
use_local_edge=True,
)
if response and isinstance(response, str):
result = response.strip().upper()
elif response and isinstance(response, dict) and "response" in response:
result = response["response"].strip().upper()
else:
return ScreenType.UNKNOWN
result = response.strip().upper() if response else "UNKNOWN"
for t in ScreenType:
if t.name in result:
if is_normal_override and t == ScreenType.MODAL:
# Prevent the LLM from hallucinating an obstacle if explicitly verified as NORMAL
return ScreenType.UNKNOWN
if signature and self.screen_memory:
self.screen_memory.store_screen(signature, t.name)
return t

9
GramAddict/core/qdrant_memory.py
View File

@@ -184,6 +184,7 @@ class QdrantBase:
self.client.upsert(
collection_name=self.collection_name,
points=[PointStruct(id=point_id, vector=safe_vector, payload=payload)],
wait=True,
)
# ABSOLUTE LOGGING: User requirement for full observability
@@ -431,7 +432,7 @@ class UIMemoryDB(QdrantBase):
if eval_result:
logger.info(
f"🧠 [Memory] Applying learned pattern for '{intent}' (EXACT MATCH, Confidence: {eval_result['effective_confidence']:.2f})",
extra={"color": "\x1b[36m"} # Cyan color
extra={"color": "\x1b[36m"}, # Cyan color
)
return eval_result["solution"]
# If exact match failed evaluation (e.g. decayed), we shouldn't fall back to vector search because it's the exact intent!
@@ -462,7 +463,7 @@ class UIMemoryDB(QdrantBase):
if eval_result:
logger.info(
f"🧠 [Memory] Applying learned pattern for '{intent}' (VECTOR MATCH, Score: {results[0].score:.3f}, Confidence: {eval_result['effective_confidence']:.2f})",
extra={"color": "\x1b[36m"} # Cyan color
extra={"color": "\x1b[36m"}, # Cyan color
)
return eval_result["solution"]
return None
@@ -515,7 +516,7 @@ class UIMemoryDB(QdrantBase):
)
logger.info(
f"📥 [Memory] Learned new pattern for '{intent}' and saved to Qdrant (ID: {point_id[:8]}...)",
extra={"color": "\x1b[35m"} # Magenta color
extra={"color": "\x1b[35m"}, # Magenta color
)
except Exception as e:
logger.debug(f"Qdrant storage error: {e}")
@@ -582,7 +583,7 @@ class UIMemoryDB(QdrantBase):
symbol = "📈 [Memory] Positive Reinforcement:" if delta > 0 else "📉 [Memory] Negative Reinforcement:"
logger.info(
f"{symbol} Confidence for '{intent}' adjusted to {new_confidence:.2f} (delta: {delta:+.2f})",
extra={"color": color}
extra={"color": color},
)
except Exception as e:
logger.debug(f"Confidence adjustment error: {e}")

83
GramAddict/core/situational_awareness.py
View File

@@ -270,7 +270,13 @@ class SituationalAwarenessEngine:
if clickable == "true":
parts.append("CLICKABLE")
if bounds:
parts.append(f"bounds={bounds}")
nums = [int(n) for n in re.findall(r"\d+", bounds)]
if len(nums) == 4:
cx = (nums[0] + nums[2]) // 2
cy = (nums[1] + nums[3]) // 2
parts.append(f"bounds={bounds} center=({cx},{cy})")
else:
parts.append(f"bounds={bounds}")
elements.append(" | ".join(parts))
@@ -406,11 +412,21 @@ class SituationalAwarenessEngine:
compressed = self._compress_xml(xml_dump)
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "NORMAL":
return SituationType.NORMAL
# ── Structural Fast-Check: Content-Creation Overlays ──
# These full-screen overlays live INSIDE Instagram's package but block
# all normal navigation. They are invisible to the foreign-app detector
# and frequently fool the LLM into thinking they are "normal" browsing.
# Detecting them structurally is O(1) and requires ZERO LLM calls.
# This is checked AFTER Qdrant to ensure that if the LLM unlearned a false positive,
# we respect the learned NORMAL state and don't infinite-loop.
creation_flow_markers = (
"quick_capture", # Camera / story capture overlay
"gallery_cancel_button", # Story gallery "Back to Home" button
@@ -427,14 +443,6 @@ class SituationalAwarenessEngine:
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "NORMAL":
return SituationType.NORMAL
# If not cached, query LLM for autonomous structural classification
try:
from GramAddict.core.config import Config
@@ -442,7 +450,7 @@ class SituationalAwarenessEngine:
prompt = (
"You are a Situation Classifier for a mobile automation agent.\n"
"Analyze the given Android UI XML dump. Is there a blocking MODAL, DIALOG, or POPUP "
"Analyze the given Android UI XML dump AND screenshot. Is there a blocking MODAL, DIALOG, or POPUP "
"covering the screen that needs to be dismissed, or is this a NORMAL usable screen?\n"
"A 'clean_sheet_container' with standard Instagram feed content is NORMAL.\n"
"A survey, rating prompt, 'not now' prompt, or permission dialog is an OBSTACLE_MODAL.\n"
@@ -459,11 +467,17 @@ class SituationalAwarenessEngine:
args = Config().args
except Exception:
pass
model = getattr(args, "ai_model", "qwen3.5:latest")
url = getattr(args, "ai_model_url", "http://localhost:11434/api/generate")
model = getattr(args, "ai_telepathic_model", "llava:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
screenshot_b64 = getattr(self.device, "get_screenshot_b64", lambda: None)()
res = query_telepathic_llm(
model=model, url=url, system_prompt="Strict JSON classifier.", user_prompt=prompt, use_local_edge=True
model=model,
url=url,
system_prompt="Strict JSON classifier.",
user_prompt=prompt,
images_b64=[screenshot_b64] if screenshot_b64 else None,
use_local_edge=True,
)
import json
@@ -504,27 +518,29 @@ class SituationalAwarenessEngine:
Called ONLY when recall AND structural planning both miss.
"""
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_llm
from GramAddict.core.llm_provider import query_telepathic_llm
try:
args = Config().args
model = getattr(args, "ai_fallback_model", "llama3.2:1b")
url = getattr(args, "ai_fallback_url", "http://localhost:11434/api/generate")
model = getattr(args, "ai_telepathic_model", "llava:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
except Exception:
model = "llama3.2:1b"
model = "llava:latest"
url = "http://localhost:11434/api/generate"
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 and return a JSON escape action.\n\n"
"Analyze the screen content (Screenshot AND XML) and return a JSON escape action.\n\n"
"Rules:\n"
"- If you see a dismiss/close/cancel/skip/not now button, click it\n"
"- If 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_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, look for 'Deny', 'Don't allow', or 'Cancel' and click it. If none exist, action must be 'back'\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": "..."}'
)
@@ -535,20 +551,31 @@ class SituationalAwarenessEngine:
user_prompt += "What action should I take to clear this obstacle and return to Instagram? Return JSON only."
try:
resp = query_llm(
screenshot_b64 = getattr(self.device, "get_screenshot_b64", lambda: None)()
resp = query_telepathic_llm(
url=url,
model=model,
prompt=user_prompt,
system=system_prompt,
format_json=True,
timeout=30,
max_tokens=300,
user_prompt=user_prompt,
system_prompt=system_prompt,
images_b64=[screenshot_b64] if screenshot_b64 else None,
temperature=0.0,
)
if resp and "response" in resp:
if resp:
import json
data = json.loads(resp["response"])
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)
if match:
data = json.loads(match.group(0))
else:
raise ValueError(f"Could not parse JSON from: {resp}")
return EscapeAction(
action_type=data.get("action", "back"),
x=int(data.get("x", 0)),