From e9201e0e308b83c28958c068052b17999098e7e2 Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Mon, 27 Apr 2026 11:55:44 +0200 Subject: [PATCH] feat(diagnostics): dump screenshots with xmls and limit retention to 5 --- GramAddict/core/bot_flow.py | 5 +- GramAddict/core/device_facade.py | 34 ++++- GramAddict/core/diagnostic_dump.py | 36 +++-- GramAddict/core/dm_engine.py | 46 +++++- GramAddict/core/physics/humanized_input.py | 12 +- GramAddict/core/situational_awareness.py | 10 +- benchmarks/run_competitive_benchmark.py | 101 ++++++++----- tests/e2e/test_e2e_dm_engine.py | 114 ++++++++++++++ tests/unit/test_diagnostic_dump_cleanup.py | 29 ++++ tests/unit/test_dm_navigation_guards.py | 163 +++++++++++++++++++++ 10 files changed, 480 insertions(+), 70 deletions(-) create mode 100644 tests/e2e/test_e2e_dm_engine.py create mode 100644 tests/unit/test_diagnostic_dump_cleanup.py create mode 100644 tests/unit/test_dm_navigation_guards.py 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")