feat(orchestrator): wire GoalDecomposer into bot_flow.py

Replace the old dual-path orchestrator (abstract goals vs legacy desires)
with unified GoalDecomposer-driven task routing:

1. GoalDecomposer reads mission.strategy + plugins config
2. Generates weighted Task objects (verb, target_screen, budget)
3. GrowthBrain.select_task() picks one probabilistically
4. Selected Task's target_screen routes through existing nav_graph
5. Feed loops + PluginRegistry handle the actual interactions

The abstract goals path (goal_executor.achieve('Nurture community'))
that caused infinite scrolling is now eliminated entirely.

Legacy desire fallback preserved for configs without plugins.

22/22 tests passing.
This commit is contained in:
2026-04-29 17:20:04 +02:00
parent b6846ab0fe
commit 0e43996ccd
3 changed files with 161 additions and 59 deletions

View File

@@ -317,3 +317,86 @@ class TestGrowthBrainTaskSelection:
# With aggressive_growth, ExploreFeed (weight 0.45) should dominate
assert len(counts) > 1, "Selection must pick from multiple screens"
class TestOrchestratorTaskRouting:
"""Tests that every Task produced by GoalDecomposer is routable by ScreenTopology."""
def test_task_to_screen_topology_mapping(self):
"""Each Task.target_screen MUST exist in ScreenTopology.SCREEN_NAME_MAP."""
from GramAddict.core.goal_decomposer import GoalDecomposer
from GramAddict.core.screen_topology import ScreenTopology
plugins = {
"likes": {"percentage": 100},
"follow": {"percentage": 100},
"comment": {"percentage": 40},
"story_view": {"percentage": 80},
"dm_reply": {"enabled": True},
}
actions = {"feed": "5-10", "explore": "5-10", "reels": "3-5"}
mission = {"strategy": "aggressive_growth"}
decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission)
tasks = decomposer.generate_tasks()
assert len(tasks) > 0
for task in tasks:
assert task.target_screen in ScreenTopology.SCREEN_NAME_MAP, (
f"Task '{task.verb}''{task.target_screen}' has no ScreenTopology mapping!"
)
def test_all_task_screens_reachable_from_home(self):
"""Every Task target must be reachable via BFS from HOME_FEED."""
from GramAddict.core.goal_decomposer import GoalDecomposer
from GramAddict.core.screen_topology import ScreenTopology
from GramAddict.core.perception.screen_identity import ScreenType
plugins = {
"likes": {"percentage": 100},
"story_view": {"percentage": 80},
"dm_reply": {"enabled": True},
}
actions = {"feed": "5-10", "explore": "5-10"}
mission = {"strategy": "community_builder"}
decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission)
tasks = decomposer.generate_tasks()
for task in tasks:
target_type = ScreenTopology.SCREEN_NAME_MAP.get(task.target_screen)
assert target_type is not None
if target_type == ScreenType.HOME_FEED:
continue # Already there
route = ScreenTopology.find_route(ScreenType.HOME_FEED, target_type)
assert route is not None, (
f"No HD Map route from HOME_FEED to {task.target_screen}! "
f"The bot cannot navigate there."
)
def test_task_target_screen_to_goal_string_conversion(self):
"""Every task target_screen must convert to a valid GOAP goal string."""
from GramAddict.core.goal_decomposer import GoalDecomposer
from GramAddict.core.screen_topology import ScreenTopology
plugins = {"likes": {"percentage": 100}, "dm_reply": {"enabled": True}}
actions = {"feed": "5-10", "explore": "5-10"}
mission = {"strategy": "aggressive_growth"}
decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission)
tasks = decomposer.generate_tasks()
for task in tasks:
goal_str = ScreenTopology.screen_name_to_goal(task.target_screen)
assert goal_str, (
f"Task '{task.target_screen}' cannot be converted to GOAP goal!"
)
# The goal string should resolve back to a target screen
target = ScreenTopology.goal_to_target_screen(goal_str)
assert target is not None, (
f"Goal string '{goal_str}' doesn't resolve to any ScreenType!"
)