2 Commits

9 changed files with 57 additions and 55 deletions

View File

@@ -496,8 +496,12 @@ class GoalExecutor:
return False
else:
# action_success is None (INCONCLUSIVE)
# We decay the memory so it unlearns if it repeatedly fails to produce a definitive success.
logger.warning(f"⚠️ [GOAP Execute] Applying AGGRESSIVE PENALTY for inconclusive action '{action}'.")
engine.decay_click(action)
# Double penalty to burn ambiguous paths faster (outer loop adds +1, so total +2 = instantly hits MAX_RETRIES)
self.action_failures[(pre_action_screen_type, action)] = (
self.action_failures.get((pre_action_screen_type, action), 0) + 1
)
return False
def _execute_recalled_path(self, steps: List[Dict], goal: str) -> bool:

View File

@@ -156,17 +156,8 @@ class GoalPlanner:
)
return None
# ── 2. Brain-Driven Decision Making (Primary Strategy) ──
# The user explicitly wants the AI to be the primary driver of goals.
from GramAddict.core.navigation.brain import ask_brain_for_action
brain_action = ask_brain_for_action(goal, screen_type.name, available, avoid_actions)
if brain_action:
logger.info(f"🧠 [Brain] Decided to execute: '{brain_action}' (to achieve: '{goal}')")
return brain_action
# ── 2. HD Map Routing (Fallback) ──
# If the Brain doesn't know what to do, try the deterministic topological map.
# ── 2. HD Map Routing (Primary Strategy for Navigation) ──
# Ground UI transitions in structural invariants. If the topological map knows the route, use it.
target_screen = ScreenTopology.goal_to_target_screen(goal)
if target_screen and target_screen != screen_type:
route = ScreenTopology.find_route(screen_type, target_screen, avoid_actions=avoid_actions)
@@ -187,6 +178,15 @@ class GoalPlanner:
f"🛡️ [HD Map] Route action '{next_action}' already explored and failed. Skipping HD Map."
)
# ── 3. Brain-Driven Decision Making (Fallback / Discovery) ──
# For non-navigation goals or when the HD Map is incomplete.
from GramAddict.core.navigation.brain import ask_brain_for_action
brain_action = ask_brain_for_action(goal, screen_type.name, available, avoid_actions)
if brain_action:
logger.info(f"🧠 [Brain] Decided to execute: '{brain_action}' (to achieve: '{goal}')")
return brain_action
# ── 2. Learned Knowledge (Qdrant) ──
required_screens = self.knowledge.get_requirements(goal)

View File

@@ -167,7 +167,8 @@ class ActionMemory:
and "row_feed_button_like" not in post_xml_lower
and "clips_viewer" not in post_xml_lower
):
return None # Still on grid, inconclusive
logger.warning(f"⚠️ [ActionMemory] Still on grid after trying to '{intent}'. Verification FAIL.")
return False # Still on grid, definitely failed
state_toggles = ["like", "save", "follow", "heart"]
is_toggle = any(t in intent_lower for t in state_toggles)
@@ -269,7 +270,10 @@ class ActionMemory:
if diff > 0:
logger.debug(f"🧠 [ActionMemory] Structural delta detected for toggle '{intent}'. Verification PASS.")
return True
else:
logger.warning(
f"⚠️ [ActionMemory] Zero structural shift (diff={diff}) for state-toggle '{intent}'. Verification FAIL."
)
return False
# If the intent is an abstract goal (like "find customers"), diff > 50 is NOT enough.
# We must force visual VLM confirmation because clicking the wrong thing (like "Create highlight")
# also produces a large diff but achieves the wrong goal.
@@ -317,6 +321,12 @@ class ActionMemory:
logger.warning(f"⚠️ [ActionMemory] Cannot visually verify abstract intent '{intent}'. Failing safe.")
return False
# If diff <= 50 for non-toggle
logger.warning(
f"⚠️ [ActionMemory] Insufficient structural change (diff={diff}) for non-toggle '{intent}'. Verification FAIL."
)
return False
def _intent_matches_node(intent: str, semantic_string: str) -> bool:
"""Checks if the clicked element semantically matches the toggle intent.

View File

@@ -62,7 +62,6 @@ class ScreenTopology:
"press back": ScreenType.HOME_FEED,
},
ScreenType.OTHER_PROFILE: {
"press back": ScreenType.HOME_FEED,
"tap home tab": ScreenType.HOME_FEED,
},
ScreenType.UNKNOWN: {

View File

@@ -72,28 +72,26 @@ def test_curiosity_navigates_to_homefeed_before_checking(
nav_graph = cognitive_stack["nav_graph"]
zero_engine = cognitive_stack["zero_engine"]
dopamine = cognitive_stack["dopamine"]
growth = cognitive_stack["growth_brain"]
# Force CHECK_CURIOSITY
monkeypatch.setattr(growth, "evaluate_governance", lambda *args, **kwargs: "CHECK_CURIOSITY")
# We want it to exit cleanly after executing a few loops, without hanging forever.
dopamine.session_limit_seconds = 0.1
dopamine.session_start = time.time()
# 5. Run the loop (starting from other_profile)
try:
_run_zero_latency_feed_loop(
device=device,
zero_engine=zero_engine,
nav_graph=nav_graph,
configs=configs,
session_state=session_state,
job_target="homefeed",
cognitive_stack=cognitive_stack,
)
except Exception:
# It will likely crash when trying to 'tap heart icon notifications'
# because the emulator state machine doesn't define what that button does.
# This is expected and perfectly fine, because it proves the bot TRIED to
# tap it AFTER navigating.
pass
# 5. Run the loop (starting from other_profile)
_run_zero_latency_feed_loop(
device=device,
zero_engine=zero_engine,
nav_graph=nav_graph,
configs=configs,
session_state=session_state,
job_target="homefeed",
cognitive_stack=cognitive_stack,
)
# 6. Assertions: Must navigate to HomeFeed FIRST
assert emulator.current_state == "home_feed", (

View File

@@ -54,9 +54,9 @@ def test_goap_planner_avoids_infinite_loop_on_masked_edge():
action_failures=action_failures,
)
# The HD Map should fail, and because the planner is trapped, it forces a restart
# The HD Map should fail, and because the planner is trapped, it falls back to back-tracking
assert (
action_avoided == "force start instagram"
action_avoided == "press back" or action_avoided == "force start instagram"
), "Planner routed BLIND into the dead end despite the edge being masked!"

View File

@@ -855,14 +855,13 @@ class TestVerifySuccessStructuralDelta:
result = memory.verify_success("like", pre_click_xml=pre_xml, post_click_xml=post_xml)
assert result is True
def test_toggle_zero_diff_is_none(self):
"""Zero XML change for a toggle = inconclusive (not confirmed)."""
def test_toggle_zero_diff_is_false(self):
"""Zero XML change for a toggle = fail."""
memory, _ = self._make_memory()
same_xml = '<node text="Like" />'
result = memory.verify_success("like", pre_click_xml=same_xml, post_click_xml=same_xml)
# Zero diff, no markers → None (inconclusive) not True
assert result is None or result is False
assert result is False
def test_follow_success_with_requested_marker(self):
"""
@@ -888,7 +887,7 @@ class TestVerifySuccessStructuralDelta:
result = memory.verify_success("view a post", pre_click_xml="<node/>", post_click_xml=post_xml)
assert result is True
def test_view_post_still_on_grid_is_inconclusive(self):
def test_view_post_still_on_grid_is_false(self):
"""
If after 'view a post' the XML still shows explore_action_bar
WITHOUT post detail markers, navigation failed.
@@ -896,7 +895,7 @@ class TestVerifySuccessStructuralDelta:
memory, _ = self._make_memory()
post_xml = '<node resource-id="com.instagram.android:id/explore_action_bar" />'
result = memory.verify_success("grid item", pre_click_xml="<node/>", post_click_xml=post_xml)
assert result is None # Inconclusive
assert result is False
def test_semantic_gate_blocks_wrong_toggle_element(self):
"""

View File

@@ -92,8 +92,7 @@ class TestBrainOutputParsing:
explored_actions=set(),
)
assert result == "tap explore tab", (
f"Brain extracted '{result}' instead of 'tap explore tab'. "
f"Expected the last-mentioned action to win."
f"Brain extracted '{result}' instead of 'tap explore tab'. " f"Expected the last-mentioned action to win."
)
def test_brain_never_returns_avoided_action(self, monkeypatch):
@@ -116,8 +115,7 @@ class TestBrainOutputParsing:
)
# The action MUST be None or one of the available actions — NEVER the masked one
assert result != "tap messages tab", (
"Brain returned an action that was not in available_actions! "
"This means the masking layer has a hole."
"Brain returned an action that was not in available_actions! " "This means the masking layer has a hole."
)
@@ -146,7 +144,7 @@ class TestBrainAvoidActionsParity:
}
planner.plan_next_step(
"open explore",
"nurture community",
screen,
action_failures={"tap messages tab": 2}, # Masked!
)
@@ -159,8 +157,7 @@ class TestBrainAvoidActionsParity:
if "available to you right now" in line:
# The masked action must NOT be in the available actions list
assert "tap messages tab" not in line, (
f"Planner passed masked action 'tap messages tab' to the Brain as available!\n"
f"Line: {line}"
f"Planner passed masked action 'tap messages tab' to the Brain as available!\n" f"Line: {line}"
)
break
else:
@@ -254,10 +251,7 @@ class TestBrainEmptyResponse:
available_actions=["tap explore tab", "scroll down"],
explored_actions=set(),
)
assert result is None, (
f"Brain returned '{result}' from a whitespace-only response! "
f"Must return None."
)
assert result is None, f"Brain returned '{result}' from a whitespace-only response! " f"Must return None."
class TestPlannerNoOpGuard:
@@ -293,8 +287,7 @@ class TestPlannerNoOpGuard:
for line in prompt.splitlines():
if "available to you right now" in line:
assert "tap profile tab" not in line, (
f"Planner passed no-op action 'tap profile tab' to Brain on OWN_PROFILE!\n"
f"Line: {line}"
f"Planner passed no-op action 'tap profile tab' to Brain on OWN_PROFILE!\n" f"Line: {line}"
)
break
else:
@@ -328,8 +321,7 @@ class TestPlannerNoOpGuard:
for line in prompt.splitlines():
if "available to you right now" in line:
assert "tap home tab" not in line, (
f"Planner passed no-op 'tap home tab' to Brain on HOME_FEED!\n"
f"Line: {line}"
f"Planner passed no-op 'tap home tab' to Brain on HOME_FEED!\n" f"Line: {line}"
)
break
else:

View File

@@ -58,7 +58,7 @@ class TestVerifySuccessGridReels:
</hierarchy>
"""
result = self.engine.verify_success("view a post", explore_xml)
assert result is None, "verify_success should return None (inconclusive) when grid is still visible"
assert result is False, "verify_success should return False when grid is still visible"
def test_profile_grid_reel_accepted(self):
"""Profile grid → Reel must also be accepted."""