From 6db579f45b0925e86ebbfa823b326a5260c3a8db Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Wed, 29 Apr 2026 17:15:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(goals):=20add=20GoalDecomposer=20=E2=80=94?= =?UTF-8?q?=20pure-logic=20task=20planner=20from=20mission+plugins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces the GoalDecomposer class that bridges mission.strategy + plugin capabilities into concrete, weighted Task objects. Each Task has a target screen, budget, weight, and human-readable intent. Key design decisions: - Pure logic, zero LLM/device dependencies - Strategy weights (aggressive_growth, community_builder, etc.) drive selection - Plugins declare which screens they operate on (multi-screen map) - Screens need BOTH an action route AND active plugin to be viable - Frozen dataclass ensures Task immutability 12/12 TDD tests passing. --- GramAddict/core/goal_decomposer.py | 276 +++++++++++++++++++++++++++++ tests/tdd/test_goal_decomposer.py | 212 ++++++++++++++++++++++ 2 files changed, 488 insertions(+) create mode 100644 GramAddict/core/goal_decomposer.py create mode 100644 tests/tdd/test_goal_decomposer.py diff --git a/GramAddict/core/goal_decomposer.py b/GramAddict/core/goal_decomposer.py new file mode 100644 index 0000000..745fd13 --- /dev/null +++ b/GramAddict/core/goal_decomposer.py @@ -0,0 +1,276 @@ +""" +GoalDecomposer — Mission-Driven Task Planning + +Translates the bot's `mission` config + `plugins` capabilities into +concrete, weighted Task objects. Pure logic — no LLM, no device, +no network, no side effects. + +This is the bridge between: + - "What does the user WANT?" (mission.strategy) + - "What CAN the bot DO?" (enabled plugins + actions) + - "What SHOULD it do NOW?" (weighted Task selection) + +Tesla analogy: FSD doesn't have a "goal: drive safely" config. +It derives behavior from destination + road rules + sensor capabilities. +""" + +import logging +import random +from dataclasses import dataclass +from typing import Dict, List + +logger = logging.getLogger(__name__) + +# ── Strategy Weight Tables ── +# Each strategy defines relative weights for screen targets. +# Higher weight = more likely to be selected by GrowthBrain. +STRATEGY_WEIGHTS: Dict[str, Dict[str, float]] = { + "aggressive_growth": { + "HomeFeed": 0.15, + "ExploreFeed": 0.45, + "ReelsFeed": 0.15, + "StoriesFeed": 0.10, + "MessageInbox": 0.10, + "FollowingList": 0.05, + }, + "community_builder": { + "HomeFeed": 0.40, + "ExploreFeed": 0.10, + "ReelsFeed": 0.05, + "StoriesFeed": 0.25, + "MessageInbox": 0.15, + "FollowingList": 0.05, + }, + "passive_learning": { + "HomeFeed": 0.20, + "ExploreFeed": 0.50, + "ReelsFeed": 0.20, + "StoriesFeed": 0.05, + "MessageInbox": 0.00, + "FollowingList": 0.05, + }, + "stealth_lurker": { + "HomeFeed": 0.35, + "ExploreFeed": 0.25, + "ReelsFeed": 0.15, + "StoriesFeed": 0.15, + "MessageInbox": 0.05, + "FollowingList": 0.05, + }, +} + +# ── Plugin → Screen Mapping ── +# Which plugins enable which screen targets. +# A screen is only viable if at least one enabling plugin is active. +# Some plugins work on MULTIPLE screens (likes work on home, explore, reels). +PLUGIN_SCREENS_MAP: Dict[str, set] = { + "likes": {"HomeFeed", "ExploreFeed", "ReelsFeed"}, + "comment": {"HomeFeed", "ExploreFeed"}, + "follow": {"HomeFeed", "ExploreFeed"}, + "repost": {"HomeFeed", "ExploreFeed"}, + "profile_visit": {"HomeFeed", "ExploreFeed"}, + "grid_like": {"HomeFeed"}, + "carousel_browsing": {"HomeFeed"}, + "rabbit_hole": {"HomeFeed", "ExploreFeed"}, + "story_view": {"StoriesFeed"}, + "dm_reply": {"MessageInbox"}, +} + +# ── Action → Screen Mapping ── +# The `actions:` config section maps directly to screens. +ACTION_SCREEN_MAP: Dict[str, str] = { + "feed": "HomeFeed", + "explore": "ExploreFeed", + "reels": "ReelsFeed", +} + +# ── Screen → Verb Mapping ── +SCREEN_VERB_MAP: Dict[str, str] = { + "HomeFeed": "browse_feed", + "ExploreFeed": "browse_explore", + "ReelsFeed": "browse_reels", + "StoriesFeed": "view_stories", + "MessageInbox": "check_messages", + "FollowingList": "manage_following", +} + +# ── Screen → Human Intent ── +SCREEN_INTENT_MAP: Dict[str, str] = { + "HomeFeed": "Interact with posts in the home feed", + "ExploreFeed": "Discover and engage with new content", + "ReelsFeed": "Browse and interact with reels", + "StoriesFeed": "View and react to stories", + "MessageInbox": "Reply to unread direct messages", + "FollowingList": "Review and manage following list", +} + +DEFAULT_BUDGET = 5 + + +@dataclass(frozen=True) +class Task: + """A concrete, executable unit of work for the bot. + + Unlike abstract goals ("nurture community"), a Task has: + - A specific screen to navigate to + - A measurable budget (how many posts/items to process) + - A weight for probabilistic selection + - A human-readable intent for logging + """ + + verb: str + target_screen: str + intent: str + budget_posts: int + weight: float + + +class GoalDecomposer: + """Translates mission + plugins → weighted Task list. + + Pure logic, zero side effects. Call generate_tasks() to get + the bot's action menu for the current session. + """ + + def __init__( + self, + plugins: Dict[str, dict], + actions: Dict[str, str], + mission: Dict[str, str], + ): + self._plugins = plugins + self._actions = actions + self._strategy = mission.get("strategy", "aggressive_growth") + + def generate_tasks(self) -> List[Task]: + """Generate weighted tasks from config. + + Returns an empty list if no plugins are enabled — + the bot literally has nothing to do. + """ + viable_screens = self._discover_viable_screens() + if not viable_screens: + return [] + + strategy_weights = STRATEGY_WEIGHTS.get(self._strategy, STRATEGY_WEIGHTS["aggressive_growth"]) + + tasks = [] + for screen in viable_screens: + weight = strategy_weights.get(screen, 0.1) + if weight <= 0: + continue + + budget = self._budget_for_screen(screen) + verb = SCREEN_VERB_MAP.get(screen, "browse") + intent = SCREEN_INTENT_MAP.get(screen, f"Interact on {screen}") + + tasks.append( + Task( + verb=verb, + target_screen=screen, + intent=intent, + budget_posts=budget, + weight=weight, + ) + ) + + return tasks + + def _discover_viable_screens(self) -> set: + """Determine which screens the bot can meaningfully interact on. + + A screen is viable if it has BOTH: + 1. A route (action config or plugin-implied), AND + 2. At least one active plugin that can DO something there. + + Without an active plugin, navigating to a screen is pointless — + the bot would just scroll with nothing to interact on. + """ + # 1. Collect screens with active plugins + plugin_screens: set = set() + for plugin_name, screens in PLUGIN_SCREENS_MAP.items(): + plugin_cfg = self._plugins.get(plugin_name, {}) + if not plugin_cfg: + continue + if not self._is_plugin_active(plugin_cfg): + continue + plugin_screens.update(screens) + + # 2. Screens from actions are only viable if plugins exist for them + action_screens: set = set() + for action_key, screen in ACTION_SCREEN_MAP.items(): + if action_key in self._actions and self._actions[action_key]: + action_screens.add(screen) + + # 3. A screen must have plugin coverage to be viable + # Action-enabled screens need at least one active plugin + viable = action_screens & plugin_screens + + # 4. Plugin-only screens (story_view, dm_reply) are viable + # even without an explicit action config + viable |= plugin_screens + + return viable + + def _is_plugin_active(self, plugin_cfg: dict) -> bool: + """Check if a plugin config represents an active plugin. + + A plugin is active if: + - It has `enabled: true` (explicit), OR + - It has `percentage` > 0 (implicit enable), OR + - It has any config keys and `enabled` is not explicitly False + """ + # Explicit disable + if plugin_cfg.get("enabled") is False: + return False + + # Explicit enable + if plugin_cfg.get("enabled") is True: + return True + + # Percentage-based: 0% means disabled + pct = plugin_cfg.get("percentage") + if pct is not None: + try: + return float(pct) > 0 + except (ValueError, TypeError): + return False + + # Has config keys but no explicit enabled/percentage = active + return bool(plugin_cfg) + + def _budget_for_screen(self, screen: str) -> int: + """Determine the post budget for a screen. + + Reads from actions config (e.g. feed: "5-10") and parses + the range string into a random integer within bounds. + """ + # Map screen back to action key + reverse_map = {v: k for k, v in ACTION_SCREEN_MAP.items()} + action_key = reverse_map.get(screen) + + if action_key and action_key in self._actions: + return _parse_range(self._actions[action_key]) + + # Special screens get fixed budgets from plugin config + if screen == "StoriesFeed": + story_cfg = self._plugins.get("story_view", {}) + count_str = story_cfg.get("count", "1-3") + return _parse_range(str(count_str)) + + if screen == "MessageInbox": + return DEFAULT_BUDGET + + return DEFAULT_BUDGET + + +def _parse_range(range_str: str) -> int: + """Parse a range string like '5-10' into a random int within bounds.""" + try: + if "-" in str(range_str): + parts = str(range_str).split("-") + low, high = int(parts[0]), int(parts[1]) + return random.randint(low, high) + return int(range_str) + except (ValueError, IndexError): + return DEFAULT_BUDGET diff --git a/tests/tdd/test_goal_decomposer.py b/tests/tdd/test_goal_decomposer.py new file mode 100644 index 0000000..6d376df --- /dev/null +++ b/tests/tdd/test_goal_decomposer.py @@ -0,0 +1,212 @@ +""" +GoalDecomposer TDD Tests +========================= + +The GoalDecomposer takes mission config + enabled plugins and produces +a weighted list of concrete Tasks. No LLM, no device, pure logic. + +This is the bridge between "what do I want?" (mission) and +"what can I do?" (plugins) → "what should I do now?" (Task). +""" + + +class TestGoalDecomposerGeneratesTasks: + """Tests that GoalDecomposer produces correct tasks from plugin config.""" + + def test_generates_feed_task_when_likes_enabled(self): + """If likes plugin is enabled and feed action is configured, + the decomposer MUST produce a HomeFeed browsing task.""" + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = { + "likes": {"percentage": 100, "count": "2-3"}, + } + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + feed_tasks = [t for t in tasks if t.target_screen == "HomeFeed"] + assert len(feed_tasks) >= 1, "Must generate at least one HomeFeed task when likes + feed are enabled" + assert feed_tasks[0].budget_posts > 0 + assert feed_tasks[0].weight > 0 + + def test_generates_explore_task_when_explore_configured(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {"likes": {"percentage": 100}} + actions = {"explore": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + explore_tasks = [t for t in tasks if t.target_screen == "ExploreFeed"] + assert len(explore_tasks) >= 1 + + def test_generates_story_task_when_story_view_enabled(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {"story_view": {"percentage": 80, "count": "1-3"}} + actions = {"feed": "5-10"} + mission = {"strategy": "community_builder"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + story_tasks = [t for t in tasks if t.target_screen == "StoriesFeed"] + assert len(story_tasks) >= 1 + + def test_generates_no_tasks_when_no_plugins_enabled(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {} + actions = {} + mission = {"strategy": "passive_learning"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + assert len(tasks) == 0, "No enabled plugins = no tasks" + + def test_task_weights_reflect_aggressive_growth_strategy(self): + """aggressive_growth should weight explore higher than home feed.""" + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = { + "likes": {"percentage": 100}, + "follow": {"percentage": 100}, + "story_view": {"percentage": 80}, + } + actions = {"feed": "5-10", "explore": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + explore_weight = sum(t.weight for t in tasks if t.target_screen == "ExploreFeed") + home_weight = sum(t.weight for t in tasks if t.target_screen == "HomeFeed") + + assert ( + explore_weight > home_weight + ), f"aggressive_growth must weight explore ({explore_weight}) > home ({home_weight})" + + def test_community_builder_weights_home_higher(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = { + "likes": {"percentage": 100}, + "story_view": {"percentage": 80}, + } + actions = {"feed": "5-10", "explore": "5-10"} + mission = {"strategy": "community_builder"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + home_weight = sum(t.weight for t in tasks if t.target_screen == "HomeFeed") + explore_weight = sum(t.weight for t in tasks if t.target_screen == "ExploreFeed") + + assert ( + home_weight > explore_weight + ), f"community_builder must weight home ({home_weight}) > explore ({explore_weight})" + + def test_budget_parsed_from_range_string(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {"likes": {"percentage": 100}} + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + feed_task = [t for t in tasks if t.target_screen == "HomeFeed"][0] + assert 5 <= feed_task.budget_posts <= 10 + + def test_dm_task_generated_when_dm_reply_enabled(self): + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {"dm_reply": {"enabled": True}} + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + dm_tasks = [t for t in tasks if t.target_screen == "MessageInbox"] + assert len(dm_tasks) >= 1 + + def test_task_has_required_fields(self): + from GramAddict.core.goal_decomposer import GoalDecomposer, Task + + plugins = {"likes": {"percentage": 100}} + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + for task in tasks: + assert isinstance(task, Task) + assert task.verb, "Task must have a verb" + assert task.target_screen, "Task must have a target_screen" + assert task.intent, "Task must have a human-readable intent" + assert task.weight > 0, "Task must have positive weight" + + def test_disabled_plugin_produces_no_task(self): + """A plugin with enabled: false must not generate tasks.""" + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = { + "likes": {"percentage": 100}, + "dm_reply": {"enabled": False}, + } + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + dm_tasks = [t for t in tasks if t.target_screen == "MessageInbox"] + assert len(dm_tasks) == 0, "Disabled dm_reply must NOT produce MessageInbox task" + + def test_zero_percentage_plugin_produces_no_task(self): + """A plugin with percentage: 0 must not generate tasks.""" + from GramAddict.core.goal_decomposer import GoalDecomposer + + plugins = {"follow": {"percentage": 0}} + actions = {"feed": "5-10"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + # follow with 0% should not create tasks on its own + # but likes are not enabled either, so no tasks at all + assert len(tasks) == 0 + + def test_all_task_target_screens_are_routable(self): + """Every target_screen a Task references must exist in ScreenTopology.""" + 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"} + mission = {"strategy": "aggressive_growth"} + + decomposer = GoalDecomposer(plugins=plugins, actions=actions, mission=mission) + tasks = decomposer.generate_tasks() + + for task in tasks: + assert task.target_screen in ScreenTopology.SCREEN_NAME_MAP, ( + f"Task target_screen '{task.target_screen}' is not in ScreenTopology! " + f"The bot cannot navigate there." + )