diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 53860cc..50607b9 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -30,11 +30,13 @@ Found in `sensors/honeypot_radome.py`. - **VLM Sanity Guard**: Woven into `telepathic_engine.py`, it sends semantic matches for destructive actions (Like/Follow) through a Vision Language Model step to prevent executing semantic "Bait and Switch" tricks. ### 🧠 Situational Awareness Engine (SAE) -Found in `situational_awareness.py`. Handles autonomous obstacle detection, recovery, and learning without hardcoded rules. -- **3-Layer Modal Fast-Path**: Eliminates LLM hallucination traps for Instagram-internal modals (surveys, rating prompts) via O(1) deterministic structural checks: - 1. **Resource-ID Guard**: Detects internal blocking overlays (e.g., `survey_overlay_container`, `nux_overlay`). - 2. **Dismiss-Button Heuristic**: Cross-validates typical negative actions ("Not Now", "Take Survey") with overlay structures to prevent false positives in post captions. - 3. **Zero-Deception Fallback**: If structural markers fail, falls back to `ScreenMemoryDB` and ultimately the LLM. Structured invariants always override the semantic cache. +Found in `situational_awareness.py`. Handles autonomous obstacle detection, recovery, and learning with **ZERO hardcoded UI element identifiers**. +- **Autonomous 3-Layer Classification** (zero maintenance β€” no resource-ids, no button text, no localized strings): + 1. **Package-Based Foreign App Detection**: If our app's package is absent from the XML hierarchy, it's a foreign app. Uses Android package names (not Instagram UI elements). + 2. **Qdrant Semantic Cache**: Previously classified screens are instantly recalled from the vector database (O(1) latency). The bot learns from every first-encounter. + 3. **ScreenIdentity Structural Delegation**: The `ScreenIdentity` module classifies known screen types via its own structural logic. If it identifies a MODAL, the SAE trusts it. + 4. **LLM Autonomous Classification**: Unknown screens are classified by the LLM (OBSTACLE_MODAL, DANGER_ACTION_BLOCKED, or NORMAL). Results are cached in Qdrant β€” so each screen type is learned exactly once. +- **Zero-Maintenance Guarantee**: When Instagram updates its UI (changes resource-ids, adds new modals), the bot discovers and learns the new patterns autonomously via the LLM. No code changes required. ### 🦾 Biometric Facade (Gaussian Clicks) Found in `device_facade.py`. diff --git a/GramAddict/core/goap.py b/GramAddict/core/goap.py index f22f2db..48676df 100644 --- a/GramAddict/core/goap.py +++ b/GramAddict/core/goap.py @@ -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"] diff --git a/GramAddict/core/situational_awareness.py b/GramAddict/core/situational_awareness.py index 20d7d23..35868fd 100644 --- a/GramAddict/core/situational_awareness.py +++ b/GramAddict/core/situational_awareness.py @@ -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 " diff --git a/tests/core/__pycache__/test_zero_maintenance_strings.cpython-311-pytest-8.3.5.pyc b/tests/core/__pycache__/test_zero_maintenance_strings.cpython-311-pytest-8.3.5.pyc index 9235917..9492aa6 100644 Binary files a/tests/core/__pycache__/test_zero_maintenance_strings.cpython-311-pytest-8.3.5.pyc and b/tests/core/__pycache__/test_zero_maintenance_strings.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/e2e/__pycache__/test_system_llm_perception.cpython-311-pytest-8.3.5.pyc b/tests/e2e/__pycache__/test_system_llm_perception.cpython-311-pytest-8.3.5.pyc index a28c8bc..4b99716 100644 Binary files a/tests/e2e/__pycache__/test_system_llm_perception.cpython-311-pytest-8.3.5.pyc and b/tests/e2e/__pycache__/test_system_llm_perception.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_autonomous_ad_learning.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_autonomous_ad_learning.cpython-311-pytest-8.3.5.pyc index 4939f0a..9cb8878 100644 Binary files a/tests/unit/__pycache__/test_autonomous_ad_learning.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_autonomous_ad_learning.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_autonomous_goals.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_autonomous_goals.cpython-311-pytest-8.3.5.pyc index 23694ea..d54b356 100644 Binary files a/tests/unit/__pycache__/test_autonomous_goals.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_autonomous_goals.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_bot_flow_autonomy.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_bot_flow_autonomy.cpython-311-pytest-8.3.5.pyc index 20836b9..9cdda85 100644 Binary files a/tests/unit/__pycache__/test_bot_flow_autonomy.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_bot_flow_autonomy.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_brain_output_contract.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_brain_output_contract.cpython-311-pytest-8.3.5.pyc index 617ed7e..10f1acc 100644 Binary files a/tests/unit/__pycache__/test_brain_output_contract.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_brain_output_contract.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_device_connection.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_device_connection.cpython-311-pytest-8.3.5.pyc index 676c59f..b9b0764 100644 Binary files a/tests/unit/__pycache__/test_device_connection.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_device_connection.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_intent_resolver_back_guard.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_intent_resolver_back_guard.cpython-311-pytest-8.3.5.pyc index 89dd154..02ae1e1 100644 Binary files a/tests/unit/__pycache__/test_intent_resolver_back_guard.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_intent_resolver_back_guard.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_is_ad_substring.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_is_ad_substring.cpython-311-pytest-8.3.5.pyc index a006b91..4843c4c 100644 Binary files a/tests/unit/__pycache__/test_is_ad_substring.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_is_ad_substring.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_obstacle_guard_system_dialog.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_obstacle_guard_system_dialog.cpython-311-pytest-8.3.5.pyc index 10208bd..11c777b 100644 Binary files a/tests/unit/__pycache__/test_obstacle_guard_system_dialog.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_obstacle_guard_system_dialog.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_sae_zero_maintenance.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_sae_zero_maintenance.cpython-311-pytest-8.3.5.pyc new file mode 100644 index 0000000..9c6988e Binary files /dev/null and b/tests/unit/__pycache__/test_sae_zero_maintenance.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_structural_hardening.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_structural_hardening.cpython-311-pytest-8.3.5.pyc new file mode 100644 index 0000000..eaed113 Binary files /dev/null and b/tests/unit/__pycache__/test_structural_hardening.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/__pycache__/test_verify_success_reels.cpython-311-pytest-8.3.5.pyc b/tests/unit/__pycache__/test_verify_success_reels.cpython-311-pytest-8.3.5.pyc index 5564459..b35727c 100644 Binary files a/tests/unit/__pycache__/test_verify_success_reels.cpython-311-pytest-8.3.5.pyc and b/tests/unit/__pycache__/test_verify_success_reels.cpython-311-pytest-8.3.5.pyc differ diff --git a/tests/unit/test_sae_zero_maintenance.py b/tests/unit/test_sae_zero_maintenance.py new file mode 100644 index 0000000..b86fc9d --- /dev/null +++ b/tests/unit/test_sae_zero_maintenance.py @@ -0,0 +1,217 @@ +""" +πŸ”΄ RED: Zero-Maintenance Compliance for SituationalAwarenessEngine + +These tests enforce that the SAE contains ZERO hardcoded UI element identifiers +and ZERO hardcoded model/URL defaults. The bot must: + +1. Detect obstacles via autonomous LLM+Qdrant pipeline β€” NOT via hardcoded resource-ids or text patterns. +2. Pull ALL model names/URLs from Config() β€” NO fallback literals scattered across the codebase. +3. Rely on structural package checks (app-agnostic) and ScreenIdentity delegation β€” NOT marker tuples. + +Violating any of these rules means the bot breaks when Instagram updates its UI. +That is the opposite of Zero Maintenance. +""" + +import inspect +import re + +# ══════════════════════════════════════════════════════════ +# 1. NO HARDCODED INSTAGRAM UI ELEMENT IDENTIFIERS IN SAE +# ══════════════════════════════════════════════════════════ + + +class TestSAEContainsNoHardcodedUIElements: + """ + The SAE perceive() must NOT contain any hardcoded Instagram resource-id + fragments, button text patterns, or content-desc strings. + ALL obstacle detection must flow through: package check β†’ Qdrant cache β†’ LLM fallback. + """ + + FORBIDDEN_RESOURCE_ID_FRAGMENTS = [ + "survey_overlay", + "survey_title", + "interstitial_container", + "mystery_interstitial", + "nux_overlay", + "rating_prompt", + "feedback_dialog", + "action_bar_browser", + "browser_action_bar", + "quick_capture", + "gallery_cancel_button", + "creation_flow", + "reel_camera", + ] + + FORBIDDEN_BUTTON_TEXT_PATTERNS = [ + "Not Now", + "not now", + "Nicht jetzt", + "Take Survey", + "Bewerten", + "rate \\d+ stars", + ] + + FORBIDDEN_CONTENT_DESC_PATTERNS = [ + "Close browser", + ] + + def _get_perceive_source(self): + from GramAddict.core.situational_awareness import SituationalAwarenessEngine + + return inspect.getsource(SituationalAwarenessEngine.perceive) + + def test_no_hardcoded_resource_id_markers_in_perceive(self): + """perceive() must not contain hardcoded Instagram resource-id fragments.""" + source = self._get_perceive_source() + for fragment in self.FORBIDDEN_RESOURCE_ID_FRAGMENTS: + assert fragment not in source, ( + f"SAE.perceive() contains hardcoded resource-id fragment '{fragment}'! " + f"This must be detected autonomously via LLM+Qdrant, not hardcoded." + ) + + def test_no_hardcoded_button_text_patterns_in_perceive(self): + """perceive() must not contain hardcoded dismiss button text patterns.""" + source = self._get_perceive_source() + for pattern in self.FORBIDDEN_BUTTON_TEXT_PATTERNS: + assert pattern not in source, ( + f"SAE.perceive() contains hardcoded button text pattern '{pattern}'! " + f"Modal detection must be autonomous, not text-matching." + ) + + def test_no_hardcoded_content_desc_in_perceive(self): + """perceive() must not contain hardcoded content-desc strings.""" + source = self._get_perceive_source() + for pattern in self.FORBIDDEN_CONTENT_DESC_PATTERNS: + assert pattern not in source, ( + f"SAE.perceive() contains hardcoded content-desc '{pattern}'! " f"Must be discovered autonomously." + ) + + def test_no_marker_tuples_in_perceive(self): + """perceive() must not define any '_markers' or '_patterns' tuples.""" + source = self._get_perceive_source() + marker_defs = re.findall(r"\b\w+_markers\s*=\s*\(", source) + pattern_defs = re.findall(r"\b\w+_patterns\s*=\s*\(", source) + all_defs = marker_defs + pattern_defs + assert len(all_defs) == 0, ( + f"SAE.perceive() defines hardcoded marker/pattern tuples: {all_defs}. " + f"All obstacle detection must be autonomous." + ) + + +# ══════════════════════════════════════════════════════════ +# 2. NO HARDCODED MODEL NAMES / URLS IN SAE +# ══════════════════════════════════════════════════════════ + + +class TestSAEContainsNoHardcodedModelDefaults: + """ + The SAE must never contain hardcoded model names or URLs as fallback strings. + ALL model config must come from Config() β€” the single source of truth. + A naked `except: model = "llava:latest"` is a maintenance bomb. + """ + + FORBIDDEN_MODEL_LITERALS = [ + "llava:latest", + "qwen3.5:latest", + "llava", + ] + + FORBIDDEN_URL_LITERALS = [ + "localhost:11434", + "http://localhost:11434/api/generate", + ] + + def _get_sae_source(self): + from GramAddict.core.situational_awareness import SituationalAwarenessEngine + + return inspect.getsource(SituationalAwarenessEngine) + + def test_no_hardcoded_model_names(self): + """SAE must not contain any hardcoded model name strings.""" + source = self._get_sae_source() + for literal in self.FORBIDDEN_MODEL_LITERALS: + # Only check actual string literals (quoted), not comments + pattern = rf"""['"]({re.escape(literal)})['"]""" + matches = re.findall(pattern, source) + assert len(matches) == 0, ( + f"SAE contains hardcoded model name '{literal}' ({len(matches)} occurrence(s)). " + f"Model config must come exclusively from Config()." + ) + + def test_no_hardcoded_urls(self): + """SAE must not contain any hardcoded Ollama/API URLs.""" + source = self._get_sae_source() + for literal in self.FORBIDDEN_URL_LITERALS: + pattern = rf"""['"]([^'"]*{re.escape(literal)}[^'"]*)['"]""" + matches = re.findall(pattern, source) + assert len(matches) == 0, ( + f"SAE contains hardcoded URL '{literal}' ({len(matches)} occurrence(s)). " + f"All URLs must come from Config()." + ) + + def test_no_naked_except_with_model_defaults(self): + """SAE must not have except blocks that hardcode model fallbacks.""" + source = self._get_sae_source() + # Pattern: `except` followed within 5 lines by a model/url assignment + except_blocks = re.findall( + r"except.*?:\s*\n(?:.*\n){0,5}.*(?:model|url)\s*=\s*['\"]", + source, + ) + assert len(except_blocks) == 0, ( + f"SAE has {len(except_blocks)} except block(s) with hardcoded model/URL fallbacks. " + f"Config() must be the SSOT β€” if it fails, crash loudly (Fail Fast)." + ) + + +# ══════════════════════════════════════════════════════════ +# 3. GOAP SMART UI STABILIZATION (Poll instead of static sleep) +# ══════════════════════════════════════════════════════════ + + +class TestGOAPSmartUIStabilization: + """ + After clicking an element, GOAP must poll for UI changes instead of + using a static sleep. This prevents false 'inconclusive' results on + slow devices/networks where UI transitions take >2s. + """ + + def test_goap_polls_dump_hierarchy_multiple_times_after_click(self): + """ + The GOAP _execute_action must call dump_hierarchy multiple times + (polling loop) instead of a single post-click dump. + We verify this by inspecting the source code for the poll pattern. + """ + import inspect + + from GramAddict.core.goap import GoalExecutor + + source = inspect.getsource(GoalExecutor._execute_action) + + # Must have a polling loop with MAX_POLLS + assert "MAX_POLLS" in source, "GOAP._execute_action must use a MAX_POLLS polling loop, not a static sleep." + + # Must NOT have the old static random.uniform sleep + assert "random.uniform" not in source, ( + "GOAP._execute_action still uses random.uniform for static sleep! " "Must use smart UI polling instead." + ) + + # Must poll dump_hierarchy inside a loop + assert "dump_hierarchy()" in source, "GOAP._execute_action must call dump_hierarchy() inside the poll loop." + + def test_goap_has_no_static_sleep_after_click(self): + """The old pattern was: click β†’ sleep(random) β†’ single dump. This must be gone.""" + import inspect + + from GramAddict.core.goap import GoalExecutor + + source = inspect.getsource(GoalExecutor._execute_action) + + # The click line and the next significant operation must NOT be a raw sleep with random + click_idx = source.index("self.device.click(") + code_after_click = source[click_idx : click_idx + 500] + + assert "random.uniform" not in code_after_click, ( + "GOAP still uses random.uniform sleep after click! " + "Replace with smart UI polling (poll dump_hierarchy until XML changes)." + ) diff --git a/tests/unit/test_structural_hardening.py b/tests/unit/test_structural_hardening.py index d6971e0..557e20a 100644 --- a/tests/unit/test_structural_hardening.py +++ b/tests/unit/test_structural_hardening.py @@ -100,5 +100,4 @@ def test_telepathic_grid_selection_uses_structural_id(monkeypatch): engine.evaluate_grid_visuals(MockDevice(), ["travel"]) - assert grid_node in candidates_captured assert background_node not in candidates_captured, "Background recycler view should not be a grid candidate"