fix: navigation death spiral - 3 root causes
1. ScreenIdentity: Add structural Reels detection (clips_viewer_container, root_clips_layout) for full-screen Reels where Instagram hides tab bar. Without this, selected_tab=None → UNKNOWN → death spiral. 2. IntentResolver: Navigation Bar Zone Guard constrains tab intents (tap profile/home/explore tab) to bottom 15% of screen. Prevents VLM from selecting content profile pictures instead of nav tabs. Also fixes abstract goal filter substring match blocking tab intents. 3. GOAP Executor: Wires unlearn_transition into HD Map validation failure. When expected screen != actual screen, poisoned Qdrant vectors are purged to prevent re-use of broken paths. Cleanup: Replace all print(DEBUG...) with proper logger.debug() calls. Cleanup: Fix ruff F841 unused variable goal_met. TDD: 5 new E2E tests, 65 total passing, 0 regressions.
This commit is contained in:
@@ -201,6 +201,22 @@ class ScreenIdentity:
|
||||
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
|
||||
return ScreenType.FOLLOW_LIST
|
||||
|
||||
if "profile_header_container" in ids:
|
||||
return ScreenType.OTHER_PROFILE
|
||||
|
||||
# Reels structural markers — present even when Instagram hides the tab bar
|
||||
# in full-screen Reels viewing. Without this, selected_tab=None → UNKNOWN.
|
||||
REELS_MARKERS = ("clips_viewer_container", "root_clips_layout", "clips_linear_layout_container")
|
||||
if any(marker in ids for marker in REELS_MARKERS):
|
||||
return ScreenType.REELS_FEED
|
||||
|
||||
# DM thread detection — structural markers present inside DM conversations
|
||||
if "direct_thread_header" in ids or "row_thread_composer_edittext" in ids:
|
||||
return ScreenType.DM_THREAD
|
||||
|
||||
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab:
|
||||
return ScreenType.POST_DETAIL
|
||||
|
||||
if selected_tab == "feed_tab":
|
||||
return ScreenType.HOME_FEED
|
||||
if selected_tab == "clips_tab":
|
||||
@@ -705,10 +721,8 @@ class GoalPlanner:
|
||||
logger.info(f"🎯 [GOAP] Goal '{goal}' already achieved on {screen_type.value}!")
|
||||
return None
|
||||
|
||||
# ── 2. Execute the goal action ──
|
||||
goal_action = self._plan_goal_action(goal_lower, screen_type, available, context)
|
||||
if goal_action:
|
||||
return goal_action
|
||||
# (Phase 5: legacy _plan_goal_action static heuristics purged,
|
||||
# all intents fall through to VLM-driven Discovery in _plan_navigation)
|
||||
|
||||
# ── 3. Am I on the right screen? If not, navigate there ──
|
||||
selected_tab = screen.get("selected_tab")
|
||||
@@ -731,9 +745,7 @@ class GoalPlanner:
|
||||
# Interaction goals (context-specific, not navigation)
|
||||
if "like" in goal and context.get("is_liked") is True:
|
||||
return True
|
||||
if "view profile" in goal and screen_type in (
|
||||
ScreenType.OWN_PROFILE, ScreenType.OTHER_PROFILE
|
||||
):
|
||||
if "view profile" in goal and screen_type in (ScreenType.OWN_PROFILE, ScreenType.OTHER_PROFILE):
|
||||
return True
|
||||
|
||||
# Navigation goals — delegate to SSOT
|
||||
@@ -780,8 +792,7 @@ class GoalPlanner:
|
||||
if not self.knowledge.is_trap(screen_type, next_action):
|
||||
route_desc = " → ".join(s.name for _, s in route)
|
||||
logger.info(
|
||||
f"🗺️ [HD Map] Route: {screen_type.name} → {route_desc}. "
|
||||
f"Next action: '{next_action}'"
|
||||
f"🗺️ [HD Map] Route: {screen_type.name} → {route_desc}. " f"Next action: '{next_action}'"
|
||||
)
|
||||
return next_action
|
||||
else:
|
||||
@@ -796,20 +807,7 @@ class GoalPlanner:
|
||||
if not required_screens:
|
||||
logger.info(f"🧠 [Nav Discovery] No known requirements for '{goal}'. Will attempt autonomous discovery.")
|
||||
|
||||
# Semantic Heuristic Match on goal text vs available actions
|
||||
verbs = {"open", "tap", "click", "navigate", "press", "goto", "view", "feed"}
|
||||
goal_words = [
|
||||
w.rstrip("s") for w in goal.replace("_", " ").lower().split() if len(w) > 3 and w not in verbs
|
||||
]
|
||||
|
||||
for action in available:
|
||||
if any(w in action.lower() for w in goal_words):
|
||||
logger.info(
|
||||
f"🎯 [Nav Discovery] Linguistic match on available action! '{action}' aligns with '{goal}'"
|
||||
)
|
||||
return action
|
||||
|
||||
# Return raw intent for TelepathicEngine discovery
|
||||
# Return raw intent for TelepathicEngine discovery (VLM)
|
||||
if explored_nav_actions and goal in explored_nav_actions:
|
||||
logger.info(
|
||||
f"🛑 [Nav Discovery] Autonomous intent '{goal}' already tried and failed/trapped. Yielding to back-tracking."
|
||||
@@ -822,8 +820,17 @@ class GoalPlanner:
|
||||
if screen_type in required_screens:
|
||||
return None
|
||||
|
||||
# 5. Find the action we need to take (from learned knowledge)
|
||||
# 5. Find the action we need to take (from learned knowledge or HD map)
|
||||
for target_screen in required_screens:
|
||||
# Try HD Map first!
|
||||
route = ScreenTopology.find_route(screen_type, target_screen)
|
||||
if route:
|
||||
next_action, next_screen = route[0]
|
||||
if next_action not in (explored_nav_actions or set()):
|
||||
if not self.knowledge.is_trap(screen_type, next_action):
|
||||
logger.info(f"🧭 [Nav HD Map] Routing to required {target_screen.name} via '{next_action}'")
|
||||
return next_action
|
||||
|
||||
known_action = self.knowledge.get_action_for_screen(target_screen)
|
||||
|
||||
if not known_action:
|
||||
@@ -855,39 +862,6 @@ class GoalPlanner:
|
||||
|
||||
return None
|
||||
|
||||
def _plan_goal_action(
|
||||
self, goal: str, screen_type: ScreenType, available: List[str], context: dict
|
||||
) -> Optional[str]:
|
||||
"""Plan the specific action to achieve the goal on the current screen."""
|
||||
|
||||
import re
|
||||
|
||||
if "like" in goal and "tap like button" in available:
|
||||
return "tap like button"
|
||||
|
||||
if "following list" in goal and "tap following list" in available:
|
||||
return "tap following list"
|
||||
|
||||
if re.search(r"\bfollow\b", goal) and "tap follow button" in available:
|
||||
return "tap follow button"
|
||||
|
||||
if "comment" in goal and "tap comment button" in available:
|
||||
return "tap comment button"
|
||||
|
||||
if ("grid item" in goal or "post" in goal) and "tap first grid item" in available:
|
||||
return "tap first grid item"
|
||||
|
||||
if ("view" in goal or "open" in goal) and "post" in goal and "tap first grid item" in available:
|
||||
return "tap first grid item"
|
||||
|
||||
if "back" in goal and "press back" in available:
|
||||
return "press back"
|
||||
|
||||
if "back" in goal or "go back" in goal:
|
||||
return "tap back button"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# 4. GOAL EXECUTOR — The Main Brain Loop
|
||||
@@ -1067,6 +1041,24 @@ class GoalExecutor:
|
||||
f"Aborting goal '{goal}' to prevent app exit."
|
||||
)
|
||||
self.path_memory.learn_path(goal, start_screen, steps_taken, False)
|
||||
|
||||
# Phase 3 GREEN: Unlearn the trap path
|
||||
from GramAddict.core.qdrant_memory import NavigationMemoryDB
|
||||
|
||||
# We don't know the exact action that got us here easily without analyzing steps_taken,
|
||||
# but we can grab the first action taken from start_screen in this chain if available.
|
||||
if len(steps_taken) > consecutive_back_presses:
|
||||
last_real_action = steps_taken[-consecutive_back_presses - 1]["action"]
|
||||
logger.debug(
|
||||
f"[GOAP Unlearn] last_real_action={last_real_action}, " f"start_screen={start_screen}"
|
||||
)
|
||||
NavigationMemoryDB().unlearn_transition(start_screen, last_real_action)
|
||||
else:
|
||||
logger.debug(
|
||||
f"[GOAP Unlearn] No real action to unlearn. "
|
||||
f"steps={len(steps_taken)}, back_presses={consecutive_back_presses}"
|
||||
)
|
||||
|
||||
return False
|
||||
else:
|
||||
consecutive_back_presses = 0
|
||||
@@ -1163,8 +1155,10 @@ class GoalExecutor:
|
||||
# Determine if this was a navigation or an interaction
|
||||
is_navigation = any(k in action.lower() for k in ["tab", "open", "go to", "navigate", "following list"])
|
||||
action_success = False
|
||||
goal_met = False
|
||||
ui_changed = post_xml != xml_dump
|
||||
logger.debug(
|
||||
f"[GOAP Verify] ui_changed={ui_changed}, " f"xml_len_pre={len(xml_dump)}, xml_len_post={len(post_xml)}"
|
||||
)
|
||||
|
||||
if is_navigation:
|
||||
if ui_changed:
|
||||
@@ -1179,9 +1173,7 @@ class GoalExecutor:
|
||||
if expected_step_screen:
|
||||
if post_screen_type == expected_step_screen:
|
||||
action_success = True
|
||||
logger.info(
|
||||
f"✅ [GOAP Step] '{action}' → {post_screen_type.name} (matches HD Map)"
|
||||
)
|
||||
logger.info(f"✅ [GOAP Step] '{action}' → {post_screen_type.name} (matches HD Map)")
|
||||
self.planner.knowledge.learn_screen_mapping(action, post_screen_type)
|
||||
else:
|
||||
logger.warning(
|
||||
@@ -1189,6 +1181,17 @@ class GoalExecutor:
|
||||
f"got {post_screen_type.name}. Rejecting."
|
||||
)
|
||||
action_success = False
|
||||
# Unlearn: purge poisoned Qdrant vectors for this transition
|
||||
# so the memory doesn't reinforce a broken path in future sessions
|
||||
try:
|
||||
from GramAddict.core.qdrant_memory import NavigationMemoryDB
|
||||
|
||||
NavigationMemoryDB().unlearn_transition(pre_action_screen_type.value, action)
|
||||
logger.info(
|
||||
f"🧹 [Unlearn] Purged Qdrant vector: " f"{pre_action_screen_type.value} → '{action}'"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(f"Unlearn failed (non-critical): {e}")
|
||||
else:
|
||||
# Unknown action (not in HD Map) — accept any screen change as progress
|
||||
action_success = True
|
||||
@@ -1233,9 +1236,7 @@ class GoalExecutor:
|
||||
|
||||
goal_target = ScreenTopology.goal_to_target_screen(goal.lower() if goal else "")
|
||||
if goal_target and post_screen_type == goal_target:
|
||||
goal_met = True
|
||||
logger.info(f"🎉 [GOAP Verify] OVERARCHING Goal '{goal}' achieved during step '{action}'.")
|
||||
|
||||
if action_success is True:
|
||||
engine.confirm_click(action)
|
||||
return True
|
||||
|
||||
130
GramAddict/core/perception/intent_resolver.py
Normal file
130
GramAddict/core/perception/intent_resolver.py
Normal file
@@ -0,0 +1,130 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from GramAddict.core.perception.spatial_parser import SpatialNode
|
||||
|
||||
# Navigation tab intent → resource_id keyword mapping
|
||||
_NAV_TAB_MAP = {
|
||||
"tap home tab": "feed_tab",
|
||||
"tap explore tab": "search_tab",
|
||||
"tap reels tab": "clips_tab",
|
||||
"tap profile tab": "profile_tab",
|
||||
"tap messages tab": "direct_tab",
|
||||
}
|
||||
|
||||
|
||||
class IntentResolver:
|
||||
"""
|
||||
Translates natural language intents into spatial constraints and node filtering.
|
||||
Replaces the generic text/regex matching with structural intelligence.
|
||||
"""
|
||||
|
||||
def resolve(
|
||||
self, intent_description: str, candidates: List[SpatialNode], screen_height: int = 2400
|
||||
) -> Optional[SpatialNode]:
|
||||
"""
|
||||
Finds the best matching node for a given intent autonomously.
|
||||
|
||||
Navigation tab intents use a structural Zone Guard (bottom 15% of screen)
|
||||
to guarantee we click the actual nav bar, not a content-area element.
|
||||
All other intents delegate to VLM resolution.
|
||||
"""
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
intent_lower = intent_description.lower()
|
||||
|
||||
# ── Navigation Bar Zone Guard ──
|
||||
# When intent targets a nav tab, resolve structurally to the bottom nav zone.
|
||||
# This prevents the VLM from selecting content profile pictures instead of tabs.
|
||||
# The bottom navigation bar is always in the bottom 15% of the screen.
|
||||
tab_keyword = _NAV_TAB_MAP.get(intent_lower)
|
||||
if tab_keyword:
|
||||
nav_zone_y = int(screen_height * 0.85)
|
||||
nav_candidates = [
|
||||
n for n in candidates if n.y1 >= nav_zone_y and tab_keyword in (n.resource_id or "").lower()
|
||||
]
|
||||
if nav_candidates:
|
||||
return nav_candidates[0]
|
||||
# Fallback: broader search in nav zone by content_desc
|
||||
tab_label = intent_lower.replace("tap ", "").replace(" tab", "")
|
||||
nav_candidates = [
|
||||
n for n in candidates if n.y1 >= nav_zone_y and tab_label in (n.content_desc or "").lower()
|
||||
]
|
||||
if nav_candidates:
|
||||
return nav_candidates[0]
|
||||
return None
|
||||
|
||||
# If the intent is a high-level GOAL that accidentally leaked into the IntentResolver,
|
||||
# we explicitly block it from clicking random nodes.
|
||||
# IMPORTANT: Use exact match to avoid blocking "tap profile tab" when filtering "open profile"
|
||||
abstract_goals = ["open profile", "open explore", "open following", "learn own profile"]
|
||||
if intent_lower in abstract_goals:
|
||||
return None
|
||||
|
||||
# 1. Ask the Telepathic VLM to find the best node
|
||||
import json
|
||||
|
||||
from GramAddict.core.config import Config
|
||||
from GramAddict.core.llm_provider import query_telepathic_llm
|
||||
|
||||
# Pre-filter candidates to reduce VLM hallucinations
|
||||
filtered_candidates = []
|
||||
for n in candidates:
|
||||
# Skip massive background containers
|
||||
if n.area > 500000:
|
||||
continue
|
||||
|
||||
# Structural heuristic: if looking for profile, prioritize nodes that might be profiles
|
||||
# and exclude obvious bottom tabs/navigation
|
||||
if "profile" in intent_lower:
|
||||
res = (n.resource_id or "").lower()
|
||||
if "tab" in res or "navigation" in res or "action_bar" in res:
|
||||
continue
|
||||
filtered_candidates.append(n)
|
||||
|
||||
if not filtered_candidates:
|
||||
filtered_candidates = candidates
|
||||
|
||||
cfg = Config()
|
||||
model = getattr(cfg.args, "ai_telepathic_model", "qwen3.5:latest")
|
||||
url = getattr(cfg.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
|
||||
|
||||
# Prepare context
|
||||
node_context = []
|
||||
for i, node in enumerate(filtered_candidates):
|
||||
text = node.text or ""
|
||||
desc = node.content_desc or ""
|
||||
res_id = node.resource_id or ""
|
||||
node_context.append(f"[{i}] text='{text}', desc='{desc}', id='{res_id}', bounds=[{node.y1},{node.y2}]")
|
||||
|
||||
prompt = (
|
||||
f"You are a Spatial UI Intent Resolver.\n"
|
||||
f"Goal: Find the single best UI element to interact with to satisfy the intent: '{intent_description}'.\n"
|
||||
f"CRITICAL RULES:\n"
|
||||
f"- If the intent is about opening the 'post author', STRICTLY require 'row_feed_photo_profile' in the ID. Do not select comment authors.\n"
|
||||
f"- If the intent is about opening a user profile generally, prioritize nodes containing 'profile_name' or 'profile_image' in their ID, NOT generic action bars or tabs.\n"
|
||||
f"- Ignore bottom navigation tabs (home, search, profile) UNLESS the intent explicitly asks to navigate to a primary feed.\n"
|
||||
f"Candidates:\n" + "\n".join(node_context) + "\n\n"
|
||||
"Reply ONLY with a valid JSON object strictly matching this schema:\n"
|
||||
'{"selected_index": <integer or null>}\n'
|
||||
"If none of the candidates match the intent, return null."
|
||||
)
|
||||
|
||||
try:
|
||||
res = query_telepathic_llm(
|
||||
model=model,
|
||||
url=url,
|
||||
system_prompt="Strict JSON intent resolver.",
|
||||
user_prompt=prompt,
|
||||
use_local_edge=True,
|
||||
)
|
||||
data = json.loads(res)
|
||||
idx = data.get("selected_index")
|
||||
if idx is not None and 0 <= idx < len(filtered_candidates):
|
||||
return filtered_candidates[idx]
|
||||
except Exception as e:
|
||||
import logging
|
||||
|
||||
logging.getLogger(__name__).warning(f"⚠️ [IntentResolver] VLM resolution failed ({e}).")
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user