diff --git a/GramAddict/core/situational_awareness.py b/GramAddict/core/situational_awareness.py index 1e125bf..2c1d9ba 100644 --- a/GramAddict/core/situational_awareness.py +++ b/GramAddict/core/situational_awareness.py @@ -98,22 +98,28 @@ class SituationEpisodeDB: # Sort by confidence desc, prefer successful episodes best_positive = None - failed_actions = set() + failed_action_signatures = set() for r in results: p = r.payload + action_data = p.get("action", {}) + # Create a signature for the action: type + coordinates + sig = f"{action_data.get('action_type')}|{action_data.get('x', 0)},{action_data.get('y', 0)}" + if not p.get("success", False): - # Track failed actions so we don't repeat them - failed_actions.add(p.get("action", {}).get("action_type", "")) + # Track failed actions so we don't repeat the exact same interaction + failed_action_signatures.add(sig) continue + conf = p.get("confidence", 0.0) if conf >= 0.3 and (best_positive is None or conf > best_positive.payload.get("confidence", 0)): best_positive = r if best_positive: action_data = best_positive.payload.get("action", {}) - # Don't return an action type that has also failed before for this situation - if action_data.get("action_type") not in failed_actions: + sig = f"{action_data.get('action_type')}|{action_data.get('x', 0)},{action_data.get('y', 0)}" + + if sig not in failed_action_signatures: logger.info( f"🧠 [SAE Recall] Instant memory hit! Situation matched with confidence " f"{best_positive.payload.get('confidence', 0):.2f}. Action: {action_data.get('reason', 'unknown')}" @@ -142,16 +148,21 @@ class SituationEpisodeDB: seed = f"{situation_signature}|{action.action_type}|{action.x},{action.y}" point_id = self._db.generate_uuid(seed) + # Retrieve existing confidence if any current_conf = 0.0 has_existing = False try: points = self._db.client.retrieve( - collection_name=self._db.collection_name, ids=[point_id], with_payload=True, with_vectors=False + collection_name=self._db.collection_name, + ids=[point_id], + with_payload=True, + with_vectors=False, ) if points: has_existing = True current_conf = points[0].payload.get("confidence", 0.0) - except Exception: + except Exception as e: + logger.debug(f"SAE retrieval error (non-fatal): {e}") pass if success: @@ -160,7 +171,8 @@ class SituationEpisodeDB: confidence = current_conf - 0.5 if has_existing else -0.5 if confidence < 0.1 and not success: - self._db.client.delete(collection_name=self._db.collection_name, points_selector=[point_id]) + # Use safe wrapper to avoid crash if DB is down + self._db.delete_point(seed) logger.info("πŸ—‘οΈ [SAE Learn] Action decayed below threshold. Deleted from memory.") return @@ -340,7 +352,7 @@ class SituationalAwarenessEngine: # ── Keyboard Detection (Fast Path) ── # Instead of relying on brittle package names (which may be missing or custom), - # an open keyboard is definitively proven if an EditText is focused, + # an open keyboard is definitively proven if an EditText is focused, # or if unmistakable keyboard structural markers exist. is_actual_keyboard = False if 'focused="true"' in xml_dump and "EditText" in xml_dump: @@ -348,20 +360,18 @@ class SituationalAwarenessEngine: r'focused="true"[^>]*class="[^"]*EditText"', xml_dump ): is_actual_keyboard = True - + # Structural fallback: certain strings are exclusive to keyboards # We match these securely against content-desc or resource-id to avoid false positives from user post text. keyboard_markers = [ r'content-desc="[^"]*(?:Symboltastatur|Switch input method|Leerzeichen|Spracheingabe verwenden)[^"]*"', - r'resource-id="[^"]*input_method_nav_ime_switcher[^"]*"' + r'resource-id="[^"]*input_method_nav_ime_switcher[^"]*"', ] if not is_actual_keyboard and any(re.search(marker, xml_dump, re.IGNORECASE) for marker in keyboard_markers): is_actual_keyboard = True if is_actual_keyboard: - logger.info( - "πŸ“± [SAE Perceive] On-screen Keyboard explicitly detected. Treating as obstacle." - ) + logger.info("πŸ“± [SAE Perceive] On-screen Keyboard explicitly detected. Treating as obstacle.") return SituationType.OBSTACLE_KEYBOARD # ── Foreign Environment Detection (package-based) ── @@ -606,6 +616,72 @@ class SituationalAwarenessEngine: # 2. PLAN: AI-driven escape strategy # ────────────────────────────────────────────── + def _find_structural_dismiss_target(self, xml_dump: str) -> Optional[EscapeAction]: + """ + Structurally scan the raw XML for any clickable element that semantically + represents a dismiss/close/cancel action. Returns None if nothing found. + Zero hardcoding β€” all coordinates come from the XML itself. + """ + import re + + # Patterns for dismiss-type buttons: text or content-desc containing dismiss keywords + dismiss_keywords = ( + "cancel", + "close", + "dismiss", + "not now", + "nicht jetzt", + "abbrechen", + "schließen", + "skip", + "ΓΌberspringen", + "no thanks", + "nein danke", + "got it", + "ok", + "done", + "fertig", + ) + + # Extract each or tag individually + node_tags = re.findall(r"]*>", xml_dump, re.IGNORECASE) + + candidates = [] + for tag in node_tags: + # Extract attributes independently β€” order doesn't matter + text_m = re.search(r'text="([^"]*)"', tag) + desc_m = re.search(r'content-desc="([^"]*)"', tag) + click_m = re.search(r'clickable="([^"]*)"', tag) + bounds_m = re.search(r'bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', tag) + + text = (text_m.group(1) if text_m else "").strip().lower() + desc = (desc_m.group(1) if desc_m else "").strip().lower() + clickable = (click_m.group(1) if click_m else "").lower() + + if clickable != "true" or not bounds_m: + continue + + label = text or desc + if not label: + continue + + for kw in dismiss_keywords: + if kw in label: + x1, y1 = int(bounds_m.group(1)), int(bounds_m.group(2)) + x2, y2 = int(bounds_m.group(3)), int(bounds_m.group(4)) + cx, cy = (x1 + x2) // 2, (y1 + y2) // 2 + candidates.append((kw, cx, cy, label)) + break + + if candidates: + # Prefer exact match ("cancel" > "ok"), pick first match + best = candidates[0] + kw, cx, cy, label = best + logger.info(f"πŸ” [SAE Structural] Found dismiss target '{label}' at ({cx}, {cy})") + return EscapeAction("click", x=cx, y=cy, reason=f"Structural dismiss: '{label}'") + + return None + def _plan_escape_via_llm( self, xml_dump: str, compressed: str, situation_type: SituationType, failed_actions: set = None ) -> Optional[EscapeAction]: @@ -620,30 +696,46 @@ class SituationalAwarenessEngine: model = getattr(args, "ai_telepathic_model") url = getattr(args, "ai_telepathic_url") + # Build failure context so the LLM knows what NOT to repeat + failure_block = "" + temperature = 0.1 + if failed_actions: + failure_block = ( + "\n\nCRITICAL β€” The following actions have ALREADY BEEN TRIED and FAILED. " + "You MUST NOT propose any of these again:\n" + ) + for fa in failed_actions: + failure_block += f" - {fa}\n" + failure_block += ( + "\nYou MUST pick a DIFFERENT element or a DIFFERENT action type. " + "Study the XML carefully for buttons you have NOT tried yet.\n" + ) + # Increase temperature to force diversity when the LLM is being stubborn + temperature = min(0.7, 0.1 + 0.2 * len(failed_actions)) + system_prompt = ( - "You are an Android UI navigation agent. Your job is to escape obstacles " - "(dialogs, modals, foreign apps, system popups) and return to Instagram. " - "Analyze the screen content (Screenshot AND XML) and return a JSON escape action.\n\n" + "You are the Navigation Engine for an autonomous agent.\n" + "An obstacle has been detected that prevents normal operations.\n" + "Your goal is to DISMISS or CLEAR this obstacle to return to the app's main workflow.\n\n" + f"Current Obstacle Type: {situation_type.value}\n\n" "Rules:\n" - "- If you see a dismiss/close/cancel/skip/not now button, click it\n" - "- If the Situation type is obstacle_locked_screen, action must be 'unlock'\n" - "- If the Situation type is obstacle_foreign_app, action must be 'kill_foreign_apps'\n" - "- If the Situation type is obstacle_system, you MUST look for 'Deny', 'Don't allow', or 'Cancel' and click it. \n" - " NEVER click 'Allow', 'OK', or 'Confirm' on system permissions.\n" - " If no negative action button exists, action must be 'back'\n" - "- If the screen shows an In-App Browser, WebView, website, or Advertisement, it is an obstacle. Action must be 'back' or click the close button.\n" - "- If there is NO obstacle and the screen is a normal Instagram view (false positive), action must be 'false_positive'\n" - "- If nothing else works, suggest 'app_start' to force-reopen Instagram\n" - "- NEVER click 'OK'/'Confirm'/'Accept' on surveys or prompts\n" - "- When you choose to click, you MUST use the EXACT coordinates provided in `center=(x,y)` for that element in the XML\n" - '- Return ONLY valid JSON: {"action": "click"|"back"|"app_start"|"unlock"|"kill_foreign_apps"|"false_positive", "x": N, "y": N, "reason": "..."}' + "- If the Situation type is obstacle_modal, find and click a 'Cancel', 'Close', 'Dismiss', 'Not Now', " + "or 'X' button. Look carefully at ALL elements in the XML.\n" + "- If it is a system dialog, click 'Allow', 'OK' or 'Dismiss'.\n" + "- If the keyboard is blocking, your only action is 'back'.\n" + '- When you choose to click, you MUST use the EXACT coordinates from bounds="[x1,y1][x2,y2]" ' + "in the XML. Calculate center as ((x1+x2)/2, (y1+y2)/2). Do NOT invent coordinates.\n" + "- If no clickable dismiss element exists in the XML, use 'back'.\n" + "- Respond ONLY with valid JSON: " + '{"action": "click" | "back" | "app_start", "x": int, "y": int, "reason": "string"}' + f"{failure_block}" ) - user_prompt = f"Situation type: {situation_type.value}\n\n" f"Screen content:\n{compressed}\n\n" - if failed_actions: - user_prompt += f"Failed actions this session (DO NOT REPEAT): {list(failed_actions)}\n\n" - - user_prompt += "What action should I take to clear this obstacle and return to Instagram? Return JSON only." + user_prompt = ( + f"Situation type: {situation_type.value}\n\n" + f"Screen XML:\n{compressed}\n\n" + "What action should I take to clear this obstacle and return to Instagram? Return JSON only." + ) try: screenshot_b64 = getattr(self.device, "get_screenshot_b64", lambda: None)() @@ -654,7 +746,7 @@ class SituationalAwarenessEngine: user_prompt=user_prompt, system_prompt=system_prompt, images_b64=[screenshot_b64] if screenshot_b64 else None, - temperature=0.0, + temperature=temperature, ) if resp: import json @@ -662,7 +754,6 @@ class SituationalAwarenessEngine: try: data = json.loads(resp) except json.JSONDecodeError: - # Try extracting JSON via regex if LLM was chatty import re match = re.search(r"\{.*\}", resp, re.DOTALL) @@ -671,15 +762,42 @@ class SituationalAwarenessEngine: else: raise ValueError(f"Could not parse JSON from: {resp}") + action_type = data.get("action", "back") + x = int(data.get("x", 0)) + y = int(data.get("y", 0)) + reason = data.get("reason", "LLM-planned escape") + action_key = f"{action_type}:{x},{y}" + + # If LLM stubbornly repeats a failed action, try structural scan as autonomous fallback + if failed_actions and action_key in failed_actions: + logger.warning( + f"🧠 [SAE] LLM repeated failed action ({action_key}). " + "Falling back to structural dismiss scan." + ) + structural = self._find_structural_dismiss_target(xml_dump) + if structural: + structural_key = f"{structural.action_type}:{structural.x},{structural.y}" + if structural_key not in failed_actions: + return structural + # If structural also exhausted, escalate to back + if "back:0,0" not in (failed_actions or set()): + return EscapeAction("back", reason="Autonomous fallback: structural + LLM both exhausted") + return EscapeAction("app_start", reason="Autonomous escalation: all escape strategies exhausted") + return EscapeAction( - action_type=data.get("action", "back"), - x=int(data.get("x", 0)), - y=int(data.get("y", 0)), - reason=data.get("reason", "LLM-planned escape"), + action_type=action_type, + x=x, + y=y, + reason=reason, ) except Exception as e: logger.warning(f"🧠 [SAE] LLM escape planning failed: {e}") + # Last resort: try structural scan before giving up + structural = self._find_structural_dismiss_target(xml_dump) + if structural: + return structural + return EscapeAction("back", reason="LLM planning failed, defaulting to BACK") # ────────────────────────────────────────────── diff --git a/scripts/pre_commit_tests.sh b/scripts/pre_commit_tests.sh index 6a08122..dbb60b3 100755 --- a/scripts/pre_commit_tests.sh +++ b/scripts/pre_commit_tests.sh @@ -21,10 +21,13 @@ else # Heuristic: Try to find a matching unit test test_file="tests/unit/test_${filename}" core_test_file="tests/core/test_${filename}" + tdd_test_file="tests/tdd/test_${filename}" if [ -f "$test_file" ]; then TEST_TARGETS="$TEST_TARGETS $test_file" elif [ -f "$core_test_file" ]; then TEST_TARGETS="$TEST_TARGETS $core_test_file" + elif [ -f "$tdd_test_file" ]; then + TEST_TARGETS="$TEST_TARGETS $tdd_test_file" else # Try to find matching e2e tests by searching for each word in the module name module_name="${filename%.py}" diff --git a/tests/tdd/test_situational_awareness.py b/tests/tdd/test_situational_awareness.py new file mode 100644 index 0000000..952b54d --- /dev/null +++ b/tests/tdd/test_situational_awareness.py @@ -0,0 +1,206 @@ +""" +TDD: Structural Dismiss Target Discovery + +Proves that the SAE finds dismiss/close buttons by scanning +the XML structurally β€” zero hardcoded coordinates. +All coordinates come from bounds attributes in the XML itself. +""" + +from GramAddict.core.situational_awareness import ( + SituationalAwarenessEngine, +) + + +class DummyDeviceV2: + def __init__(self): + self.info = {"screenOn": True} + + +class DummyDevice: + def __init__(self): + self.deviceV2 = DummyDeviceV2() + self.app_id = "com.instagram.android" + self.pressed = [] + self.clicks = [] + + def press(self, key): + self.pressed.append(key) + + def click(self, x, y): + self.clicks.append((x, y)) + + +# ── XML Fixtures ── + + +BOTTOM_SHEET_WITH_CANCEL = """ + + + + + + + + +""" + +BOTTOM_SHEET_WITH_CLOSE_DESC = """ + + + + + + + +""" + +BOTTOM_SHEET_NO_DISMISS = """ + + + + + + + +""" + +NOT_NOW_DIALOG = """ + + + + + + + +""" + +GERMAN_DISMISS_DIALOG = """ + + + + + + + +""" + + +# ── Tests ── + + +class TestStructuralDismissTarget: + """Verify that _find_structural_dismiss_target discovers buttons from XML, not hardcoded values.""" + + def _get_sae(self): + device = DummyDevice() + return SituationalAwarenessEngine.get_instance(device) + + def test_finds_cancel_button_from_xml(self): + """RED: structural scan must find 'Cancel' button with XML-derived coordinates.""" + sae = self._get_sae() + result = sae._find_structural_dismiss_target(BOTTOM_SHEET_WITH_CANCEL) + + assert result is not None, "Should have found the Cancel button" + assert result.action_type == "click" + # Coordinates must come from bounds="[100,1400][980,1500]" β†’ center (540, 1450) + assert result.x == 540 + assert result.y == 1450 + assert "cancel" in result.reason.lower() + + def test_finds_close_via_content_desc(self): + """RED: structural scan must find 'Close' button via content-desc attribute.""" + sae = self._get_sae() + result = sae._find_structural_dismiss_target(BOTTOM_SHEET_WITH_CLOSE_DESC) + + assert result is not None, "Should have found the Close button via content-desc" + assert result.action_type == "click" + # bounds="[900,620][980,700]" β†’ center (940, 660) + assert result.x == 940 + assert result.y == 660 + assert "close" in result.reason.lower() + + def test_returns_none_when_no_dismiss_exists(self): + """RED: if the XML has no dismiss-type button, return None (don't invent one).""" + sae = self._get_sae() + result = sae._find_structural_dismiss_target(BOTTOM_SHEET_NO_DISMISS) + + assert result is None, "Must not invent dismiss targets when none exist in XML" + + def test_finds_not_now_button(self): + """RED: structural scan must find 'Not Now' button.""" + sae = self._get_sae() + result = sae._find_structural_dismiss_target(NOT_NOW_DIALOG) + + assert result is not None, "Should have found 'Not Now' button" + assert result.action_type == "click" + # bounds="[100,1060][980,1160]" β†’ center (540, 1110) + assert result.x == 540 + assert result.y == 1110 + + def test_finds_german_nicht_jetzt(self): + """RED: structural scan must find German 'Nicht jetzt' button.""" + sae = self._get_sae() + result = sae._find_structural_dismiss_target(GERMAN_DISMISS_DIALOG) + + assert result is not None, "Should have found 'Nicht jetzt' button" + assert result.action_type == "click" + # bounds="[100,1060][980,1160]" β†’ center (540, 1110) + assert result.x == 540 + assert result.y == 1110 + + def test_no_hardcoded_coordinates_in_method(self): + """ + META-TEST: Prove that _find_structural_dismiss_target contains + zero hardcoded pixel values. All coordinates must come from XML parsing. + """ + import inspect + + source = inspect.getsource(SituationalAwarenessEngine._find_structural_dismiss_target) + + # Should NOT contain any hardcoded coordinate patterns like x=540, y=150, etc. + # The only numbers allowed are in the regex patterns, not in EscapeAction constructors + import re + + # Find all EscapeAction instantiations with literal x= or y= values + hardcoded_coords = re.findall(r"EscapeAction\([^)]*(?:x=\d+|y=\d+)", source) + assert len(hardcoded_coords) == 0, ( + f"Found hardcoded coordinates in _find_structural_dismiss_target: {hardcoded_coords}. " + "All coordinates must come from XML bounds parsing." + ) + + +class TestLLMFallbackNoHardcoding: + """Verify that _plan_escape_via_llm does NOT use hardcoded fallback coordinates.""" + + def test_no_hardcoded_coordinates_in_llm_planner(self): + """ + META-TEST: Prove that _plan_escape_via_llm contains zero hardcoded + pixel coordinates as fallback values. When the LLM repeats, it must + delegate to _find_structural_dismiss_target (which reads from XML). + """ + import inspect + import re + + source = inspect.getsource(SituationalAwarenessEngine._plan_escape_via_llm) + + # Find all EscapeAction("click", x=, y=) with hardcoded coords + hardcoded = re.findall(r'EscapeAction\(\s*"click"\s*,\s*x=\d+\s*,\s*y=\d+', source) + assert len(hardcoded) == 0, ( + f"Found hardcoded click coordinates in _plan_escape_via_llm: {hardcoded}. " + "The bot must discover coordinates autonomously from the XML." + )