Hardening: Centralized zero-trust CognitiveStack validation and removed synthetic e2e test mocks

GramAddict/core/bot_flow.py

@@ -685,6 +685,7 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod,
 
     if cognitive_stack is None:
         cognitive_stack = {}
+    _validate_cognitive_stack(cognitive_stack, "ProfileInteraction")
 
     if hasattr(session_state, "my_username") and username == session_state.my_username:
         logger.info(f"🤝 [Deep Interaction] Skipping own profile @{username} to prevent self-interactions.")
@@ -824,6 +825,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta
 
     from colorama import Fore
 
+    _validate_cognitive_stack(cognitive_stack, "StoriesLoop")
     logger.info("🎬 [StoriesFeed] Starting native story binging loop...", extra={"color": f"{Fore.CYAN}"})
 
     dopamine = cognitive_stack.get("dopamine")
@@ -893,6 +895,36 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta
     return "FEED_EXHAUSTED"
 
 
+def _validate_cognitive_stack(cognitive_stack, context_name):
+    """
+    Validates that the cognitive stack has all required engine dependencies
+    injected properly. This is the ultimate zero-trust guard for plugins.
+    """
+    if not isinstance(cognitive_stack, dict):
+        raise TypeError(f"[{context_name}] CognitiveStack must be a dict, got {type(cognitive_stack)}")
+
+    required_engines = [
+        "dopamine",
+        "darwin",
+        "resonance",
+        "active_inference",
+        "growth_brain",
+        "swarm",
+        "writer",
+        "nav_graph",
+        "zero_engine",
+        "telepathic",
+    ]
+
+    missing = [eng for eng in required_engines if eng not in cognitive_stack or cognitive_stack[eng] is None]
+    if missing:
+        import logging
+
+        logger = logging.getLogger(__name__)
+        logger.error(f"🚨 [{context_name}] CognitiveStack missing required engines: {missing}")
+        raise ValueError(f"[{context_name}] CognitiveStack missing required engines: {missing}")
+
+
 def _run_zero_latency_feed_loop(
     device, zero_engine, nav_graph, configs, session_state, job_target, cognitive_stack, is_reels=False
 ):
@@ -906,6 +938,7 @@ def _run_zero_latency_feed_loop(
     - Darwin is the SOLE dwell controller → no duplicate sleep calls
     - SwarmProtocol emits pheromones after successful interactions
     """
+    _validate_cognitive_stack(cognitive_stack, "FeedLoop")
     logger.info(f"🔄 Entering Zero-Latency Interaction Pool. Feed: {job_target}")
 
     dopamine = cognitive_stack.get("dopamine")
@@ -1055,6 +1088,7 @@ def _run_zero_latency_search_loop(
     """
     Executes the autonomous Search & Interact logic.
     """
+    _validate_cognitive_stack(cognitive_stack, "SearchLoop")
     logger.info("🧠 [Search Engine] Initiating keyword discovery...", extra={"color": f"{Style.BRIGHT}{Fore.CYAN}"})
 
     import random
@@ -1080,6 +1114,16 @@ def _run_zero_latency_search_loop(
         xml = device.dump_hierarchy()
         telepathic = cognitive_stack.get("telepathic")
 
+        # ── Perimeter Guard: Verify we're still inside Instagram ──
+        if xml:
+            search_packages = set(re.findall(r'package="([^"]+)"', xml))
+            search_app_id = getattr(device, "app_id", "com.instagram.android")
+            if search_packages and search_app_id not in search_packages:
+                logger.error(f"🚨 [SearchLoop] FOREIGN APP DETECTED! Packages: {search_packages}. Aborting loop.")
+                device.press("back")
+                sleep(1.5)
+                return "CONTEXT_LOST"
+
         # Find search bar
         search_bar = telepathic.find_best_node(xml, "Search edit text box or magnifying glass input", device=device)
         if search_bar:

tests/e2e/test_workflow_feed_foreign_app_escape.py

@@ -11,13 +11,12 @@ import datetime
 from tests.e2e.conftest import E2EDeviceStub, load_fixture_xml
 
 
-def test_feed_loop_escapes_foreign_app(e2e_configs):
+def test_feed_loop_escapes_foreign_app(e2e_configs, e2e_cognitive_stack_factory):
     """
     Simulates: Feed → Play Store.
     The loop must detect com.android.vending and abort immediately.
     """
     from GramAddict.core.bot_flow import _run_zero_latency_feed_loop
-    from GramAddict.core.dopamine_engine import DopamineEngine
 
     feed_xml = load_fixture_xml("home_feed_with_ad.xml")
     play_store_xml = load_fixture_xml("play_store_from_story_link.xml")
@@ -25,27 +24,8 @@ def test_feed_loop_escapes_foreign_app(e2e_configs):
     # Feed -> Play Store
     device = E2EDeviceStub([feed_xml, play_store_xml])
 
-    dopamine = DopamineEngine()
-    dopamine.session_limit_seconds = 999
-
-    # Needs a dummy zero_engine and nav_graph
-    class DummyZeroEngine:
-        pass
-
-    class DummyNavGraph:
-        def do(self, action):
-            pass
-
-        def navigate_to(self, state, engine):
-            pass
-
-    cognitive_stack = {
-        "dopamine": dopamine,
-        "zero_engine": DummyZeroEngine(),
-        "nav_graph": DummyNavGraph(),
-        "radome": None,
-        "active_inference": None,
-    }
+    cognitive_stack = e2e_cognitive_stack_factory(device)
+    cognitive_stack["dopamine"].session_limit_seconds = 999
 
     session_state = type(
         "S", (), {"startTime": datetime.datetime.now(), "check_limit": lambda *args, **kwargs: (False, False)}
@@ -55,7 +35,13 @@ def test_feed_loop_escapes_foreign_app(e2e_configs):
     e2e_configs.args.speed_multiplier = 0.01
 
     result = _run_zero_latency_feed_loop(
-        device, DummyZeroEngine(), DummyNavGraph(), e2e_configs, session_state, "HomeFeed", cognitive_stack
+        device,
+        cognitive_stack["zero_engine"],
+        cognitive_stack["nav_graph"],
+        e2e_configs,
+        session_state,
+        "HomeFeed",
+        cognitive_stack,
     )
 
     # Must return CONTEXT_LOST

tests/e2e/test_workflow_stories_foreign_app_escape.py

@@ -18,13 +18,12 @@ import datetime
 from tests.e2e.conftest import E2EDeviceStub, load_fixture_xml
 
 
-def test_story_loop_escapes_foreign_app(e2e_configs):
+def test_story_loop_escapes_foreign_app(e2e_configs, e2e_cognitive_stack_factory):
     """
     Simulates: Story → Story → Story → Play Store (via swipe-up link).
     The loop must detect com.android.vending and abort immediately.
     """
     from GramAddict.core.bot_flow import _run_zero_latency_stories_loop
-    from GramAddict.core.dopamine_engine import DopamineEngine
 
     story_xml = load_fixture_xml("story_view_before_linkout.xml")
     play_store_xml = load_fixture_xml("play_store_from_story_link.xml")
@@ -32,9 +31,8 @@ def test_story_loop_escapes_foreign_app(e2e_configs):
     # 3 story iterations → then Play Store appears
     device = E2EDeviceStub([story_xml, story_xml, story_xml, play_store_xml])
 
-    dopamine = DopamineEngine()
-    dopamine.session_limit_seconds = 999
-    cognitive_stack = {"dopamine": dopamine}
+    cognitive_stack = e2e_cognitive_stack_factory(device)
+    cognitive_stack["dopamine"].session_limit_seconds = 999
 
     session_state = type("S", (), {"startTime": datetime.datetime.now()})()