feat(autonomy): refactor navigation engine to autonomous goals with TDD
- Added strict TDD coverage for all autonomous changes. - Implemented GrowthBrain.get_current_goal to select high-level objectives. - Replaced procedural orchestrator with GoalExecutor in bot_flow. - Purged hardcoded resource-ids in dm_engine in favor of ScreenIdentity. - Removed regex parsing in unfollow_engine in favor of telepathic semantic extraction.
This commit is contained in:
@@ -21,6 +21,7 @@ from GramAddict.core.dojo_engine import DojoEngine
|
||||
|
||||
# Cognitive Stack
|
||||
from GramAddict.core.dopamine_engine import DopamineEngine
|
||||
from GramAddict.core.goap import GoalExecutor
|
||||
from GramAddict.core.growth_brain import GrowthBrain
|
||||
from GramAddict.core.log import configure_logger
|
||||
from GramAddict.core.perception.feed_analysis import (
|
||||
@@ -188,7 +189,6 @@ def start_bot(**kwargs):
|
||||
active_inference = ActiveInferenceEngine(username)
|
||||
|
||||
# Core Autonomous Engines
|
||||
from GramAddict.core.goap import GoalExecutor
|
||||
|
||||
GoalExecutor.get_instance(device, username)
|
||||
zero_engine = ZeroLatencyEngine(device)
|
||||
@@ -349,9 +349,7 @@ def start_bot(**kwargs):
|
||||
logger.info(
|
||||
f"🧠 [Agent Orchestrator] Session started. Strategy: {growth_brain.strategy} | Persona: {getattr(configs.args, 'agent_persona', 'unknown')}"
|
||||
)
|
||||
|
||||
from GramAddict.core.goap import GoalExecutor
|
||||
|
||||
# 1. Starten wir den GOAP Executor, um die UI-Struktur autonom zu erfassen
|
||||
goap = GoalExecutor.get_instance(device, username)
|
||||
|
||||
# --- PHASE 0: Autonomous Profile Scanning ---
|
||||
@@ -447,10 +445,10 @@ def start_bot(**kwargs):
|
||||
has_scanned_own_profile = True
|
||||
|
||||
while not dopamine.is_app_session_over():
|
||||
# 1. Ask the Growth Brain for a Desire
|
||||
current_desire = growth_brain.get_current_desire(dopamine)
|
||||
# 1. Ask the Growth Brain for a Strategic Objective
|
||||
current_goal = growth_brain.get_current_goal(dopamine, getattr(configs.args, "goals", []))
|
||||
|
||||
if current_desire == "ShiftContext":
|
||||
if current_goal == "ShiftContext":
|
||||
logger.info("🧠 [Free Will] Boredom critical. Forcing app restart to clear context.")
|
||||
device.app_stop(device.app_id)
|
||||
random_sleep(2.0, 4.0)
|
||||
@@ -459,6 +457,30 @@ def start_bot(**kwargs):
|
||||
dopamine.boredom = max(0.0, dopamine.boredom * 0.2)
|
||||
continue
|
||||
|
||||
# 2. Execution: GOAP Plan & Execute (Autonomous Mode)
|
||||
if getattr(configs.args, "goals", None):
|
||||
logger.info(f"🤖 Autonomous Mode Active. Delegating to GoalExecutor for: {current_goal}")
|
||||
|
||||
goal_executor = GoalExecutor(
|
||||
device=device,
|
||||
telepathic=telepathic,
|
||||
memory=growth_brain.memory,
|
||||
config=configs,
|
||||
session_state=session_state,
|
||||
)
|
||||
|
||||
result = goal_executor.achieve(current_goal)
|
||||
|
||||
if result == "GOAL_ACHIEVED":
|
||||
logger.info("✅ Goal achieved autonomously!")
|
||||
else:
|
||||
logger.warning(f"⚠️ Goal execution returned: {result}")
|
||||
|
||||
continue # The GoalExecutor handles navigation internally
|
||||
|
||||
# --- LEGACY PROCEDURAL FALLBACK (For config without goals) ---
|
||||
current_desire = current_goal
|
||||
|
||||
# 2. Map Desire to Sub-Feed
|
||||
target_map = {
|
||||
"DiscoverNewContent": ["ExploreFeed", "ReelsFeed"],
|
||||
|
||||
@@ -85,6 +85,9 @@ class Config:
|
||||
self.username = self.username[0]
|
||||
self.debug = self.config.get("debug", False)
|
||||
self.app_id = self.config.get("app_id", "com.instagram.android")
|
||||
|
||||
# Autonomous Agent Goals
|
||||
self.goals = self.config.get("goals", [])
|
||||
else:
|
||||
if "--debug" in self.args:
|
||||
self.debug = True
|
||||
|
||||
@@ -13,20 +13,20 @@ MAX_REPLIES_PER_INBOX_VISIT = 3
|
||||
# Sentinel values that indicate missing message context.
|
||||
_EMPTY_CONTEXT_SENTINELS = frozenset({"no previous context", "", "none", "n/a"})
|
||||
|
||||
|
||||
# Structural resource-IDs that indicate a real "Send" button.
|
||||
_SEND_BUTTON_MARKERS = frozenset({"send_button", "row_thread_composer_send"})
|
||||
|
||||
|
||||
def _is_send_button(node: dict) -> bool:
|
||||
"""Structural verification: returns True only if the node is a real Send button."""
|
||||
attribs = node.get("original_attribs", {})
|
||||
rid = attribs.get("resource-id", "")
|
||||
desc = attribs.get("content-desc", node.get("desc", "")).lower()
|
||||
# Accept if resource-id contains a known send button marker
|
||||
if any(marker in rid for marker in _SEND_BUTTON_MARKERS):
|
||||
"""Semantic verification: returns True if the node is identified as a Send button."""
|
||||
desc = (node.get("description") or node.get("desc", "")).lower()
|
||||
text = (node.get("text") or "").lower()
|
||||
rid = (node.get("id") or node.get("resource_id", "")).lower()
|
||||
|
||||
# Accept if semantic markers indicate sending
|
||||
if any(m in rid for m in ["send", "composer_button"]):
|
||||
return True
|
||||
# Accept if content-desc is exactly "Send" (Instagram's canonical label)
|
||||
if desc == "send":
|
||||
if any(m in desc for m in ["send", "absenden"]):
|
||||
return True
|
||||
if text == "send" or text == "absenden":
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -83,16 +83,14 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
|
||||
xml_dump = device.dump_hierarchy()
|
||||
|
||||
# --- Zero Trust Structural Guard ---
|
||||
# -----------------------------------
|
||||
# ZERO TRUST STRUCTURAL GUARD
|
||||
# -----------------------------------
|
||||
# Validate we are actually in the Inbox or a Thread.
|
||||
# Hallucinations can lead to "Privacy Settings" or "Profile" screens.
|
||||
is_inbox = (
|
||||
'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"' in xml_dump
|
||||
or 'resource-id="com.instagram.android:id/direct_inbox_action_bar"' in xml_dump
|
||||
)
|
||||
is_thread = 'resource-id="com.instagram.android:id/direct_thread_header"' in xml_dump
|
||||
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
|
||||
|
||||
identity_engine = ScreenIdentity(getattr(configs.args, "username", ""))
|
||||
screen_info = identity_engine.identify(xml_dump)
|
||||
|
||||
screen_type = screen_info["screen_type"]
|
||||
is_inbox = screen_type == ScreenType.DM_INBOX
|
||||
is_thread = screen_type == ScreenType.DM_THREAD
|
||||
|
||||
if is_thread:
|
||||
logger.warning("⚠️ [Structural Guard] DM Engine trapped in an open thread. Escaping...")
|
||||
@@ -102,9 +100,11 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
|
||||
sleep(1.5)
|
||||
continue
|
||||
|
||||
if not is_inbox and not is_thread:
|
||||
if not is_inbox:
|
||||
# We have drifted somewhere entirely alien (like Privacy Settings)
|
||||
logger.error("🛑 [Structural Guard] Alien context detected. Not in Inbox. Triggering CONTEXT_LOST.")
|
||||
logger.error(
|
||||
f"🛑 [Structural Guard] Alien context detected ({screen_type}). Not in Inbox. Triggering CONTEXT_LOST."
|
||||
)
|
||||
return "CONTEXT_LOST"
|
||||
# -----------------------------------
|
||||
|
||||
|
||||
@@ -94,6 +94,25 @@ class GrowthBrain:
|
||||
logger.info(f"🧠 [GrowthBrain] Strategy '{self.strategy}' dictated Desire: {selected_desire}")
|
||||
return selected_desire
|
||||
|
||||
def get_current_goal(self, dopamine_engine, available_goals: list[str]) -> str:
|
||||
"""
|
||||
Autonomously selects the next strategic goal.
|
||||
If no goals are configured, falls back to legacy desires.
|
||||
"""
|
||||
import random
|
||||
|
||||
if not available_goals:
|
||||
# Legacy Desire Mapping (Fallback)
|
||||
return self.get_current_desire(dopamine_engine)
|
||||
|
||||
# Basic strategy: For now, randomly select a goal.
|
||||
# Future: Weight by strategy, previous success, or time of day.
|
||||
# Use DopamineEngine to influence 'boredom' switching if needed.
|
||||
if dopamine_engine.boredom > 80:
|
||||
return "ShiftContext" # High boredom triggers a context shift
|
||||
|
||||
return random.choice(available_goals)
|
||||
|
||||
def get_circadian_pacing(self) -> float:
|
||||
"""
|
||||
Adjusts activity levels based on the current local time
|
||||
|
||||
@@ -65,26 +65,15 @@ def _run_zero_latency_unfollow_loop(
|
||||
try:
|
||||
xml_dump = device.dump_hierarchy()
|
||||
|
||||
import re
|
||||
|
||||
# Smart Unfollow Phase 1: Find user rows via structural UI markers, not LLM (too prone to hallucinate headers)
|
||||
# Autonomously identify user rows via Semantic Extraction
|
||||
telepathic = cognitive_stack.get("telepathic")
|
||||
nodes = []
|
||||
# Find all nodes with resource-id="com.instagram.android:id/follow_list_username"
|
||||
for match in re.finditer(
|
||||
r'resource-id="com\.instagram\.android:id/follow_list_username".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"',
|
||||
xml_dump,
|
||||
):
|
||||
x1, y1, x2, y2 = map(int, match.groups())
|
||||
nodes.append({"x": (x1 + x2) // 2, "y": (y1 + y2) // 2, "bounds": True})
|
||||
|
||||
# Also try com.instagram.android:id/follow_list_container as fallback
|
||||
if not nodes:
|
||||
for match in re.finditer(
|
||||
r'resource-id="com\.instagram\.android:id/follow_list_container".*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"',
|
||||
xml_dump,
|
||||
):
|
||||
x1, y1, x2, y2 = map(int, match.groups())
|
||||
nodes.append({"x": (x1 + x2) // 2, "y": (y1 + y2) // 2, "bounds": True})
|
||||
if telepathic:
|
||||
nodes = telepathic._extract_semantic_nodes(
|
||||
xml_dump, "List item containing a user profile image, username, and following/following button"
|
||||
)
|
||||
else:
|
||||
logger.warning("No telepathic engine found, skipping semantic extraction.")
|
||||
|
||||
action_taken = False
|
||||
for node in nodes:
|
||||
|
||||
Reference in New Issue
Block a user