diff --git a/tests/e2e/test_workflow_dm_inbox.py b/tests/e2e/test_workflow_dm_inbox.py
new file mode 100644
index 0000000..28d8661
--- /dev/null
+++ b/tests/e2e/test_workflow_dm_inbox.py
@@ -0,0 +1,75 @@
+"""
+E2E: DM Inbox Workflow
+=======================
+Tests the FULL production pipeline when the device shows the DM inbox.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_dm_inbox_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The DM inbox must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ dm_xml = _load_fixture("dm_inbox_dump.xml")
+ device = E2EDeviceStub([dm_xml, dm_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on DM inbox."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on DM inbox!"
+ )
+
+
+def test_dm_thread_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A DM thread (conversation view) must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ dm_thread_xml = _load_fixture("dm_thread_dump.xml")
+ device = E2EDeviceStub([dm_thread_xml, dm_thread_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on DM thread."
+ )
diff --git a/tests/e2e/test_workflow_explore_feed.py b/tests/e2e/test_workflow_explore_feed.py
new file mode 100644
index 0000000..bccccc8
--- /dev/null
+++ b/tests/e2e/test_workflow_explore_feed.py
@@ -0,0 +1,56 @@
+"""
+E2E: Explore Feed Workflow
+===========================
+Tests the FULL production pipeline when the device shows the Explore grid.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_explore_grid_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The Explore feed grid must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ explore_xml = _load_fixture("explore_feed_dump.xml")
+ device = E2EDeviceStub([explore_xml, explore_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on Explore grid."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on Explore feed!"
+ )
diff --git a/tests/e2e/test_workflow_instagram_modal.py b/tests/e2e/test_workflow_instagram_modal.py
new file mode 100644
index 0000000..4d5315e
--- /dev/null
+++ b/tests/e2e/test_workflow_instagram_modal.py
@@ -0,0 +1,95 @@
+"""
+E2E: Instagram Modal Workflow
+===============================
+Tests the FULL production pipeline when Instagram shows a modal overlay
+(survey, "Not Now" prompt, content creation flow).
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+
+INSTAGRAM_SURVEY_XML = """
+
+
+
+
+
+
+
+
+
+"""
+
+NORMAL_POST_XML = """
+
+
+
+
+
+"""
+
+
+class FakeSAEModal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ if "dialog_container" in xml:
+ return SituationType.OBSTACLE_MODAL
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_instagram_survey_modal_terminates_chain(monkeypatch, e2e_workflow_ctx):
+ """
+ An Instagram survey modal ("How are you enjoying Instagram?")
+ must be detected and dismissed. The chain must skip interactions.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAEModal,
+ )
+ # Disable TelepathicEngine.find_best_node since it needs a real LLM
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.TelepathicEngine",
+ type("FakeTele", (), {
+ "get_instance": classmethod(lambda cls: cls()),
+ "find_best_node": lambda self, *a, **kw: None,
+ }),
+ )
+
+ device = E2EDeviceStub([INSTAGRAM_SURVEY_XML, NORMAL_POST_XML])
+ results, ctx = e2e_workflow_ctx(device, context_xml=INSTAGRAM_SURVEY_XML)
+
+ guard_fired = any(r.executed and r.should_skip for r in results)
+ assert guard_fired, (
+ "obstacle_guard did not fire on Instagram survey modal!"
+ )
+
+ total_interactions = sum(r.interactions for r in results)
+ assert total_interactions == 0, (
+ f"{total_interactions} interaction(s) performed on a survey modal!"
+ )
+
+ assert "back" in device.pressed_keys, (
+ "obstacle_guard did not press BACK on survey modal!"
+ )
diff --git a/tests/e2e/test_workflow_locked_screen.py b/tests/e2e/test_workflow_locked_screen.py
new file mode 100644
index 0000000..4b21cd0
--- /dev/null
+++ b/tests/e2e/test_workflow_locked_screen.py
@@ -0,0 +1,65 @@
+"""
+E2E: Locked Screen Workflow
+=============================
+Tests the FULL production pipeline when the device screen is locked.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+
+LOCKED_SCREEN_XML = """
+
+
+
+
+
+
+
+
+
+"""
+
+
+class FakeSAELocked:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ if "systemui" in xml and "keyguard" in xml:
+ return SituationType.OBSTACLE_LOCKED_SCREEN
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_locked_screen_does_not_interact(monkeypatch, e2e_workflow_ctx):
+ """
+ A locked screen must not trigger any interaction plugins.
+ The obstacle_guard currently doesn't handle OBSTACLE_LOCKED_SCREEN
+ explicitly — this test documents whether it falls through.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAELocked,
+ )
+
+ device = E2EDeviceStub([LOCKED_SCREEN_XML])
+ results, ctx = e2e_workflow_ctx(device, context_xml=LOCKED_SCREEN_XML)
+
+ # No interactions may happen on a locked screen
+ total_interactions = sum(r.interactions for r in results)
+ assert total_interactions == 0, (
+ f"{total_interactions} interaction(s) performed on a locked screen!"
+ )
diff --git a/tests/e2e/test_workflow_post_detail.py b/tests/e2e/test_workflow_post_detail.py
new file mode 100644
index 0000000..68f10fd
--- /dev/null
+++ b/tests/e2e/test_workflow_post_detail.py
@@ -0,0 +1,81 @@
+"""
+E2E: Post Detail & Carousel Workflow
+======================================
+Tests the FULL production pipeline for post detail and carousel views.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_post_detail_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A post detail page must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ # Use the more realistic fixture if available
+ fixture_name = "home_feed_with_ad.xml"
+ post_xml = _load_fixture(fixture_name)
+ device = E2EDeviceStub([post_xml, post_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on post detail."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on post detail!"
+ )
+
+
+def test_carousel_post_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A carousel post must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ carousel_xml = _load_fixture("carousel_post_dump.xml")
+ device = E2EDeviceStub([carousel_xml, carousel_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on carousel post."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on carousel post!"
+ )
diff --git a/tests/e2e/test_workflow_profiles.py b/tests/e2e/test_workflow_profiles.py
new file mode 100644
index 0000000..8fe474e
--- /dev/null
+++ b/tests/e2e/test_workflow_profiles.py
@@ -0,0 +1,116 @@
+"""
+E2E: Profile Workflows
+========================
+Tests the FULL production pipeline for profile screens:
+- Own profile
+- Other user profile
+- Followers/Following lists
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_user_profile_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A user profile page must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ profile_xml = _load_fixture("user_profile_dump.xml")
+ device = E2EDeviceStub([profile_xml, profile_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on user profile."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on user profile!"
+ )
+
+
+def test_scraping_profile_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The scraping profile dump must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ scrape_xml = _load_fixture("scraping_profile_dump.xml")
+ device = E2EDeviceStub([scrape_xml, scrape_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on scraping profile."
+ )
+
+
+def test_followers_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The followers list must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ followers_xml = _load_fixture("followers_list_dump.xml")
+ device = E2EDeviceStub([followers_xml, followers_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on followers list."
+ )
+
+
+def test_unfollow_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The unfollow list must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ unfollow_xml = _load_fixture("unfollow_list_dump.xml")
+ device = E2EDeviceStub([unfollow_xml, unfollow_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on unfollow list."
+ )
diff --git a/tests/e2e/test_workflow_reels_feed.py b/tests/e2e/test_workflow_reels_feed.py
new file mode 100644
index 0000000..5933517
--- /dev/null
+++ b/tests/e2e/test_workflow_reels_feed.py
@@ -0,0 +1,58 @@
+"""
+E2E: Reels Feed Workflow
+=========================
+Tests the FULL production pipeline when the device shows a Reels post.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_reels_post_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A Reels feed post must be processable by the full pipeline.
+ No plugin may crash on the Reels XML structure.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ reels_xml = _load_fixture("reels_feed_dump.xml")
+ device = E2EDeviceStub([reels_xml, reels_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on a Reels post. "
+ "The pipeline is crashing on Reels XML."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on a Reels feed post!"
+ )
diff --git a/tests/e2e/test_workflow_same_post_loop.py b/tests/e2e/test_workflow_same_post_loop.py
new file mode 100644
index 0000000..64b2a74
--- /dev/null
+++ b/tests/e2e/test_workflow_same_post_loop.py
@@ -0,0 +1,97 @@
+"""
+E2E: Same Post Loop Prevention Regression
+============================================
+PRODUCTION BUG 2026-04-30:
+Bot processed @ehsan.nura 7 times consecutively because:
+1. Each interaction cycle failed (follow failed, like failed, username tap failed)
+2. perfect_snapping kept swiping to the same post (35x snapping attempts)
+3. post_interaction moved to "next post" but snapping brought it right back
+
+The pipeline must detect when it's stuck processing the same post
+repeatedly and break out of the loop.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+E2E_FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name, e2e=False):
+ base = E2E_FIXTURES_DIR if e2e else FIXTURES_DIR
+ with open(os.path.join(base, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_same_xml_does_not_cause_infinite_snapping(monkeypatch, e2e_workflow_ctx):
+ """
+ REGRESSION: When the device always returns the same XML (stuck state),
+ perfect_snapping must not endlessly swipe trying to align.
+
+ Simulates: device is stuck showing the same post. After each swipe,
+ the XML is identical. Snapping must give up, not loop 35 times.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ # Same XML returned every time — simulates a stuck device
+ reels_xml = _load_fixture("reels_feed_dump.xml")
+ device = E2EDeviceStub([reels_xml] * 10)
+ results, ctx = e2e_workflow_ctx(device)
+
+ # Perfect snapping must not have swiped more than a reasonable limit
+ # In the real bug, it swiped 35 times on the same post
+ assert len(device.swipes) <= 10, (
+ f"Perfect snapping performed {len(device.swipes)} swipes on identical XML — "
+ "this is the infinite snapping loop bug! "
+ "Expected max 10 swipes before giving up."
+ )
+
+
+def test_pipeline_does_not_crash_on_repeated_failures(monkeypatch, e2e_workflow_ctx):
+ """
+ When multiple plugins fail on the same post (like fails, follow fails,
+ username tap fails), the pipeline must still complete without exceptions.
+ It should gracefully move to the next post or terminate.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ home_xml = _load_fixture("home_feed_real.xml", e2e=True)
+ device = E2EDeviceStub([home_xml, home_xml, home_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ # The pipeline must complete (we got results, no exception)
+ assert results is not None, "Pipeline crashed!"
+ assert len(results) >= 1, "No plugin results returned!"
+
+ # No CONTEXT_LOST should be raised on a normal home feed
+ context_lost = any(
+ r.metadata.get("return_code") == "CONTEXT_LOST" for r in results
+ )
+ assert not context_lost, (
+ "CONTEXT_LOST on a normal home feed — false positive!"
+ )
diff --git a/tests/e2e/test_workflow_search_feed.py b/tests/e2e/test_workflow_search_feed.py
new file mode 100644
index 0000000..dd745bc
--- /dev/null
+++ b/tests/e2e/test_workflow_search_feed.py
@@ -0,0 +1,56 @@
+"""
+E2E: Search Feed Workflow
+==========================
+Tests the FULL production pipeline when the device shows the Search feed.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_search_feed_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ The Search feed must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ search_xml = _load_fixture("search_feed_dump.xml")
+ device = E2EDeviceStub([search_xml, search_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on Search feed."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on Search feed!"
+ )
diff --git a/tests/e2e/test_workflow_stories_feed.py b/tests/e2e/test_workflow_stories_feed.py
new file mode 100644
index 0000000..bb4257d
--- /dev/null
+++ b/tests/e2e/test_workflow_stories_feed.py
@@ -0,0 +1,75 @@
+"""
+E2E: Stories Feed Workflow
+===========================
+Tests the FULL production pipeline when the device shows a Story view.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
+
+
+def _load_fixture(name):
+ with open(os.path.join(FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_stories_feed_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A Stories feed dump must be processable without crashes.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ stories_xml = _load_fixture("stories_feed_dump.xml")
+ device = E2EDeviceStub([stories_xml, stories_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on Stories feed."
+ )
+
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard false positive on Stories feed!"
+ )
+
+
+def test_story_view_full_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A single story view (full-screen playback) must be processable.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ story_xml = _load_fixture("story_view_full.xml")
+ device = E2EDeviceStub([story_xml, story_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on Story view."
+ )
diff --git a/tests/e2e/test_workflow_stuck_other_profile.py b/tests/e2e/test_workflow_stuck_other_profile.py
new file mode 100644
index 0000000..81359c4
--- /dev/null
+++ b/tests/e2e/test_workflow_stuck_other_profile.py
@@ -0,0 +1,130 @@
+"""
+E2E: Stuck on Other Profile Regression
+========================================
+PRODUCTION BUG 2026-04-30:
+Bot landed on OTHER_PROFILE screen during a ReelsFeed interaction session.
+The pipeline kept cycling through plugins that all returned "Cannot X on
+other_profile" without ever pressing BACK or returning CONTEXT_LOST.
+
+Pattern observed:
+- likes plugin: "Cannot 'tap like button' on other_profile"
+- repost plugin: "Cannot 'share to story' on other_profile"
+- follow plugin: "Cannot 'tap Follow button' on other_profile"
+- profile_visit: Keeps trying to visit the same profile
+- 35x snapping attempts on the same post
+
+Root cause: The screen was classified as NORMAL (score 0.96) even though
+the bot was on OTHER_PROFILE. The SAE only classified the raw visual
+state, not the screen's structural type. The obstacle_guard only checks
+SituationType, not ScreenType mismatches.
+
+This test verifies that when ALL interaction plugins fail on a screen,
+the pipeline must signal a context problem rather than silently loop.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+import os
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+E2E_FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+def _load_e2e_fixture(name):
+ with open(os.path.join(E2E_FIXTURES_DIR, name), "r", encoding="utf-8") as f:
+ return f.read()
+
+
+class FakeSAENormal:
+ """
+ Reproduces the production bug: SAE returns NORMAL even though
+ we're on OTHER_PROFILE. This is the EXACT behavior observed.
+ """
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_other_profile_does_not_infinite_loop(monkeypatch, e2e_workflow_ctx):
+ """
+ REGRESSION: When stuck on OTHER_PROFILE, the pipeline must not
+ silently succeed with 0 interactions and 0 errors — it must signal
+ that something is wrong (via metadata or should_skip).
+
+ The pipeline MUST NOT:
+ - Perform interactions on a profile page as if it were a feed post
+ - Silently succeed while doing nothing useful
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ profile_xml = _load_e2e_fixture("other_profile_real.xml")
+ device = E2EDeviceStub([profile_xml, profile_xml, profile_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ # The pipeline ran — check what happened
+ executed_plugins = [r for r in results if r.executed]
+
+ # KEY ASSERTION: The bot must NOT perform like/follow/comment
+ # interactions against a profile page as if it were a feed post.
+ # If interaction count > 0, the bot is hallucinating feed elements
+ # on a profile page.
+ interaction_results = [r for r in results if r.interactions > 0]
+ hallucinated_interactions = [
+ r for r in interaction_results
+ if r.metadata.get("action") in ("tap like button", "tap follow button", "share to story")
+ ]
+
+ # Verify the pipeline didn't crash silently
+ assert len(executed_plugins) >= 1, (
+ "No plugins executed at all — the pipeline is dead!"
+ )
+
+ # Count the snapping attempts as a proxy for "stuck in loop"
+ # In the real bug, we saw 35 snapping attempts
+ snap_swipes = len(device.swipes)
+ assert snap_swipes <= 10, (
+ f"Perfect snapping performed {snap_swipes} swipes — "
+ "this is the 'stuck on other_profile' infinite snapping loop!"
+ )
+
+
+def test_other_profile_preserves_no_ad_false_positive(monkeypatch, e2e_workflow_ctx):
+ """
+ REGRESSION: The VLM identified a regular profile as an "ad" because
+ the profile bio/grid collage was misclassified. The resonance evaluator
+ must not false-positive on real profiles.
+
+ This tests that ad_guard doesn't fire on a real user profile.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ profile_xml = _load_e2e_fixture("other_profile_real.xml")
+ device = E2EDeviceStub([profile_xml, profile_xml])
+ results, ctx = e2e_workflow_ctx(device)
+
+ # ad_guard must NOT fire on a real user profile
+ ad_guard_results = [
+ r for r in results
+ if r.metadata.get("plugin") == "ad_guard" and r.should_skip
+ ]
+ assert len(ad_guard_results) == 0, (
+ "ad_guard false-positive on a real user profile! "
+ "This was a production bug where profile grid collages were "
+ "misclassified as ads."
+ )