diff --git a/GramAddict/core/bot_flow.py b/GramAddict/core/bot_flow.py
index 08955ea..ede221d 100644
--- a/GramAddict/core/bot_flow.py
+++ b/GramAddict/core/bot_flow.py
@@ -51,7 +51,6 @@ from GramAddict.core.physics.timing import (
wait_for_story_loaded as _wait_for_story_loaded_impl,
)
from GramAddict.core.q_nav_graph import QNavGraph
-from GramAddict.core.qdrant_memory import ParasocialCRMDB
from GramAddict.core.resonance_engine import ResonanceEngine
from GramAddict.core.sensors.honeypot_radome import HoneypotRadome
from GramAddict.core.session_state import SessionState, SessionStateEncoder
@@ -178,8 +177,11 @@ def start_bot(**kwargs):
)
persona_interests = [p.strip() for p in persona_raw.split(",") if p.strip()] if persona_raw else []
+ from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
+
dopamine = DopamineEngine()
crm_db = ParasocialCRMDB()
+ dm_memory_db = DMMemoryDB()
resonance_oracle = ResonanceEngine(username, persona_interests=persona_interests, crm=crm_db)
active_inference = ActiveInferenceEngine(username)
@@ -235,6 +237,7 @@ def start_bot(**kwargs):
"telepathic": telepathic,
"darwin": darwin,
"crm": crm_db,
+ "dm_memory": dm_memory_db,
}
from GramAddict.core.behaviors import PluginRegistry
diff --git a/GramAddict/core/device_facade.py b/GramAddict/core/device_facade.py
index 047a63d..ca8d3b8 100644
--- a/GramAddict/core/device_facade.py
+++ b/GramAddict/core/device_facade.py
@@ -299,19 +299,51 @@ class DeviceFacade:
xml = self.deviceV2.dump_hierarchy(compressed=True)
# Continuous Session Tracing
+ import shutil
from datetime import datetime
try:
+ traces_root = os.path.join("debug", "session_traces")
if not hasattr(self, "_trace_counter"):
self._trace_counter = 0
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
- self._trace_dir = os.path.join("debug", "session_traces", ts)
+ self._trace_dir = os.path.join(traces_root, ts)
os.makedirs(self._trace_dir, exist_ok=True)
+ # Cleanup: keep only last 5 session folders
+ try:
+ if os.path.exists(traces_root):
+ folders = [
+ os.path.join(traces_root, d)
+ for d in os.listdir(traces_root)
+ if os.path.isdir(os.path.join(traces_root, d))
+ ]
+ folders.sort(key=os.path.getmtime)
+ while len(folders) > 5:
+ oldest = folders.pop(0)
+ shutil.rmtree(oldest, ignore_errors=True)
+ logger.info(f"๐งน [Cleanup] Removed old session trace: {oldest}")
+ except Exception as e:
+ logger.debug(f"Failed to cleanup old traces: {e}")
+
self._trace_counter += 1
trace_path = os.path.join(self._trace_dir, f"{self._trace_counter:05d}.xml")
with open(trace_path, "w", encoding="utf-8") as f:
f.write(xml)
+
+ # Dump screenshot as well
+ try:
+ import base64
+
+ screenshot_b64 = self.get_screenshot_b64()
+ if screenshot_b64:
+ screenshot_data = base64.b64decode(screenshot_b64)
+ screenshot_path = trace_path.replace(".xml", ".jpg")
+ with open(screenshot_path, "wb") as f:
+ f.write(screenshot_data)
+ except Exception as e:
+ logger.debug(f"Failed to capture screenshot for session trace: {e}")
+
except Exception as e:
logger.debug(f"Failed to write session trace: {e}")
diff --git a/GramAddict/core/diagnostic_dump.py b/GramAddict/core/diagnostic_dump.py
index 97446ee..375786b 100644
--- a/GramAddict/core/diagnostic_dump.py
+++ b/GramAddict/core/diagnostic_dump.py
@@ -18,19 +18,11 @@ from datetime import datetime
logger = logging.getLogger(__name__)
DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "debug", "xml_dumps")
-MAX_DUMPS_PER_CATEGORY = 50
-
+MAX_DUMPS_PER_CATEGORY = 5
def dump_ui_state(device, reason: str, extra_context: dict = None):
"""
- Capture and save the current UI hierarchy to disk for debugging.
-
- Args:
- device: The uiautomator2 device facade.
- reason: Short tag for the failure type. Used for filename grouping.
- Examples: 'context_lost', 'vlm_hallucination', 'nav_failure',
- 'stuck_on_post', 'unexpected_screen'
- extra_context: Optional dict with additional metadata (intent, expected state, etc.)
+ Capture and save the current UI hierarchy and screenshot to disk for debugging.
"""
try:
os.makedirs(DUMP_DIR, exist_ok=True)
@@ -48,16 +40,28 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
with open(filepath, "w", encoding="utf-8") as f:
f.write(xml)
+ # Capture and write screenshot
+ try:
+ import base64
+ screenshot_b64 = device.screenshot_b64()
+ if screenshot_b64:
+ screenshot_data = base64.b64decode(screenshot_b64)
+ screenshot_path = filepath.replace(".xml", ".jpg")
+ with open(screenshot_path, "wb") as f:
+ f.write(screenshot_data)
+ except Exception as e:
+ logger.debug(f"[Diagnostic] Could not capture screenshot: {e}")
+
# Write companion metadata JSON
meta = {
"reason": reason,
"timestamp": ts,
"xml_file": filename,
+ "screenshot_file": filename.replace(".xml", ".jpg")
}
# Capture the session log if available
try:
import shutil
-
from GramAddict.core.log import get_log_file_config
log_name, log_dir, _, _ = get_log_file_config()
@@ -77,7 +81,7 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
with open(meta_path, "w", encoding="utf-8") as f:
json.dump(meta, f, indent=2, ensure_ascii=False)
- logger.info(f"๐ธ [Diagnostic] UI state and session log dumped for '{reason}': {filepath}")
+ logger.info(f"๐ธ [Diagnostic] UI state, screenshot, and session log dumped for '{reason}': {filepath}")
# Rotate old dumps for this category
_rotate_dumps(safe_reason)
@@ -89,7 +93,6 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
logger.debug(f"[Diagnostic] Could not dump UI state: {e}")
return None
-
def _rotate_dumps(category_prefix: str):
"""Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category."""
try:
@@ -100,8 +103,15 @@ def _rotate_dumps(category_prefix: str):
for f in files_to_remove:
xml_path = os.path.join(DUMP_DIR, f)
meta_path = xml_path.replace(".xml", ".meta.json")
+ log_path = xml_path.replace(".xml", ".log")
+ img_path = xml_path.replace(".xml", ".jpg")
+
os.remove(xml_path)
if os.path.exists(meta_path):
os.remove(meta_path)
+ if os.path.exists(log_path):
+ os.remove(log_path)
+ if os.path.exists(img_path):
+ os.remove(img_path)
except Exception:
pass
diff --git a/GramAddict/core/dm_engine.py b/GramAddict/core/dm_engine.py
index 3390603..085af53 100644
--- a/GramAddict/core/dm_engine.py
+++ b/GramAddict/core/dm_engine.py
@@ -20,7 +20,6 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
telepathic = cognitive_stack.get("telepathic")
dopamine = cognitive_stack.get("dopamine")
- crm = cognitive_stack.get("crm")
from GramAddict.core.bot_flow import _humanized_click, sleep
from GramAddict.core.llm_provider import query_llm
@@ -44,6 +43,31 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
try:
xml_dump = device.dump_hierarchy()
+ # --- Zero Trust Structural Guard ---
+ # If the navigation engine failed or the UI shifted, we MUST NOT hallucinate
+ # interactions on the wrong screen (e.g., Privacy Settings).
+ is_thread = "direct_thread_header" in xml_dump or "row_thread_composer_edittext" in xml_dump
+ is_inbox = (
+ "action_bar_title" in xml_dump
+ or "thread_list" in xml_dump
+ or "direct_inbox_header" in xml_dump
+ or "unread" in xml_dump.lower()
+ )
+
+ if is_thread:
+ logger.warning("โ ๏ธ [Structural Guard] DM Engine trapped in an open thread. Escaping...")
+ device.press("back")
+ from GramAddict.core.bot_flow import sleep
+
+ sleep(1.5)
+ continue
+
+ if not is_inbox and not is_thread:
+ # We have drifted somewhere entirely alien (like Privacy Settings)
+ logger.error("๐ [Structural Guard] Alien context detected. Not in Inbox. Triggering CONTEXT_LOST.")
+ return "CONTEXT_LOST"
+ # -----------------------------------
+
# Step 1: Find unread conversation threads
unread_threads = telepathic._extract_semantic_nodes(
xml_dump, "find unread message threads or unread badges", threshold=0.7
@@ -115,12 +139,19 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
)
session_state.totalMessages += 1
- if crm:
- crm.log_sent_dm("unknown_target", response_text, "", [])
+ dm_memory = cognitive_stack.get("dm_memory")
+ if dm_memory:
+ dm_memory.log_sent_dm("unknown_target", response_text, "", [])
# Return back to inbox
device.press("back")
- sleep(1.0)
+ sleep(1.5)
+
+ # If keyboard was open, the first back only closed it. Check if still in thread.
+ check_xml = device.dump_hierarchy()
+ if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
+ device.press("back")
+ sleep(1.0)
dopamine.boredom += random.uniform(5.0, 15.0)
failed_attempts = 0
@@ -136,6 +167,13 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
except Exception as e:
logger.error(f"โ ๏ธ [Anomaly Handler] Exception in DM Loop: {e}")
device.press("back")
+ sleep(1.0)
+
+ check_xml = device.dump_hierarchy()
+ if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
+ device.press("back")
+ sleep(1.0)
+
failed_attempts += 1
if failed_attempts > 2:
return "CONTEXT_LOST"
diff --git a/GramAddict/core/physics/humanized_input.py b/GramAddict/core/physics/humanized_input.py
index ec86da0..f4deec4 100644
--- a/GramAddict/core/physics/humanized_input.py
+++ b/GramAddict/core/physics/humanized_input.py
@@ -151,16 +151,12 @@ def humanized_scroll(device, is_skip=False, resonance_score=None):
def humanized_click(device, x, y, double=False, sleep_mod=1.0):
"""Simulates a human tap with biomechanical jitter and micro-drift."""
- body = PhysicsBody.get_session_instance(device)
- injector = SendEventInjector.get_instance(device)
def single_tap():
- points = BezierGesture.tap_curve(x, y, body)
- # Tap timing: 40-90ms contact time
- tap_duration = random.uniform(40, 90)
- timing = BezierGesture.compute_sigmoid_timing(len(points), tap_duration)
-
- injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
+ # Apply biomechanical jitter
+ jx = int(x + random.gauss(0, 5))
+ jy = int(y + random.gauss(0, 5))
+ device.shell(f"input tap {jx} {jy}")
if double:
# For double tap, the timing is extremely critical (<300ms between taps).
diff --git a/GramAddict/core/situational_awareness.py b/GramAddict/core/situational_awareness.py
index b04b4b9..9baaf13 100644
--- a/GramAddict/core/situational_awareness.py
+++ b/GramAddict/core/situational_awareness.py
@@ -418,13 +418,15 @@ class SituationalAwarenessEngine:
"reel_camera", # Reel recording interface
)
- # Guard: Check against compressed string to ensure these markers ONLY appear
- # as resource IDs (e.g. "id=quick_capture_...") and not as plain text in
- # user comments/bios (which would look like "text='... creation_flow ...'")
- if any(re.search(rf"id=[^\s|]*{marker}", compressed, re.IGNORECASE) for marker in creation_flow_markers):
+ # Guard: Use the RAW xml_dump to avoid truncation of root containers (Z-index filtering),
+ # but ensure we only match inside resource-id attributes to prevent false positives from user text.
+ if any(
+ re.search(rf'resource-id="[^"]*{marker}[^"]*"', xml_dump, re.IGNORECASE) for marker in creation_flow_markers
+ ):
logger.info("๐ง [SAE Perceive] Content-creation overlay detected structurally โ OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
+
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
diff --git a/benchmarks/run_competitive_benchmark.py b/benchmarks/run_competitive_benchmark.py
index 7683a95..47ccbe1 100644
--- a/benchmarks/run_competitive_benchmark.py
+++ b/benchmarks/run_competitive_benchmark.py
@@ -100,7 +100,7 @@ def get_installed_ollama_models():
return []
-def benchmark_model(model_name: str, url: str, force: bool = False):
+def benchmark_model(model_name: str, url: str, force: bool = False, iterations: int = 3):
db = load_json(BENCHMARKS_FILE) or {"models": {}}
scenarios_data = load_json(SCENARIOS_FILE)
if not scenarios_data:
@@ -138,49 +138,69 @@ def benchmark_model(model_name: str, url: str, force: bool = False):
"Return: {\"index\": number, \"reason\": \"...\"}"
)
- start_time = time.time()
- try:
- resp_str = query_telepathic_llm(model_name, url, system_prompt, user_prompt)
- latency = int((time.time() - start_time) * 1000)
- total_latency += latency
- except Exception as e:
- print(f" โ API Request failed for scenario {scenario['id']}: {e}")
- passed_all = False
- continue
+ scenario_latencies = []
+ scenario_scores = []
+ successes = 0
- raw_points = 0
- try:
- clean = resp_str.strip()
- if clean.startswith("```json"):
- clean = clean[7:]
- if clean.endswith("```"):
- clean = clean[:-3]
- data = json.loads(clean)
-
- # Points for structural adherence
- if "index" in data and "reason" in data:
- raw_points += 40
-
- # Points for correctness
- if data["index"] == scenario["target_index"]:
- raw_points += 60
- print(f" โ
Correct index ({data['index']}).")
- else:
- passed_all = False
- print(f" โ Wrong index ({data['index']}). Target was {scenario['target_index']}.")
- else:
+ for _ in range(iterations):
+ start_time = time.time()
+ try:
+ resp_str = query_telepathic_llm(model_name, url, system_prompt, user_prompt)
+ latency = int((time.time() - start_time) * 1000)
+ scenario_latencies.append(latency)
+ except Exception as e:
+ print(f" โ API Request failed for scenario {scenario['id']}: {e}")
passed_all = False
- print(" โ JSON missing fields.")
- except Exception:
- passed_all = False
- print(" โ JSON Parsing failed.")
+ continue
- results_detail[scenario["id"]] = raw_points
- total_raw += raw_points
+ raw_points = 0
+ try:
+ clean = resp_str.strip()
+ if clean.startswith("```json"):
+ clean = clean[7:]
+ if clean.endswith("```"):
+ clean = clean[:-3]
+ data = json.loads(clean)
+
+ # Points for structural adherence
+ if "index" in data and "reason" in data:
+ raw_points += 40
+
+ # Points for correctness
+ if data["index"] == scenario["target_index"]:
+ raw_points += 60
+ successes += 1
+ else:
+ print(f" โ Wrong index ({data.get('index')}). Target was {scenario['target_index']}.")
+ else:
+ print(" โ JSON missing fields.")
+ except Exception:
+ print(" โ JSON Parsing failed.")
+
+ scenario_scores.append(raw_points)
+
+ avg_scenario_score = int(sum(scenario_scores) / len(scenario_scores)) if scenario_scores else 0
+ avg_scenario_latency = int(sum(scenario_latencies) / len(scenario_latencies)) if scenario_latencies else 0
+
+ pass_rate = (successes / iterations) * 100
+ if pass_rate < 100.0:
+ passed_all = False
+
+ print(
+ f" Result: {pass_rate:.0f}% Pass Rate | Avg Score: {avg_scenario_score}/100 | Avg Latency: {avg_scenario_latency}ms"
+ )
+
+ results_detail[scenario["id"]] = {
+ "avg_score": avg_scenario_score,
+ "pass_rate": pass_rate,
+ "latency": avg_scenario_latency,
+ }
+ total_raw += avg_scenario_score
+ total_latency += avg_scenario_latency
avg_latency = total_latency // len(scenarios) if scenarios else 0
print(
- f"\n๐ {model_name} Result: {'PASS' if passed_all else 'FAIL'} | Score: {total_raw} | Latency: {avg_latency}ms"
+ f"\n๐ {model_name} Result: {'PASS' if passed_all else 'FAIL'} | Avg Score: {total_raw} | Latency: {avg_latency}ms"
)
if model_name not in db["models"]:
@@ -212,6 +232,9 @@ if __name__ == "__main__":
parser.add_argument("--url", type=str, help="Explicit endpoint URL")
parser.add_argument("--force", action="store_true", help="Force re-testing")
parser.add_argument("--all-ollama", action="store_true", help="Automatically find and test all local Ollama models")
+ parser.add_argument(
+ "--iterations", type=int, default=3, help="Number of iterations per scenario to measure reliability"
+ )
args, unknown = parser.parse_known_args()
@@ -241,5 +264,5 @@ if __name__ == "__main__":
sys.exit(1)
for m, u in set(models_to_test):
- benchmark_model(m, u, args.force)
+ benchmark_model(m, u, args.force, args.iterations)
time.sleep(1)
diff --git a/tests/e2e/test_e2e_dm_engine.py b/tests/e2e/test_e2e_dm_engine.py
new file mode 100644
index 0000000..cb913b7
--- /dev/null
+++ b/tests/e2e/test_e2e_dm_engine.py
@@ -0,0 +1,114 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
+from GramAddict.core.session_state import SessionState
+
+
+@pytest.fixture
+def mock_device():
+ device = MagicMock()
+ # Initial inbox state
+ device.dump_hierarchy.return_value = ""
+ return device
+
+
+@pytest.fixture
+def mock_cognitive_stack():
+ telepathic = MagicMock()
+ dopamine = MagicMock()
+ dopamine.is_app_session_over.return_value = False
+ dopamine.wants_to_change_feed.side_effect = [False, True]
+ dopamine.boredom = 0
+
+ dm_memory = MagicMock()
+
+ resonance = MagicMock()
+ resonance.persona_prompt = "You are a friendly bot."
+ resonance.args.ai_model = "test-model"
+
+ return {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": dm_memory, "resonance": resonance}
+
+
+def test_e2e_dm_full_flow_success(mock_device, mock_cognitive_stack):
+ """
+ E2E scenario:
+ 1. Found 1 unread message.
+ 2. Opened chat.
+ 3. Read context.
+ 4. Generated response.
+ 5. Sent message.
+ 6. Guarded back-navigation (keyboard closed + activity exit).
+ """
+ telepathic = mock_cognitive_stack["telepathic"]
+
+ hierarchy_items = [
+ "Inbox with unread", # Loop 1 start
+ "Thread view", # Context read
+ "Thread view", # Input field find
+ "Thread view", # Send button find
+ "", # Navigation check AFTER back
+ "Inbox View", # Loop 2 start (exit)
+ "Inbox View", # Buffer
+ ]
+ hierarchy_iterator = iter(hierarchy_items)
+ mock_device.dump_hierarchy.side_effect = lambda: next(hierarchy_iterator)
+
+ # Semantic node responses
+ telepathic._extract_semantic_nodes.side_effect = [
+ [{"x": 100, "y": 100, "text": "New Message"}], # unread_threads
+ [{"text": "Hello there!"}], # msg_nodes (context)
+ [{"x": 200, "y": 200}], # input_nodes
+ [{"x": 300, "y": 300}], # send_nodes
+ [], # Loop 2: no unread
+ [], # Buffer
+ ]
+
+ mock_cognitive_stack["dopamine"].boredom = 0
+ mock_cognitive_stack["dopamine"].wants_to_change_feed.side_effect = [False, True, True]
+
+ session_state = MagicMock(spec=SessionState)
+ session_state.check_limit.return_value = False
+ session_state.totalMessages = 0
+
+ mock_configs = MagicMock()
+ mock_configs.args.disable_ai_messaging = False
+ mock_configs.args.ai_condenser_model = "test-model"
+ mock_configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
+
+ with (
+ patch("GramAddict.core.llm_provider.query_llm", return_value={"response": "Hi! How can I help?"}),
+ patch("GramAddict.core.bot_flow._humanized_click"),
+ patch("GramAddict.core.bot_flow.sleep"),
+ patch("GramAddict.core.stealth_typing.ghost_type"),
+ ):
+ result = _run_zero_latency_dm_loop(
+ mock_device, MagicMock(), MagicMock(), mock_configs, session_state, "target", mock_cognitive_stack
+ )
+
+ assert result == "BOREDOM_CHANGE_FEED"
+ # Ensure navigation at least attempted to exit
+ assert mock_device.press.call_count >= 2
+ mock_device.press.assert_called_with("back")
+
+
+def test_e2e_dm_no_messages(mock_device, mock_cognitive_stack):
+ """
+ E2E scenario: No messages found, exit immediately.
+ """
+ telepathic = mock_cognitive_stack["telepathic"]
+ mock_cognitive_stack["dopamine"].wants_to_change_feed.return_value = True
+
+ telepathic._extract_semantic_nodes.return_value = [] # No unreads
+
+ session_state = MagicMock(spec=SessionState)
+ session_state.check_limit.return_value = False
+
+ result = _run_zero_latency_dm_loop(
+ mock_device, MagicMock(), MagicMock(), MagicMock(), session_state, "target", mock_cognitive_stack
+ )
+
+ assert result == "BOREDOM_CHANGE_FEED"
+ # Should only press back once to exit Inbox
+ assert mock_device.press.call_count == 1
diff --git a/tests/unit/test_diagnostic_dump_cleanup.py b/tests/unit/test_diagnostic_dump_cleanup.py
new file mode 100644
index 0000000..eef4c59
--- /dev/null
+++ b/tests/unit/test_diagnostic_dump_cleanup.py
@@ -0,0 +1,29 @@
+import os
+import shutil
+import base64
+from unittest.mock import MagicMock, patch
+from GramAddict.core.behaviors.simulator import BehaviorSimulator
+
+def test_device_facade_session_trace_cleanup(tmp_path):
+ facade = BehaviorSimulator()
+ facade.get_screenshot_b64 = MagicMock(return_value=base64.b64encode(b"fake_jpg_data").decode("utf-8"))
+
+ with patch("GramAddict.core.device_facade.os.path.join", side_effect=os.path.join), \
+ patch("GramAddict.core.device_facade.os.makedirs"):
+
+ def fake_join(*args):
+ if args[0] == "debug" and args[1] == "session_traces":
+ return str(tmp_path / "session_traces")
+ return os.path.join(*args)
+
+ with patch("GramAddict.core.device_facade.os.path.join", side_effect=fake_join):
+ traces_root = tmp_path / "session_traces"
+ traces_root.mkdir(parents=True, exist_ok=True)
+
+ for i in range(6):
+ d = traces_root / f"old_session_{i}"
+ d.mkdir()
+
+ facade.dump_hierarchy()
+ remaining_folders = [f for f in os.listdir(traces_root) if os.path.isdir(os.path.join(traces_root, f))]
+ assert len(remaining_folders) <= 5
diff --git a/tests/unit/test_dm_navigation_guards.py b/tests/unit/test_dm_navigation_guards.py
new file mode 100644
index 0000000..6edc3d5
--- /dev/null
+++ b/tests/unit/test_dm_navigation_guards.py
@@ -0,0 +1,163 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
+from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
+
+
+def test_parasocial_crm_missing_log_sent_dm():
+ """
+ RED: This test proves that ParasocialCRMDB lacks log_sent_dm,
+ which caused the crash in the last production run.
+ """
+ crm = ParasocialCRMDB()
+ with pytest.raises(AttributeError) as excinfo:
+ crm.log_sent_dm("test_user", "hello", "bio", [])
+
+ assert "'ParasocialCRMDB' object has no attribute 'log_sent_dm'" in str(excinfo.value)
+
+
+def test_dm_engine_uses_correct_dm_memory_logging(monkeypatch):
+ """
+ RED: This test checks if dm_engine correctly uses dm_memory from cognitive_stack.
+ Note: I am writing this to PROVE the fix is necessary.
+ """
+ mock_device = MagicMock()
+ mock_device.dump_hierarchy.return_value = ""
+
+ mock_telepathic = MagicMock()
+ # Mock unread thread found
+ mock_thread = {"x": 100, "y": 100, "text": "Mariischen"}
+
+ def side_effect_logging(*args, **kwargs):
+ res = next(iterator)
+ print(f"DEBUG: extract_semantic_nodes called with {args[1]}. Returning {res}")
+ return res
+
+ iterator = iter(
+ [
+ [mock_thread], # Step 1: unread threads
+ [{"text": "Hey"}], # Step 2: context
+ [{"x": 200, "y": 200}], # Step 3: input field
+ [{"x": 300, "y": 300}], # Step 4: send button
+ [], # Step 5: next iteration no unread
+ ]
+ )
+ mock_telepathic._extract_semantic_nodes.side_effect = side_effect_logging
+
+ mock_dopamine = MagicMock()
+ mock_dopamine.is_app_session_over.return_value = False
+ mock_dopamine.wants_to_change_feed.return_value = True # exit after one
+ mock_dopamine.boredom = 0
+
+ mock_crm = ParasocialCRMDB()
+ mock_dm_memory = MagicMock(spec=DMMemoryDB)
+ mock_resonance = MagicMock()
+ mock_resonance.persona_prompt = "Persona prompt"
+ mock_resonance.args.ai_model = "test-model"
+
+ cognitive_stack = {
+ "telepathic": mock_telepathic,
+ "dopamine": mock_dopamine,
+ "crm": mock_crm,
+ "dm_memory": mock_dm_memory,
+ "resonance": mock_resonance,
+ }
+
+ mock_session = MagicMock()
+ mock_session.check_limit.return_value = False
+ mock_session.totalMessages = 0
+
+ # Mock LLM response
+ monkeypatch.setattr("GramAddict.core.llm_provider.query_llm", lambda **k: {"response": "hi"})
+ monkeypatch.setattr("GramAddict.core.bot_flow._humanized_click", lambda *a: None)
+ monkeypatch.setattr("GramAddict.core.bot_flow.sleep", lambda *a: None)
+ monkeypatch.setattr("GramAddict.core.stealth_typing.ghost_type", lambda *a, **k: None)
+
+ mock_configs = MagicMock()
+ mock_configs.args.disable_ai_messaging = False
+ mock_configs.args.ai_condenser_model = "test-model"
+ mock_configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
+
+ # This should NOT crash now because I fixed it, but we are testing the logic.
+
+ _run_zero_latency_dm_loop(
+ mock_device, MagicMock(), MagicMock(), mock_configs, mock_session, "target", cognitive_stack
+ )
+
+ # Verify dm_memory was used, NOT crm
+ mock_dm_memory.log_sent_dm.assert_called_once()
+
+
+def test_dm_navigation_double_back_guard():
+ """
+ Verifies that if we are still in a thread after one back press,
+ we press back again.
+ """
+ mock_device = MagicMock()
+ # Mocking hierarchy sequence
+ mock_device.dump_hierarchy.side_effect = [
+ "Inbox", # Loop 1 start
+ "Thread", # Context read
+ "Thread", # Send button find
+ "", # Navigation check AFTER back
+ "Inbox", # Loop 2 start (exit)
+ "Inbox", # Buffer
+ ]
+
+ mock_telepathic = MagicMock()
+ mock_telepathic._extract_semantic_nodes.side_effect = [
+ [{"x": 1, "y": 1}], # unread found in Loop 1
+ [{"text": "msg"}], # context
+ [{"x": 2, "y": 2}], # input
+ [{"x": 3, "y": 3}], # send
+ [], # Loop 2: no unread
+ [], # Buffer
+ ]
+
+ mock_dopamine = MagicMock()
+ mock_dopamine.is_app_session_over.return_value = False
+ mock_dopamine.wants_to_change_feed.side_effect = [False, True, True] # exit after Loop 1
+ mock_dopamine.boredom = 0
+
+ mock_resonance = MagicMock()
+ mock_resonance.persona_prompt = "Persona"
+ mock_resonance.args.ai_model = "model"
+
+ cognitive_stack = {
+ "telepathic": mock_telepathic,
+ "dopamine": mock_dopamine,
+ "dm_memory": MagicMock(),
+ "resonance": mock_resonance,
+ }
+
+ mock_session = MagicMock()
+ mock_session.check_limit.return_value = False
+ mock_session.totalMessages = 0
+
+ mock_configs = MagicMock()
+ mock_configs.args.disable_ai_messaging = False
+ mock_configs.args.ai_condenser_model = "test-model"
+ mock_configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
+
+ from unittest.mock import patch
+
+ import GramAddict.core.dm_engine as dm_engine
+
+ with (
+ patch("GramAddict.core.llm_provider.query_llm", return_value={"response": "hi"}),
+ patch("GramAddict.core.bot_flow._humanized_click"),
+ patch("GramAddict.core.bot_flow.sleep"),
+ patch("GramAddict.core.stealth_typing.ghost_type"),
+ ):
+ dm_engine._run_zero_latency_dm_loop(
+ mock_device, MagicMock(), MagicMock(), mock_configs, mock_session, "target", cognitive_stack
+ )
+
+ # Expected calls:
+ # 1. First back from success flow (thread -> inbox)
+ # 2. Second back from guard check (if still in thread)
+ # 3. Third back from inbox exit (boredom check)
+ assert mock_device.press.call_count == 3
+ mock_device.press.assert_called_with("back")