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

purge: remove ALL hardcoded UI markers + model defaults from SAE

Browse Source 
TDD cycle (RED → GREEN):

SAE perceive() — Zero Maintenance:
- Removed ALL hardcoded marker tuples (instagram_modal_markers,
  creation_flow_markers, dismiss_button_patterns, blocked_markers)
- Removed ALL hardcoded model names (llava:latest, qwen3.5:latest)
- Removed ALL hardcoded URLs (localhost:11434)
- Removed naked except blocks with model fallback defaults

New autonomous flow (zero hardcoded UI identifiers):
1. Package-based foreign app detection (Android-level, not app-specific)
2. Qdrant semantic cache (O(1) recall of learned screen types)
3. ScreenIdentity structural delegation (MODAL/FOREIGN_APP)
4. LLM autonomous classification (first-encounter learning then cached)

GOAP smart UI polling:
- Replaced static random.uniform(1.6, 2.8) sleep with MAX_POLLS=5
  polling loop that detects actual UI transitions

Model config centralized via _get_model_config():
- ALL model/URL lookups go through Config() SSOT
- Fail Fast - no silent hardcoded fallbacks

Tests: 9 new, 119 passing, 0 regressions
This commit is contained in:
Marc Mintel
2026-05-04 00:07:16 +02:00
parent 32731ed7ec
commit c98e2caaa1
18 changed files with 335 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
GramAddict/core/goap.py
View File

@@ -388,12 +388,22 @@ class GoalExecutor:
# Execute click
self.device.click(obj=best_node)
import random
time.sleep(random.uniform(1.6, 2.8))
# Verify success via Goal Context + Screen Feedback
post_xml = self.device.dump_hierarchy()
# ── Smart UI Stabilization Poll ──
# Instead of a static sleep (which is either too long on fast devices or
# too short on slow ones), we poll dump_hierarchy multiple times.
# As soon as the XML changes, we know the UI has transitioned.
MAX_POLLS = 5
POLL_INTERVAL = 0.5 # seconds between polls
post_xml = xml_dump # Start with pre-click state
for poll in range(MAX_POLLS):
time.sleep(POLL_INTERVAL)
post_xml = self.device.dump_hierarchy()
if post_xml != xml_dump:
logger.debug(f"[GOAP Poll] UI change detected on poll {poll + 1}/{MAX_POLLS}.")
break
else:
logger.debug(f"[GOAP Poll] No UI change after {MAX_POLLS} polls ({MAX_POLLS * POLL_INTERVAL}s).")
pre_action_screen = self.perceive(xml_dump) # Screen state BEFORE the click
post_screen = self.perceive(post_xml)
post_screen_type = post_screen["screen_type"]

269
GramAddict/core/situational_awareness.py
View File

@@ -291,45 +291,53 @@ class SituationalAwarenessEngine:
stable = re.sub(r"Battery \d+ per cent", "Battery NN per cent", stable)
return hashlib.sha256(stable.encode()).hexdigest()[:32]
def _get_model_config(self):
"""
Get model/URL config from Config() — the SINGLE source of truth.
Crashes loudly if config is unavailable (Fail Fast, no silent defaults).
"""
from GramAddict.core.config import Config
cfg = Config()
args = cfg.args
return {
"model": getattr(args, "ai_model", None),
"url": getattr(args, "ai_model_url", None),
"telepathic_model": getattr(args, "ai_telepathic_model", None),
"telepathic_url": getattr(args, "ai_telepathic_url", None),
}
def perceive(self, xml_dump: str) -> SituationType:
"""
Fast structural classification — NO LLM needed for perception.
Uses package names + structural markers to classify.
Autonomous situation classification — ZERO hardcoded UI element identifiers.
Flow:
1. Empty/invalid XML → FOREIGN_APP
2. Hardware check (screen off) → LOCKED_SCREEN
3. Package-based detection (app-agnostic, zero maintenance):
- Permission controller packages → OBSTACLE_SYSTEM
- App package missing → OBSTACLE_FOREIGN_APP
4. Qdrant semantic cache → instant recall of learned screen types
5. ScreenIdentity structural delegation → if MODAL, return OBSTACLE_MODAL
6. LLM classification fallback → autonomous discovery of new obstacle types
NO hardcoded resource-ids, NO hardcoded button texts, NO localized strings.
The bot discovers and learns ALL obstacles autonomously via the LLM + Qdrant loop.
"""
if not xml_dump or not isinstance(xml_dump, str):
return SituationType.OBSTACLE_FOREIGN_APP
blocked_markers = [
"try again later",
"action blocked",
"restrict certain activity",
"help us confirm you own",
"confirm it's you",
"später erneut versuchen",
"bestätige, dass du es bist",
"handlung blockiert",
"eingeschränkt",
]
# Guard: Check if the text matches are relatively isolated (e.g. short strings).
# If the string is buried inside a 200-character caption, it's a false positive.
# We can regex match text="..." attributes that are less than 60 characters total,
# OR just use the compressed string where text is capped at 60 chars anyway.
compressed_lower = self._compress_xml(xml_dump).lower()
if any(re.search(rf"(?:text|desc)='[^']*?{m}[^']*?'", compressed_lower) for m in blocked_markers):
# To be extra safe against false positives, check if there's a dialog/modal container
if "dialog" in compressed_lower or "bottom_sheet" in compressed_lower or "alert" in compressed_lower:
return SituationType.DANGER_ACTION_BLOCKED
# ── Hardware Guard: Screen Off / Locked ──
if not getattr(self.device.deviceV2, "info", {}).get("screenOn", True):
logger.info("📱 [SAE Perceive] Screen is physically OFF.")
return SituationType.OBSTACLE_LOCKED_SCREEN
# ── System Dialog / Permission Detect (Fast Path) ──
packages = set(re.findall(r'package=["\']([^"\']+)["\']', xml_dump))
# ── System Dialog / Permission Detect (package-based, app-agnostic) ──
packages = set(re.findall(r'package=["\'](.[^"\']+)["\']+', xml_dump))
app_id = getattr(self.device, "app_id", "com.instagram.android")
# Permission controller packages are Android system-level — not app-specific.
# These will NEVER change with an Instagram update. Completely safe to check.
system_dialog_pkgs = {
"com.google.android.permissioncontroller",
"com.android.permissioncontroller",
@@ -339,46 +347,24 @@ class SituationalAwarenessEngine:
logger.info("📱 [SAE Perceive] System permission dialog explicitly detected.")
return SituationType.OBSTACLE_SYSTEM
# ── Foreign Environment Detection (package-based) ──
# If the main app package is completely absent from the UI hierarchy,
# OR if there's a dominant foreign package and no app package, we might have lost the app.
# If our app is on screen, we trust we are in the app (even if a custom keyboard is open).
# We only trigger foreign app classification if our app is completely missing from the screen.
is_foreign = False
if packages and app_id not in packages:
is_foreign = True
# ── Foreign Environment Detection (package-based, zero maintenance) ──
# If our app package is completely absent from the screen → foreign app.
# Package names are Android-level identifiers, NOT Instagram UI elements.
is_foreign = bool(packages) and app_id not in packages
if is_foreign:
# ── Tier 1: Known Foreign Packages (O(1) — ZERO LLM) ──
# Production bug 2026-05-03: Play Store was detected via slow LLM path.
# For these well-known packages, a set lookup is instant and infallible.
KNOWN_FOREIGN_PACKAGES = {
"com.android.vending", # Play Store
"com.android.chrome", # Chrome
"com.google.android.chrome", # Chrome (Google build)
"com.google.android.youtube", # YouTube
"org.mozilla.firefox", # Firefox
"com.opera.browser", # Opera
"com.brave.browser", # Brave
"com.microsoft.emmx", # Edge
"com.sec.android.app.sbrowser", # Samsung Browser
}
# Any package that isn't our app AND isn't just systemui = foreign
dominant_pkgs = packages - {"com.android.systemui"}
fast_match = dominant_pkgs & KNOWN_FOREIGN_PACKAGES
if fast_match:
logger.info(
f"🚨 [SAE Perceive] Known foreign package: {fast_match} → "
f"OBSTACLE_FOREIGN_APP (O(1) fast-path, no LLM needed)"
)
if dominant_pkgs:
logger.info(f"🚨 [SAE Perceive] Foreign package detected: {dominant_pkgs} → OBSTACLE_FOREIGN_APP")
return SituationType.OBSTACLE_FOREIGN_APP
# ── Tier 2: Unknown/Ambiguous Packages → LLM Classification ──
# Only SystemUI-only or rare custom packages reach this path.
# SystemUI-only edge case: could be lock screen, notification shade, etc.
# Use LLM for disambiguation.
try:
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
cfg = self._get_model_config()
screen_off = not getattr(self.device.deviceV2, "info", {}).get("screenOn", True)
prompt = (
@@ -391,17 +377,9 @@ class SituationalAwarenessEngine:
f"XML:\n{self._compress_xml(xml_dump)[:2500]}"
)
args = {}
try:
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")
res = query_telepathic_llm(
model=model,
url=url,
model=cfg["model"],
url=cfg["url"],
system_prompt="Strict JSON classifier.",
user_prompt=prompt,
use_local_edge=True,
@@ -412,140 +390,86 @@ class SituationalAwarenessEngine:
situ_str = data.get("situation", "")
if situ_str == "OBSTACLE_LOCKED_SCREEN":
logger.info("🧠 [Smart Perceive] SystemUI definitively classified as: LOCKED_SCREEN.")
logger.info("🧠 [Smart Perceive] SystemUI classified as: LOCKED_SCREEN.")
return SituationType.OBSTACLE_LOCKED_SCREEN
elif situ_str == "OBSTACLE_SYSTEM":
logger.info("🧠 [Smart Perceive] SystemUI definitively classified as: SYSTEM_DIALOG.")
logger.info("🧠 [Smart Perceive] SystemUI classified as: SYSTEM_DIALOG.")
return SituationType.OBSTACLE_SYSTEM
else:
logger.info("🧠 [Smart Perceive] SystemUI classified as: FOREIGN_APP / NOTIFICATION.")
logger.info("🧠 [Smart Perceive] SystemUI classified as: FOREIGN_APP.")
return SituationType.OBSTACLE_FOREIGN_APP
except Exception as e:
logger.warning(f"⚠️ [Smart Perceive] LLM Classification failed ({e}). Defaulting to FOREIGN_APP.")
return SituationType.OBSTACLE_FOREIGN_APP
# ── Modal/Obstacle Detection (Autonomous LLM + Memory) ──
# We explicitly query ScreenMemoryDB. If unknown, we ask the LLM.
# This replaces ALL brittle string/ID matching for modals.
# ── In-App Obstacle Detection (100% autonomous, ZERO hardcoded UI identifiers) ──
# The bot learns ALL obstacle types via the LLM + Qdrant feedback loop.
# First encounter: LLM classifies → result is cached in Qdrant.
# All subsequent encounters: instant O(1) recall from cache.
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
compressed = self._compress_xml(xml_dump)
# ── Priority 1: Qdrant Semantic Cache (O(1), zero LLM calls) ──
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
elif cached_type == "DANGER_ACTION_BLOCKED":
return SituationType.DANGER_ACTION_BLOCKED
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
"creation_flow", # Post creation wizard
"reel_camera", # Reel recording interface
)
# ── Priority 2: ScreenIdentity structural delegation ──
# ScreenIdentity classifies the screen using its own structural logic.
# If it says MODAL → trust it (it uses the same zero-trust approach).
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
# Guard: Use the RAW xml_dump to avoid truncation of root containers (Z-index filtering),
# but ensure we only match inside resource-id attributes to prevent false positives from user text.
if any(
re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE) for marker in creation_flow_markers
):
logger.info("🧠 [SAE Perceive] Content-creation overlay detected structurally → OBSTACLE_MODAL")
screen_id = ScreenIdentity(getattr(self.device, "bot_username", ""))
screen_result = screen_id.identify(xml_dump)
screen_type = screen_result.get("screen_type", ScreenType.UNKNOWN)
if screen_type == ScreenType.MODAL:
logger.info("🧠 [SAE Perceive] ScreenIdentity classified as MODAL → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
# ── Structural Fast-Check: Instagram-Internal Modal Overlays ──
# Surveys, rating prompts, and interstitial modals live INSIDE Instagram's
# package but block normal interaction. They share a common structural
# pattern: a container resource-id containing "survey", "interstitial",
# or "nux_" (new-user-experience), plus dismiss buttons ("Not Now").
# Detecting them structurally is O(1) and eliminates LLM hallucination risk.
instagram_modal_markers = (
"survey_overlay_container", # "How are you enjoying Instagram?" survey
"survey_title", # Survey title text view
"interstitial_container", # Generic interstitial blocker
"mystery_interstitial", # Unknown/dynamic interstitials
"nux_overlay", # New-user-experience onboarding modals
"rating_prompt", # App Store rating prompt
"feedback_dialog", # Feedback collection dialogs
)
if any(
re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE)
for marker in instagram_modal_markers
):
logger.info("🧠 [SAE Perceive] Instagram modal overlay detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
if screen_type == ScreenType.FOREIGN_APP:
logger.info("🧠 [SAE Perceive] ScreenIdentity classified as FOREIGN_APP → OBSTACLE_FOREIGN_APP")
return SituationType.OBSTACLE_FOREIGN_APP
# Fallback heuristic: detect modals by dismiss-button text patterns.
# If we see "Not Now" or "Take Survey" as button text inside Instagram, it's a modal.
# Guard: match ONLY inside short text attributes (< 40 chars) to avoid caption false positives.
dismiss_button_patterns = (
r'text="Not Now"',
r'text="not now"',
r'text="Nicht jetzt"', # German: "Not Now"
r'text="Take Survey"',
r'text="rate \d+ stars?"', # "rate 5 stars"
r'text="Bewerten"', # German: "Rate"
)
has_dismiss_button = any(re.search(p, xml_dump, re.IGNORECASE) for p in dismiss_button_patterns)
if has_dismiss_button:
# Cross-validate: must also have a container that looks like a dialog/overlay
# (not just a random "Not Now" text in a DM thread or post caption)
has_overlay_structure = bool(
re.search(
r'resource-id="[^"]*(?:overlay|dialog|interstitial|survey|sheet|prompt)[^"]*"',
xml_dump,
re.IGNORECASE,
)
or re.search(r'resource-id="[^"]*button_(?:negative|positive)[^"]*"', xml_dump, re.IGNORECASE)
)
if has_overlay_structure:
logger.info("🧠 [SAE Perceive] Instagram dismiss-button modal detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
# If ScreenIdentity positively identified a known screen type (not UNKNOWN),
# we trust it as NORMAL — no LLM needed.
if screen_type != ScreenType.UNKNOWN:
screen_memory.store_screen(compressed, "NORMAL")
return SituationType.NORMAL
# If not cached, query LLM for autonomous structural classification
# ── Priority 3: LLM autonomous classification (first-encounter learning) ──
# This is the ONLY path that reaches the LLM. After classification,
# the result is cached in Qdrant — so this screen type is learned forever.
try:
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
cfg = self._get_model_config()
prompt = (
"You are a Situation Classifier for a mobile automation agent.\n"
"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"
"An 'Add to story' screen, camera interface, 'quick_capture' layout, gallery picker, "
"or ANY content-creation flow (reel recording, post editor, live mode) is an OBSTACLE_MODAL — "
"it blocks normal navigation and must be dismissed.\n"
"Respond ONLY with a valid JSON object strictly matching this schema: "
'{"situation": "OBSTACLE_MODAL" | "NORMAL"}\n\n'
"Analyze the given Android UI XML dump AND screenshot. Classify the screen into one of:\n"
"- OBSTACLE_MODAL: Any blocking overlay, dialog, popup, survey, rating prompt, "
"browser window, camera/creation flow, or any UI that blocks normal feed browsing.\n"
"- DANGER_ACTION_BLOCKED: Instagram's rate-limit or action-block warning "
"(e.g. 'Try Again Later', 'Action Blocked', or any restriction/verification screen).\n"
"- NORMAL: A standard usable screen (feed, explore, profile, DM, etc.)\n\n"
"Respond ONLY with a valid JSON object: "
'{"situation": "OBSTACLE_MODAL" | "DANGER_ACTION_BLOCKED" | "NORMAL"}\n\n'
f"XML:\n{compressed[:2500]}"
)
args = {}
try:
args = Config().args
except Exception:
pass
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,
model=cfg["telepathic_model"],
url=cfg["telepathic_url"],
system_prompt="Strict JSON classifier.",
user_prompt=prompt,
images_b64=[screenshot_b64] if screenshot_b64 else None,
@@ -560,6 +484,10 @@ class SituationalAwarenessEngine:
logger.info("🧠 [Smart Perceive] Screen classified as: OBSTACLE_MODAL.")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
elif situ_str == "DANGER_ACTION_BLOCKED":
logger.info("🧠 [Smart Perceive] Screen classified as: DANGER_ACTION_BLOCKED.")
screen_memory.store_screen(compressed, "DANGER_ACTION_BLOCKED")
return SituationType.DANGER_ACTION_BLOCKED
else:
logger.info("🧠 [Smart Perceive] Screen classified as: NORMAL.")
screen_memory.store_screen(compressed, "NORMAL")
@@ -589,16 +517,11 @@ class SituationalAwarenessEngine:
LLM-powered escape planning for situations where structural scan fails.
Called ONLY when recall AND structural planning both miss.
"""
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
try:
args = Config().args
model = getattr(args, "ai_telepathic_model", "llava:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
except Exception:
model = "llava:latest"
url = "http://localhost:11434/api/generate"
cfg = self._get_model_config()
model = cfg["telepathic_model"]
url = cfg["telepathic_url"]
system_prompt = (
"You are an Android UI navigation agent. Your job is to escape obstacles "