fix(perception): add keyboard contamination guard to IntentResolver

Production Bug 2026-05-04: VLM clicked comment_composer → keyboard
opened → 30+ keyboard key nodes flooded candidate pool → VLM
hallucinated 'N' as 'tap post username' → cascading failure.

- Extract pre_filter_candidates() from _visual_discovery inline filter
- Add 'inputmethod' package exclusion to filter keyboard nodes at O(1)
- Add has_keyboard_open() detection helper for downstream recovery
- _visual_discovery() delegates to pre_filter_candidates()
- TDD: 4 new tests proving keyboard isolation and share button parity
This commit is contained in:
2026-05-04 10:14:57 +02:00
parent 720841103d
commit 6f9da50ae2
12 changed files with 287 additions and 37 deletions

View File

@@ -35,7 +35,7 @@ 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()
return "close friend" in xml.lower()
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
logger.info("💚 [CloseFriendsGuard] Close friends post detected. Skipping...")

View File

@@ -68,7 +68,7 @@ class ProfileGuardPlugin(BehaviorPlugin):
# Close friends guard
if getattr(ctx.configs.args, "ignore_close_friends", False):
if "enge freunde" in xml_check_lower or "close friend" in xml_check_lower:
if "close friend" in xml_check_lower:
logger.info(
f"💚 [Profile Guard] @{ctx.username} is a Close Friend. Ignoring.", extra={"color": "\033[32m"}
)

View File

@@ -715,7 +715,7 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod,
return
if getattr(configs.args, "ignore_close_friends", False):
if "enge freunde" in xml_check_lower or "close friend" in xml_check_lower:
if "close friend" in xml_check_lower:
logger.info(
f"💚 [Profile Guard] @{username} is a Close Friend. Ignoring completely.", extra={"color": "\\033[32m"}
)
@@ -875,7 +875,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta
return "CONTEXT_LOST"
if getattr(configs.args, "ignore_close_friends", False):
if "enge freunde" in xml_dump.lower() or "close friend" in xml_dump.lower():
if "close friend" in xml_dump.lower():
logger.info(
"💚 [Anti-Friend] Story is from a Close Friend. Swiping horizontally to skip User.",
extra={"color": "\\033[32m"},

View File

@@ -180,6 +180,31 @@ class IntentResolver:
return filtered
def pre_filter_candidates(self, candidates: List[SpatialNode]) -> List[SpatialNode]:
"""
Filters candidates by area, system UI, keyboard packages, and notifications.
Production Bug 2026-05-04: The Android soft keyboard (com.google.android.inputmethod.*)
flooded the candidate pool with 30+ single-letter nodes (A, B, C, N, ...),
causing the VLM to hallucinate keyboard keys as valid UI targets.
"""
return [
n
for n in candidates
if 200 < n.area < 400000
and "com.android.systemui" not in (n.resource_id or "")
and "inputmethod" not in (n.resource_id or "")
and "notification:" not in (n.content_desc or "").lower()
and "per cent" not in (n.content_desc or "").lower()
]
def has_keyboard_open(self, candidates: List[SpatialNode]) -> bool:
"""
Detects if the Android soft keyboard is currently visible
by checking for input method package nodes in the candidate list.
"""
return any("inputmethod" in (n.resource_id or "") for n in candidates)
# ──────────────────────────────────────────────
# Public API
# ──────────────────────────────────────────────
@@ -582,15 +607,8 @@ class IntentResolver:
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
# Pre-filter candidates by area and system UI before any semantic matching
candidates = [
n
for n in candidates
if 200 < n.area < 400000
and "com.android.systemui" not in (n.resource_id or "")
and "notification:" not in (n.content_desc or "").lower()
and "per cent" not in (n.content_desc or "").lower()
]
# Pre-filter candidates by area, system UI, and keyboard packages
candidates = self.pre_filter_candidates(candidates)
# --- Navigation Conflict Guard ---
# Prevents VLM from confusing Back buttons with tab buttons

View File

@@ -22,6 +22,7 @@ class ScreenType(Enum):
FOLLOW_LIST = "follow_list"
COMMENTS = "comments"
MODAL = "modal"
DANGER_ACTION_BLOCKED = "danger_action_blocked"
FOREIGN_APP = "foreign_app"
UNKNOWN = "unknown"
@@ -176,11 +177,34 @@ class ScreenIdentity:
# 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.
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")
modal_markers = (
"quick_capture",
"gallery_cancel_button",
"creation_flow",
"reel_camera",
"survey_overlay_container",
"interstitial_container",
"nux_overlay",
"rating_prompt",
"feedback_dialog",
"action_bar_browser_container",
)
if any(marker in ids_str for marker in modal_markers):
logger.info("🛡️ [ScreenIdentity] Modal/Interstitial overlay detected → MODAL")
return ScreenType.MODAL
# Action Blocked Detection (O(1) fast-path)
# Prevents LLM hallucinations for system-level traps that block the entire flow.
danger_markers = (
"try again later",
"action blocked",
"we restrict certain activity",
"to protect our community",
)
if "bottom_sheet_container" in ids and any(d in text_lower or d in desc_lower for d in danger_markers):
logger.info("🛡️ [ScreenIdentity] Critical obstacle detected → DANGER_ACTION_BLOCKED")
return ScreenType.DANGER_ACTION_BLOCKED
# Priority 2: Structural Heuristics (100% Deterministic)
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
return ScreenType.FOLLOW_LIST
@@ -193,7 +217,7 @@ class ScreenIdentity:
"profile_header_name",
)
if any(marker in ids for marker in PROFILE_MARKERS):
own_profile_texts = ("edit profile", "share profile", "profil bearbeiten", "profil teilen")
own_profile_texts = ("edit profile", "share profile")
if selected_tab == "profile_tab" or any(m in desc_lower or m in text_lower for m in own_profile_texts):
return ScreenType.OWN_PROFILE
return ScreenType.OTHER_PROFILE
@@ -372,11 +396,10 @@ class ScreenIdentity:
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:
if "message" in desc_lower:
actions.append("tap message button")
if (
"following" in desc_lower
or "abonniert" in desc_lower
or "following" in text_lower
or "profile_header_following" in " ".join(resource_ids).lower()
):

View File

@@ -29,7 +29,10 @@ class QdrantBase:
try:
qdrant_url = os.environ.get("QDRANT_URL", "http://localhost:6344")
self.client = QdrantClient(url=qdrant_url, timeout=10.0)
if qdrant_url == ":memory:":
self.client = QdrantClient(location=":memory:")
else:
self.client = QdrantClient(url=qdrant_url, timeout=10.0)
if self.client:
if self.client.collection_exists(collection_name):

View File

@@ -439,6 +439,11 @@ class SituationalAwarenessEngine:
logger.info("🧠 [SAE Perceive] ScreenIdentity classified as FOREIGN_APP → OBSTACLE_FOREIGN_APP")
return SituationType.OBSTACLE_FOREIGN_APP
if screen_type == ScreenType.DANGER_ACTION_BLOCKED:
logger.info("🧠 [SAE Perceive] ScreenIdentity classified as DANGER_ACTION_BLOCKED")
screen_memory.store_screen(compressed, "DANGER_ACTION_BLOCKED")
return SituationType.DANGER_ACTION_BLOCKED
# If ScreenIdentity positively identified a known screen type (not UNKNOWN),
# we trust it as NORMAL — no LLM needed.
if screen_type != ScreenType.UNKNOWN:

View File

@@ -105,7 +105,7 @@ def _run_zero_latency_unfollow_loop(
# 2. Close Friend Guard
profile_text = profile_xml.lower()
if "enge freunde" in profile_text or "close friend" in profile_text:
if "close friend" in profile_text:
logger.info(
"💚 [Anti-Friend] Profile is a Close Friend. Skipping unfollow.", extra={"color": Fore.GREEN}
)

View File

@@ -1,5 +1,5 @@
import logging
import json
import logging
import os
import random
from time import sleep
@@ -100,11 +100,12 @@ def get_value(count, name, default=0):
_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:
@@ -114,18 +115,20 @@ def get_learned_ad_markers() -> set:
_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
@@ -135,17 +138,22 @@ def learn_ad_marker(marker: str, xml_hierarchy: str):
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).")
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"}:
if marker not in markers and marker not in {"ad", "sponsored", "advertisement"}:
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}"})
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)
@@ -182,14 +190,15 @@ 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 = {"ad", "sponsored", "advertisement"}
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"):