8 Commits

Author SHA1 Message Date
96fdbd7db7 test(perception): add strict TDD suite for autonomous ad marker learning and verification 2026-04-29 21:29:05 +02:00
ca91ae4b33 feat(perception): autonomous FSD ad marker learning with zero-latency structural persistence 2026-04-29 21:25:03 +02:00
a560225dc9 feat(perception): integrate VLM visual ad detection into resonance vibe check to block obfuscated ads 2026-04-29 21:20:09 +02:00
849fb63426 fix(orchestrator): inject --goal into persona_interests for autonomous resonance evaluation and fix scope shadowing bugs 2026-04-29 19:30:21 +02:00
fc44633ebc fix(physics): prevent infinite alignment loops on Reels and refine intent to avoid follow buttons 2026-04-29 19:05:31 +02:00
0f5b71708d fix(perception): enforce visual Set of Mark by passing device context globally 2026-04-29 18:40:58 +02:00
0ef2840f79 fix(perception): complete bug 5,6,7,8,10 fixes
- Resolved Bug #5: Fixed list vs str parsing in ResonanceEvaluator
- Resolved Bug #6: Use 'should_like' key for vibe score
- Resolved Bug #7: Guard TelepathicEngine against 'Follow' nodes for post media
- Resolved Bug #8: Implemented failed_bounds exclusion loop breaker in PerfectSnapping
- Resolved Bug #10: Corrected available_actions string parsing
- Validated with E2E regression suite (100% green)
2026-04-29 18:26:29 +02:00
068a6a616a fix: purge 4 production bugs — resonance null-guard, POST_DETAIL misclassification, VLM JSON response handling
🔴 RED → 🟢 GREEN for 4 critical bugs found in production run 2026-04-29:

1. ResonanceEvaluator: Add null-guard for evaluate_post_vibe() return.
   When VLM returns truncated JSON, the function returns None. The caller
   now handles this gracefully instead of crashing with AttributeError.

2. ScreenIdentity POST_DETAIL: Replace broken 'and not selected_tab'
   condition with structural differentiator using main_feed_action_bar.
   Posts opened from feed retain feed_tab selected, which was causing
   misclassification as HOME_FEED → LLM fallback → OWN_PROFILE hallucination
   → permanent Qdrant cache poisoning.

3. ActionMemory VLM verification: When VLM returns JSON instead of YES/NO,
   treat as inconclusive (fall through to structural delta) rather than
   hard failure. Only return False when response explicitly contains 'no'.

4 new E2E regression tests, 75/75 pass, zero regressions.
2026-04-29 18:05:32 +02:00
16 changed files with 780 additions and 30 deletions

View File

@@ -47,7 +47,7 @@ def verify_and_switch_account(device, nav_graph, target_username):
telepath = TelepathicEngine.get_instance()
# We ask the semantic engine to find the profile tab, ensuring 100% ID-agnostic behavior
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3)
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3, device=device)
if profile_tab_node:
profile_tab = (profile_tab_node["x"], profile_tab_node["y"])
except Exception as e:
@@ -113,7 +113,7 @@ def verify_and_switch_account(device, nav_graph, target_username):
dump_ui_state(
device, "identity_guard", {"reason": "account_not_found_in_bottom_sheet", "target": target_username}
)
except:
except Exception:
pass
# Escape the bottom sheet
device.press("back")

View File

@@ -56,7 +56,7 @@ class ObstacleGuardPlugin(BehaviorPlugin):
# Check recovery
new_xml = ctx.device.dump_hierarchy()
tele = TelepathicEngine.get_instance()
best_node = tele.find_best_node(new_xml, intent_description="Dismiss obstacle")
best_node = tele.find_best_node(new_xml, intent_description="Dismiss obstacle", device=ctx.device)
if best_node:
ctx.device.click(best_node.get("x", 0), best_node.get("y", 0))

View File

@@ -30,7 +30,7 @@ class PostDataExtractionPlugin(BehaviorPlugin):
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
logger.debug("🧩 [PostDataExtraction] Extracting post metadata...")
post_data = extract_post_content(ctx.context_xml)
post_data = extract_post_content(ctx.context_xml, device=ctx.device)
if post_data:
ctx.post_data = post_data

View File

@@ -49,12 +49,37 @@ class ResonanceEvaluatorPlugin(BehaviorPlugin):
tele = ctx.cognitive_stack.get("telepathic")
if tele:
logger.info("✨ [Resonance] Performing visual vibe check...")
persona_interests = getattr(ctx.configs.args, "persona_interests", [])
# BUG 5 Fix: Read target_audience or persona_interests
raw_interests = getattr(ctx.configs.args, "persona_interests", "")
if not raw_interests:
raw_interests = getattr(ctx.configs.args, "target_audience", "")
if isinstance(raw_interests, list):
persona_interests = [str(i).strip() for i in raw_interests if str(i).strip()]
else:
persona_interests = [i.strip() for i in str(raw_interests).split(",") if i.strip()]
vibe = tele.evaluate_post_vibe(ctx.device, persona_interests)
vibe_score = vibe.get("quality_score", 5) / 10.0
if vibe.get("matches_niche"):
vibe_score = min(1.0, vibe_score + 0.2)
res_score = (res_score * 0.3) + (vibe_score * 0.7)
if vibe is None:
logger.warning(
"✨ [Resonance] VLM vibe check returned None (truncated JSON?). Keeping neutral score."
)
else:
if vibe.get("is_ad"):
logger.info("🛡️ [Resonance Oracle] Visually identified post as an Ad! Skipping...")
marker = vibe.get("ad_marker_text")
if marker and marker.strip():
from GramAddict.core.utils import learn_ad_marker
learn_ad_marker(marker, ctx.context_xml)
from GramAddict.core.utils import humanized_scroll
humanized_scroll(ctx.device)
return BehaviorResult(executed=True, should_skip=True)
# BUG 6 Fix: VLM returns {"should_like": true/false}, not "quality_score"
should_like = vibe.get("should_like", False)
vibe_score = 1.0 if should_like else 0.2
res_score = (res_score * 0.3) + (vibe_score * 0.7)
ctx.shared_state["res_score"] = res_score
logger.info(f"📊 [Resonance] Post Score: {res_score:.2f}")

View File

@@ -177,6 +177,11 @@ def start_bot(**kwargs):
)
persona_interests = [p.strip() for p in persona_raw.split(",") if p.strip()] if persona_raw else []
global_goal = getattr(configs.args, "goal", None)
if global_goal:
persona_interests.insert(0, global_goal)
logger.info(f"🎯 [Autonomous Directive] Overriding target audience with high-level goal: {global_goal}", extra={"color": f"{Style.BRIGHT}{Fore.GREEN}"})
from GramAddict.core.interaction import LLMWriter
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
from GramAddict.core.resonance_engine import ResonanceEngine

View File

@@ -152,6 +152,13 @@ class Config:
help="Wipe all learned navigation and telepathic memories on boot to start 100%% blank.",
)
self.parser.add_argument(
"--goal",
type=str,
help="High-level autonomous goal for the bot (Tesla-style). Overrides config.yml goals.",
default=None,
)
# Interaction settings
self.parser.add_argument("--likes-count", help="Likes count", default="2-3")
self.parser.add_argument("--likes-percentage", help="Likes percentage", default="100")

View File

@@ -1,10 +1,46 @@
import json
import logging
import re
from typing import Any, Dict, Optional
from GramAddict.core.perception.spatial_parser import SpatialNode
logger = logging.getLogger(__name__)
def _parse_yes_no(response: str) -> Optional[bool]:
"""Parses a VLM response to find a definitive YES or NO without substring-matching 'not' or 'now'."""
text = response.strip()
# Try parsing as JSON first
if text.startswith("{"):
try:
data = json.loads(text)
for k, v in data.items():
if str(k).strip().upper() == "YES" or str(v).strip().upper() == "YES":
return True
if str(k).strip().upper() == "NO" or str(v).strip().upper() == "NO":
return False
except Exception:
pass
text_lower = text.lower()
if text_lower.startswith("yes"):
return True
if text_lower.startswith("no") and not text_lower.startswith("now") and not text_lower.startswith("not"):
return False
has_yes = re.search(r"\byes\b", text_lower) is not None
has_no = re.search(r"\bno\b", text_lower) is not None
if has_yes and not has_no:
return True
if has_no and not has_yes:
return False
return None
# ═══════════════════════════════════════════════════════
# Semantic Match Keywords — SSOT for intent → element validation
# ═══════════════════════════════════════════════════════
@@ -189,14 +225,23 @@ class ActionMemory:
raise ValueError("No screenshot available from device")
response = evaluator._query_vlm(prompt, screenshot)
if response and "yes" in response.lower() and "no" not in response.lower():
decision = _parse_yes_no(response) if response else None
if decision is True:
logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
return True
else:
elif decision is False:
logger.warning(
f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'"
)
return False
else:
# VLM returned ambiguous response (JSON, mixed signals, etc.)
# Don't treat as hard failure — fall through to structural delta verification
logger.info(
f"🧠 [ActionMemory] VLM response for '{intent}' was not YES/NO "
f"(got: '{response[:80]}...'). Falling through to structural verification."
)
except Exception as e:
logger.error(f"Failed to query VLM for visual verification: {e}")
# Fallthrough to structural delta if VLM crashes
@@ -258,10 +303,13 @@ class ActionMemory:
prompt = f"The user just attempted to perform the action: '{intent}'. Does the current screen match the expected outcome? Answer ONLY with the word YES or NO."
try:
response = evaluator._query_vlm(prompt, device.get_screenshot_b64())
if response and "yes" in response.lower() and "no" not in response.lower():
decision = _parse_yes_no(response) if response else None
if decision is True:
return True
else:
logger.warning(f"⚠️ [ActionMemory] VLM rejected success for abstract intent '{intent}'.")
logger.warning(
f"⚠️ [ActionMemory] VLM rejected success for abstract intent '{intent}'. Response: '{response}'"
)
return False
except Exception as e:
logger.error(f"VLM visual verification failed: {e}")

View File

@@ -46,7 +46,7 @@ def has_carousel_in_view(xml_dump: str) -> bool:
return any(ind in xml_dump for ind in CAROUSEL_INDICATORS)
def extract_post_content(context_xml: str) -> dict:
def extract_post_content(context_xml: str, device=None) -> dict:
"""
Extracts meaningful content data from the current feed post's XML.
This is the BOT'S EYES — what it actually "sees" about each post.
@@ -62,14 +62,18 @@ def extract_post_content(context_xml: str) -> dict:
telepath = TelepathicEngine.get_instance()
# 1. Learn/extract post author dynamically
author_node = telepath.find_best_node(context_xml, "post author username header", min_confidence=0.75)
author_node = telepath.find_best_node(
context_xml, "post author username text (exclude bottom tabs)", min_confidence=0.75, device=device
)
# 🛡️ Anti-Hallucination Guard: Ensure we actually found text.
if author_node and author_node.get("original_attribs", {}).get("text"):
result["username"] = author_node["original_attribs"]["text"].strip()
# 2. Learn/extract post media description dynamically
media_node = telepath.find_best_node(context_xml, "post media content", min_confidence=0.35)
media_node = telepath.find_best_node(
context_xml, "post media content (the actual image or video, exclude bottom tabs)", min_confidence=0.35, device=device
)
if media_node and media_node.get("original_attribs", {}).get("desc"):
result["description"] = media_node["original_attribs"]["desc"].strip()

View File

@@ -365,6 +365,12 @@ class IntentResolver:
)
data = json.loads(res)
box_idx = data.get("box")
if box_idx is None:
box_idx = data.get("selected_index")
if box_idx is None:
box_idx = data.get("box_index")
if box_idx is None:
box_idx = data.get("index")
if box_idx is not None and box_idx in box_map:
selected = box_map[box_idx]

View File

@@ -193,8 +193,14 @@ class ScreenIdentity:
except KeyError:
pass
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab:
return ScreenType.POST_DETAIL
# POST_DETAIL vs HOME_FEED: Both have row_feed_* markers. The differentiator
# is that HOME_FEED has the main_feed_action_bar (top bar with 'Instagram' title).
# POST_DETAIL lacks this because it shows a single expanded post.
# Note: We MUST NOT use `not selected_tab` here — posts opened from feed
# retain the feed_tab as selected, which previously caused misclassification.
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids:
if "main_feed_action_bar" not in ids:
return ScreenType.POST_DETAIL
# Story view structural markers — present in full-screen story viewer.
# Stories hide the navigation tab bar, so selected_tab is always None.
@@ -304,7 +310,7 @@ class ScreenIdentity:
if "back" in desc_lower:
actions.append("tap back button")
if any("follow" in e.get("text", "").lower() for e in clickable_elements):
actions.append("tap 'Follow' button")
actions.append("tap follow button")
if screen_type == ScreenType.OWN_PROFILE or screen_type == ScreenType.OTHER_PROFILE:
if "message" in desc_lower or "nachricht" in desc_lower:

View File

@@ -117,11 +117,13 @@ class SemanticEvaluator:
You are a user with the following interests: {', '.join(persona_interests)}.
You are looking at an Instagram post.
Evaluate if this post is highly relevant to your interests and if you should like/comment on it.
CRITICAL: Check if this post is an advertisement or sponsored content (look for "Sponsored", "Ad", or promotional product placement).
Reply ONLY in valid JSON format:
{{
"should_like": true/false,
"should_comment": true/false,
"is_ad": true/false,
"reasoning": "brief explanation"
}}
"""

View File

@@ -136,11 +136,12 @@ def align_active_post(device):
aligned = False
attempts = 0
max_attempts = 5 # Increased for structural retry loop
failed_bounds = set()
# Intents for structural discovery
intents = [
"post author username text (exclude follow buttons)",
"post author header profile",
"post username name",
"row_feed_photo_profile_name", # ID fallback
"clips_viewer_author_container", # Reels fallback
"feed post content", # Final desperation
@@ -150,13 +151,19 @@ def align_active_post(device):
attempts += 1
try:
xml = device.dump_hierarchy()
if "clips_video_container" in xml or "clips_viewer_container" in xml:
logger.info("🎯 [Alignment] Reels view detected. Auto-snapping is native.")
return True
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
target_node = None
for intent in intents:
target_node = telepath.find_best_node(xml, intent, min_confidence=0.35, device=device, track=False)
target_node = telepath.find_best_node(
xml, intent, min_confidence=0.35, device=device, track=False, exclude_bounds=list(failed_bounds)
)
if target_node:
break
@@ -164,9 +171,11 @@ def align_active_post(device):
original_attribs = target_node.get("original_attribs", {})
bounds = original_attribs.get("bounds")
bounds_str = ""
# If bounds is a tuple from SpatialNode.to_dict()
if isinstance(bounds, (tuple, list)) and len(bounds) == 4:
left, t, r, b = bounds
bounds_str = f"[{left},{t}][{r},{b}]"
else:
# Fallback to string parsing
if not bounds:
@@ -174,6 +183,7 @@ def align_active_post(device):
m = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", str(bounds))
if m:
left, t, r, b = map(int, m.groups())
bounds_str = f"[{left},{t}][{r},{b}]"
else:
logger.warning(f"📐 [Alignment] Could not parse bounds: {bounds}")
continue
@@ -184,6 +194,7 @@ def align_active_post(device):
h = info.get("displayHeight", 2400)
if t > h * 0.85:
logger.debug(f"📐 [Alignment] Rejecting node at y={t} (too low, likely bottom bar)")
failed_bounds.add(bounds_str)
continue
header_y = (t + b) // 2

View File

@@ -54,10 +54,14 @@ class TelepathicEngine:
# ──────────────────────────────────────────────
def find_best_node(
self, xml_string: str, intent_description: str, device=None, track: bool = True, **kwargs
self,
xml_string: str,
intent_description: str,
device=None,
track: bool = True,
exclude_bounds: list[str] = None,
**kwargs,
) -> Optional[dict]:
print("FIND_BEST_NODE CALLED")
"""
Public facade for resolving a node.
Translates Android UI bounds into standard GramAddict node dicts.
@@ -73,6 +77,14 @@ class TelepathicEngine:
# 2. Extract interactable candidates
candidates = self._parser.get_clickable_nodes(root)
if exclude_bounds:
filtered_candidates = []
for c in candidates:
bounds_str = f"[{c.x1},{c.y1}][{c.x2},{c.y2}]"
if bounds_str not in exclude_bounds:
filtered_candidates.append(c)
candidates = filtered_candidates
# 3. Resolve intent against candidates
best_node = self._resolver.resolve(intent_description, candidates, device=device)
@@ -80,6 +92,16 @@ class TelepathicEngine:
logger.warning(f"No viable nodes found for intent: '{intent_description}'")
return None
# 3.1 BUG 7 Fix: Semantic Guard for 'post media content'
intent_lower = intent_description.lower()
semantic_str = (
(best_node.text or "") + " " + (best_node.content_desc or "") + " " + (best_node.resource_id or "")
).lower()
if "post media content" in intent_lower:
if "follow" in semantic_str.replace("_", " "):
logger.warning("🚫 [SpatialEngine] VLM selected a 'Follow' button for 'post media content'. Blocked.")
return None
# 3.5 Following Button Guard
if "follow" in intent_description.lower() and "unfollow" not in intent_description.lower():
semantic = (

View File

@@ -1,4 +1,6 @@
import logging
import json
import os
import random
from time import sleep
@@ -95,6 +97,62 @@ def get_value(count, name, default=0):
return default
_LEARNED_AD_MARKERS_FILE = os.path.join(os.getcwd(), "learned_ad_markers.json")
_LEARNED_AD_MARKERS_CACHE = None
def get_learned_ad_markers() -> set:
global _LEARNED_AD_MARKERS_CACHE
if _LEARNED_AD_MARKERS_CACHE is not None:
return _LEARNED_AD_MARKERS_CACHE
if os.path.exists(_LEARNED_AD_MARKERS_FILE):
try:
with open(_LEARNED_AD_MARKERS_FILE, "r") as f:
_LEARNED_AD_MARKERS_CACHE = set(json.load(f))
except Exception as e:
logger.error(f"Failed to load learned ad markers: {e}")
_LEARNED_AD_MARKERS_CACHE = set()
else:
_LEARNED_AD_MARKERS_CACHE = set()
return _LEARNED_AD_MARKERS_CACHE
def learn_ad_marker(marker: str, xml_hierarchy: str):
global _LEARNED_AD_MARKERS_CACHE
if not marker or len(marker) > 30:
return
marker = marker.strip().lower()
# Structural verification: the VLM-suggested marker MUST exist as an exact node text/desc in the current UI!
import xml.etree.ElementTree as ET
try:
root = ET.fromstring(xml_hierarchy)
found_in_ui = False
for node in root.iter("node"):
text = node.attrib.get("text", "").strip().lower()
desc = node.attrib.get("content-desc", "").strip().lower()
if text == marker or desc == marker:
found_in_ui = True
break
if not found_in_ui:
logger.debug(f"🧠 [Autonomous FSD] Rejected hallucinated Ad marker '{marker}' (not found as exact node match in UI).")
return
except Exception:
return
markers = get_learned_ad_markers()
if marker not in markers and marker not in {"ad", "sponsored", "advertisement", "gesponsert", "anzeige", "werbung"}:
markers.add(marker)
logger.info(f"🧠 [Autonomous FSD] Verified and Learned new Ad marker: '{marker}'. Persisting for zero-latency detection.", extra={"color": f"{Style.BRIGHT}{Fore.GREEN}"})
try:
with open(_LEARNED_AD_MARKERS_FILE, "w") as f:
json.dump(list(markers), f)
except Exception as e:
logger.error(f"Failed to save learned ad markers: {e}")
def is_ad(xml_hierarchy: str, cognitive_stack: dict = None) -> bool:
"""
Checks if the current view contains an advertisement using autonomous learning.
@@ -125,26 +183,34 @@ def is_ad(xml_hierarchy: str, cognitive_stack: dict = None) -> bool:
# Standalone label patterns: match only when the text/desc IS the ad marker,
# not when "ad" appears inside longer phrases like "Create messaging ad"
AD_EXACT_LABELS = {"ad", "sponsored", "advertisement", "gesponsert", "anzeige", "werbung"}
AD_EXACT_LABELS.update(get_learned_ad_markers())
try:
root = ET.fromstring(xml_hierarchy)
# Check if we are in a feed (to prevent false positives on profiles with 'Ad Tools' buttons)
from GramAddict.core.perception.feed_analysis import FEED_MARKERS
in_feed = any(marker in xml_hierarchy for marker in FEED_MARKERS)
for node in root.iter("node"):
attrib = node.attrib
content_desc = attrib.get("content-desc", "")
text = attrib.get("text", "")
res_id = attrib.get("resource-id", "")
# Structural check (Instagram specific)
# Structural check (Instagram specific) is always trusted
if any(marker_id in res_id for marker_id in AD_RESOURCE_IDS):
return True
# Exact label match: only trigger when the entire text/desc
# IS an ad marker (e.g. text="Ad", content-desc="Sponsored")
# This prevents false positives from "Create messaging ad"
if text.strip().lower() in AD_EXACT_LABELS:
return True
if content_desc.strip().lower() in AD_EXACT_LABELS:
return True
# We ONLY trust this if we are actually in a feed, to prevent triggering
# on the "Ad Tools" / "Ad" buttons present on business profiles.
if in_feed:
if text.strip().lower() in AD_EXACT_LABELS:
return True
if content_desc.strip().lower() in AD_EXACT_LABELS:
return True
except Exception:
pass

View File

@@ -0,0 +1,499 @@
"""
🔴 RED Phase — Production Bug Regression Tests
================================================
Evidence: Production runs 2026-04-29 17:50 and 18:08
These tests expose production bugs discovered in live runs:
1. ResonanceEvaluator crashes on truncated VLM JSON (NoneType.get) ✅ FIXED
2. ScreenMemoryDB stores wrong classification → self-reinforcing hallucination ✅ FIXED
3. SpatialEngine accepts semantically mismatched VLM selection (tracking)
4. ActionMemory VLM verification treats non-YES/NO JSON as hard failure ✅ FIXED
5. persona_interests always empty → VLM evaluates blindly
6. Resonance scoring ignores VLM should_like → always 0.50
7. Follow blocked on reels/explore due to action string mismatch
Each test MUST fail (RED) before any production code is touched.
"""
import argparse
import os
import pytest
from GramAddict.core.behaviors import BehaviorContext, BehaviorResult
from GramAddict.core.behaviors.resonance_evaluator import ResonanceEvaluatorPlugin
from GramAddict.core.perception.action_memory import ActionMemory
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
FIXTURE_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
def _load_fixture(name: str) -> str:
"""Load a real XML fixture file. Fails hard if missing."""
path = os.path.join(FIXTURE_DIR, name)
if not os.path.exists(path):
pytest.fail(f"MISSING REAL DUMP: '{name}' not found in {FIXTURE_DIR}", pytrace=False)
with open(path, "r") as f:
return f.read()
# ════════════════════════════════════════════════════════
# BUG 1: ResonanceEvaluator crashes on truncated VLM JSON
# ════════════════════════════════════════════════════════
class TestResonanceEvaluatorTruncatedJSON:
"""
Evidence from run 2026-04-29 17:50:45:
WARNING | Failed to evaluate post vibe: Unterminated string starting at: line 4 column 18
ERROR | 🧩 [Plugin] Error executing resonance_evaluator: 'NoneType' object has no attribute 'get'
Root cause: evaluate_post_vibe() returns None on JSON parse failure.
The caller in ResonanceEvaluatorPlugin.execute() does zero null-checking
before calling .get() on the result.
"""
def test_resonance_evaluator_survives_none_vibe_result(self, make_real_device_with_xml, monkeypatch):
"""
When evaluate_post_vibe() returns None (truncated JSON, LLM timeout, etc.),
the ResonanceEvaluator must NOT crash. It must gracefully default to a
neutral score and continue the pipeline.
"""
xml = _load_fixture("home_feed_real.xml")
device = make_real_device_with_xml(xml)
# Force visual vibe check to trigger by setting percentage to 100
args = argparse.Namespace(
visual_vibe_check_percentage=100,
interact_percentage=100,
persona_interests=["photography", "nature"],
)
from GramAddict.core.config import Config
config = Config(first_run=True)
config.args = args
config.config = {
"plugins": {
"resonance_evaluator": {"visual_vibe_check_percentage": 100},
}
}
# Stub telepathic engine that returns None (simulating truncated JSON)
class StubTelepathic:
def evaluate_post_vibe(self, device, persona_interests):
return None # <-- This is what happens when JSON parsing fails
ctx = BehaviorContext(
device=device,
configs=config,
session_state={},
cognitive_stack={"telepathic": StubTelepathic(), "resonance": None},
shared_state={},
post_data={"description": "Beautiful sunset"},
)
plugin = ResonanceEvaluatorPlugin()
# This MUST NOT raise AttributeError: 'NoneType' object has no attribute 'get'
result = plugin.execute(ctx)
assert isinstance(result, BehaviorResult), "Plugin must return a BehaviorResult, not crash"
assert "res_score" in ctx.shared_state, "Plugin must set res_score even on VLM failure"
# ════════════════════════════════════════════════════════
# BUG 2: ScreenMemoryDB poisoning → OWN_PROFILE hallucination
# ════════════════════════════════════════════════════════
class TestScreenMemoryPoisoning:
"""
Evidence from run 2026-04-29 17:50:47:
DEBUG | DEBUG LLM PAYLOAD: response='OWN_PROFILE', thinking=''
INFO | 🧠 [ScreenMemory] Learned new layout mapping: OWN_PROFILE
INFO | 🧠 [ScreenMemory] Cache Hit! Screen recognized as: OWN_PROFILE (Score: 1.00)
WARNING | 🚫 [GOAP] Cannot 'tap like button' on own_profile
Root cause: _classify_screen (screen_identity.py:196) has:
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab:
The `and not selected_tab` condition means that when a post is opened FROM
the home feed (where feed_tab remains selected), POST_DETAIL is NEVER detected.
The method falls through to `if selected_tab == "feed_tab": return HOME_FEED`.
Then if the LLM hallucinates OWN_PROFILE, it gets stored in Qdrant and
poisons all subsequent classifications via cache hits.
"""
def test_post_detail_detected_even_when_feed_tab_selected(self):
"""
When post_detail structural markers are present (row_feed_button_like,
row_feed_photo_profile_name), the screen MUST be classified as POST_DETAIL
even when feed_tab is selected (which is the norm for posts opened from feed).
Currently line 196 has `and not selected_tab` which blocks this detection.
"""
si = ScreenIdentity(bot_username="testuser")
xml = _load_fixture("post_detail_real.xml")
result = si.identify(xml)
assert result["screen_type"] == ScreenType.POST_DETAIL, (
f"post_detail_real.xml has row_feed_button_like + row_feed_photo_profile_name "
f"but was classified as {result['screen_type']}. The 'and not selected_tab' "
f"condition on line 196 prevents POST_DETAIL detection when feed_tab is selected."
)
# ════════════════════════════════════════════════════════
# BUG 3: ActionMemory VLM verification treats JSON as failure
# ════════════════════════════════════════════════════════
class TestActionMemoryVLMVerificationGarbage:
"""
Evidence from run 2026-04-29 17:51:01:
DEBUG | DEBUG LLM PAYLOAD: response='{ "intent": "tap post username", ... } { "}'
WARNING | ⚠️ [ActionMemory] VLM visual verification FAILED for 'tap post username'. VLM replied: '...'
WARNING | ❌ [ActionMemory] Click failed for 'tap post username'. Applying penalty.
Root cause: The VLM returned a JSON object instead of "YES"/"NO".
The YES/NO check (line 192) treats ANY non-YES response as hard failure,
even when the JSON content actually confirms success.
"""
def test_verify_success_does_not_hard_fail_on_json_response(self, make_real_device_with_xml, monkeypatch):
"""
When the VLM returns a JSON response (instead of YES/NO), verify_success()
must NOT treat it as a hard failure. It should attempt to parse the JSON
and check for success indicators.
Currently, the code on line 192 of action_memory.py does:
if response and "yes" in response.lower() and "no" not in response.lower():
This fails for any JSON response, causing false negative penalties.
"""
from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
xml = _load_fixture("home_feed_real.xml")
# Need TWO XMLs: pre-click and post-click (different to trigger UI change detection)
xml_post = xml.replace("Home", "Profile of user123")
device = make_real_device_with_xml([xml, xml_post])
# Stub the VLM to return JSON instead of YES/NO (exactly what happened in production)
vlm_json_response = (
'{ "intent": "tap post username", ' "\"element_tapped\": \"text: 'View Profile', desc: 'View Profile'\" }"
)
# Monkeypatch _query_vlm on the CLASS so any instance picks it up
monkeypatch.setattr(
SemanticEvaluator,
"_query_vlm",
lambda self, prompt, screenshot: vlm_json_response,
)
# Also ensure device.get_screenshot_b64 returns something so VLM path fires
monkeypatch.setattr(
type(device),
"get_screenshot_b64",
lambda self: "fake_base64_screenshot_data",
)
memory = ActionMemory()
from GramAddict.core.perception.spatial_parser import SpatialNode
node = SpatialNode(
text="View Profile",
content_desc="View Profile",
resource_id="com.instagram.android:id/context_menu_item",
bounds=(100, 200, 300, 400),
clickable=True,
)
memory.track_click("tap post username", node, xml)
# Call verify_success with low confidence (triggers VLM branch)
result = memory.verify_success(
intent="tap post username",
pre_click_xml=xml,
post_click_xml=xml_post,
device=device,
confidence=0.0,
)
# BUG: The VLM returned valid JSON acknowledging the intent. The YES/NO
# parser treats this as failure because JSON doesn't contain "yes".
# This causes a false-negative penalty on a correct action.
assert result is not False, (
f"verify_success() returned {result} (hard failure) for a VLM JSON response "
f"that actually acknowledges the intent. The YES/NO check on line 192 "
f"must be made more robust to handle structured VLM responses."
)
# ════════════════════════════════════════════════════════
# BUG 4: ScreenIdentity Qdrant cache ordering vulnerability
# ════════════════════════════════════════════════════════
class TestScreenIdentityCacheOrdering:
"""
The _classify_screen method in screen_identity.py has TWO critical bugs:
BUG A: Line 196 has `and not selected_tab` which prevents POST_DETAIL detection
when any tab is selected (which is ALWAYS the case for posts opened from feed).
BUG B: Line 188-194 checks Qdrant cache BEFORE the structural POST_DETAIL heuristic.
Combined with BUG A, this means the LLM fallback fires, potentially hallucinates,
and the hallucination is permanently cached in Qdrant.
"""
def test_post_detail_not_misclassified_as_home_feed(self):
"""
The post_detail_real.xml fixture has:
- row_feed_button_like (POST_DETAIL structural marker)
- row_feed_photo_profile_name (POST_DETAIL structural marker)
- feed_tab selected=true (because the post was opened FROM the home feed)
Current code returns HOME_FEED because:
1. Line 196 `and not selected_tab` blocks POST_DETAIL
2. Line 216 `if selected_tab == "feed_tab"` catches it as HOME_FEED
This is the ROOT CAUSE of the OWN_PROFILE poisoning:
when the structural check fails, the LLM fallback fires and hallucinates.
"""
si = ScreenIdentity(bot_username="testuser")
xml = _load_fixture("post_detail_real.xml")
result = si.identify(xml)
# This fixture has explicit POST_DETAIL markers. It must NOT be HOME_FEED.
assert result["screen_type"] != ScreenType.HOME_FEED, (
"post_detail_real.xml was classified as HOME_FEED. "
"The 'and not selected_tab' condition on line 196 prevents POST_DETAIL "
"detection when feed_tab is selected, causing misclassification."
)
assert result["screen_type"] == ScreenType.POST_DETAIL, f"Expected POST_DETAIL but got {result['screen_type']}"
# ════════════════════════════════════════════════════════
# BUG 5: persona_interests is ALWAYS empty
# ════════════════════════════════════════════════════════
class TestResonancePersonaInterestsEmpty:
"""
Evidence from run 2026-04-29 18:10:48:
INFO | 👁️ [Vision Core] Evaluating post vibe against:
(empty — no interests passed!)
Root cause: resonance_evaluator.py:52 reads:
persona_interests = getattr(ctx.configs.args, "persona_interests", [])
But the config has "mission.target_audience""persona_interests" doesn't exist
in the config schema. Always falls back to [].
The VLM prompt says "You are a user with the following interests: ." → blind eval.
"""
def test_persona_interests_are_not_empty_when_target_audience_set(self):
"""
When config has mission.target_audience set, the ResonanceEvaluator
must pass those interests to the VLM.
"""
import argparse
from GramAddict.core.config import Config
config = Config(first_run=True)
config.args = argparse.Namespace(
visual_vibe_check_percentage=100,
interact_percentage=100,
target_audience="travel, landscape, nature, mountain photography",
persona_interests="",
)
# The ResonanceEvaluator should extract persona interests
raw_interests = getattr(config.args, "persona_interests", "")
if not raw_interests:
raw_interests = getattr(config.args, "target_audience", "")
persona_interests = [i.strip() for i in raw_interests.split(",") if i.strip()]
assert len(persona_interests) == 4, (
f"persona_interests is {persona_interests!r} (empty or wrong). "
f"The config has mission.target_audience='travel, landscape, nature, "
f"mountain photography' but this is never wired into persona_interests."
)
# ════════════════════════════════════════════════════════
# BUG 6: Resonance scoring ignores VLM should_like response
# ════════════════════════════════════════════════════════
class TestResonanceShouldLikeFieldMismatch:
"""
Evidence from run 2026-04-29 18:11:38:
VLM response: {"should_like": true, "should_comment": false, "reasoning": "..."}
But: 📊 [Resonance] Post Score: 0.50 ← didn't change!
"""
def test_resonance_score_reflects_should_like_true(self):
"""
When VLM returns should_like=true, the resonance score must increase.
"""
vlm_response = {
"should_like": True,
"should_comment": False,
"reasoning": "Beautiful mountain landscape matching travel interests",
}
# The new code check:
should_like = vlm_response.get("should_like", False)
vibe_score = 1.0 if should_like else 0.2
assert vibe_score > 0.50, (
f"vibe_score is {vibe_score} even though should_like=True. " f"It must read 'should_like' instead."
)
# ════════════════════════════════════════════════════════
# BUG 10: Follow blocked on REELS_FEED (action string mismatch)
# ════════════════════════════════════════════════════════
class TestFollowBlockedOnReelsFeed:
"""
Evidence from run 2026-04-29 18:10:59:
WARNING | 🚫 [GOAP] Cannot 'tap 'Follow' button' on reels_feed
('tap follow button' not available on this screen)
Root cause: screen_identity.py:312-313 adds:
actions.append("tap 'Follow' button") ← with quotes
But q_nav_graph.py:141 checks for:
"follow": "tap follow button" ← without quotes
"tap 'Follow' button" != "tap follow button" → Follow is NEVER available.
"""
def test_follow_action_string_matches_nav_graph_check(self):
"""
The action string for follow in available_actions must match
what q_nav_graph.do() checks for. Currently there's a string mismatch:
screen_identity adds "tap 'Follow' button" but nav_graph checks "tap follow button".
"""
si = ScreenIdentity(bot_username="testuser")
xml = _load_fixture("reels_feed_real.xml")
result = si.identify(xml)
available = result["available_actions"]
# q_nav_graph.do() checks: "tap follow button" in available
# (see q_nav_graph.py:141)
assert "tap follow button" in available, (
f"'tap follow button' not in available_actions: {available}. "
f"The screen_identity adds \"tap 'Follow' button\" (with quotes) "
f"but q_nav_graph checks for 'tap follow button' (without quotes). "
f"This string mismatch blocks ALL follows on reels/explore."
)
# ════════════════════════════════════════════════════════
# BUG 7: SpatialEngine blocks 'Follow' buttons for 'post media content'
# ════════════════════════════════════════════════════════
class TestBug7FollowButtonGuard:
def test_follow_button_blocked(self):
"""
When the intent is 'post media content', TelepathicEngine.find_best_node
must reject nodes that have 'follow' in their semantic string.
"""
from GramAddict.core.telepathic_engine import TelepathicEngine
tele = TelepathicEngine.get_instance()
# Simulate a parse tree returning a Follow button
class DummyParser:
def parse(self, xml):
return True
def get_clickable_nodes(self, root):
from GramAddict.core.perception.spatial_parser import SpatialNode
node = SpatialNode("node1", 0, 0, 100, 100)
node.text = "Follow"
node.content_desc = "Follow User"
node.clickable = True
return [node]
class DummyResolver:
def resolve(self, intent, candidates, device=None):
return candidates[0] if candidates else None
tele._parser = DummyParser()
tele._resolver = DummyResolver()
result = tele.find_best_node("<xml/>", "post media content", track=False)
assert result is None, "TelepathicEngine should block 'Follow' button for 'post media content' intent!"
# ════════════════════════════════════════════════════════
# BUG 8: PerfectSnapping Bounds Exclusion
# ════════════════════════════════════════════════════════
class TestBug8PerfectSnappingBoundsExclusion:
def test_exclude_bounds_filters_candidates(self):
"""
TelepathicEngine.find_best_node must filter out candidates whose bounds
match those in the `exclude_bounds` list.
"""
from GramAddict.core.telepathic_engine import TelepathicEngine
tele = TelepathicEngine.get_instance()
class DummyParser:
def parse(self, xml):
return True
def get_clickable_nodes(self, root):
from GramAddict.core.perception.spatial_parser import SpatialNode
# Create two nodes with correct constructor args
node1 = SpatialNode(
bounds=(10, 10, 50, 50),
node_id="1",
class_name="",
text="Candidate 1",
content_desc="",
resource_id="",
clickable=True,
scrollable=False,
)
node2 = SpatialNode(
bounds=(100, 100, 150, 150),
node_id="2",
class_name="",
text="Candidate 2",
content_desc="",
resource_id="",
clickable=True,
scrollable=False,
)
return [node1, node2]
class DummyResolver:
def resolve(self, intent, candidates, device=None):
# Just return the first available candidate to see which survived
return candidates[0] if candidates else None
tele._parser = DummyParser()
tele._resolver = DummyResolver()
# Without exclusion, Candidate 1 should be picked
result1 = tele.find_best_node("<xml/>", "test intent", track=False)
assert result1 is not None and result1["text"] == "Candidate 1"
# Exclude Candidate 1 bounds: "[10,10][50,50]"
result2 = tele.find_best_node("<xml/>", "test intent", track=False, exclude_bounds=["[10,10][50,50]"])
assert result2 is not None and result2["text"] == "Candidate 2", "Candidate 1 was not excluded properly!"

View File

@@ -0,0 +1,49 @@
import pytest
import os
import json
from GramAddict.core.utils import is_ad, learn_ad_marker, get_learned_ad_markers
@pytest.fixture(autouse=True)
def clean_ad_markers():
# Clean up any existing learned markers file before and after tests
file_path = os.path.join(os.getcwd(), "learned_ad_markers.json")
if os.path.exists(file_path):
os.remove(file_path)
import GramAddict.core.utils
GramAddict.core.utils._LEARNED_AD_MARKERS_CACHE = None
yield
if os.path.exists(file_path):
os.remove(file_path)
GramAddict.core.utils._LEARNED_AD_MARKERS_CACHE = None
def test_learn_ad_marker_validates_against_xml():
xml_dump = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy>
<node class="android.widget.TextView" text="Sponsorisé" resource-id="com.instagram.android:id/some_id" content-desc=""/>
</hierarchy>
"""
# Attempt to learn a hallucinated marker
learn_ad_marker("Hallucination", xml_dump)
assert "hallucination" not in get_learned_ad_markers()
# Attempt to learn an actual marker present in XML
learn_ad_marker("Sponsorisé", xml_dump)
assert "sponsorisé" in get_learned_ad_markers()
def test_is_ad_uses_learned_markers():
xml_dump = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy>
<node class="android.widget.TextView" text="Sponsorisé" resource-id="com.instagram.android:id/some_id" content-desc=""/>
<node resource-id="row_feed_photo_profile_name" text="someone" />
</hierarchy>
"""
# Initially, it shouldn't recognize "Sponsorisé" because it's not in the hardcoded list
assert is_ad(xml_dump) is False
# Learn the new marker
learn_ad_marker("Sponsorisé", xml_dump)
# Now, is_ad should return True immediately without VLM
assert is_ad(xml_dump) is True