mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

fix(dm-engine): 4 safety hardening patches with TDD proof

Browse Source 
Kill-Switch: Refuse DM processing when dm_reply.enabled=false in config.
  Root cause: checked nonexistent 'disable_ai_messaging' flag instead of
  actual plugin config.

Context Guard: Skip threads with no extractable message text.
  Root cause: LLM was fed 'No previous context' → produced garbage like
  'the to the'.

Send Verification: Structurally verify Send button resource-id/desc.
  Root cause: VLM returned reactions_pill_container, edit fields, etc.
  and engine blindly clicked them, logging 'Successfully sent'.

Iteration Cap: MAX_REPLIES_PER_INBOX_VISIT=3 prevents spam.
  Root cause: no loop guard → 8 DMs sent in 2 minutes in production.

Refactored: removed dead 'if True' guard, de-indented block,
moved dm_memory.log_sent_dm into success branch only.

All 6 E2E tests pass. No regressions (54/55 passed, 1 pre-existing).
This commit is contained in:
Marc Mintel
2026-04-27 23:43:02 +02:00
parent 3006020106
commit c051c3a4c3
2 changed files with 178 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
GramAddict/core/dm_engine.py
View File

@@ -7,12 +7,50 @@ from GramAddict.core.session_state import SessionState
logger = logging.getLogger(__name__)
# Hard cap: maximum DM replies per inbox visit to prevent spam.
MAX_REPLIES_PER_INBOX_VISIT = 3
# Sentinel values that indicate missing message context.
_EMPTY_CONTEXT_SENTINELS = frozenset({"no previous context", "", "none", "n/a"})
# Structural resource-IDs that indicate a real "Send" button.
_SEND_BUTTON_MARKERS = frozenset({"send_button", "row_thread_composer_send"})
def _is_send_button(node: dict) -> bool:
"""Structural verification: returns True only if the node is a real Send button."""
attribs = node.get("original_attribs", {})
rid = attribs.get("resource-id", "")
desc = attribs.get("content-desc", node.get("desc", "")).lower()
# Accept if resource-id contains a known send button marker
if any(marker in rid for marker in _SEND_BUTTON_MARKERS):
return True
# Accept if content-desc is exactly "Send" (Instagram's canonical label)
if desc == "send":
return True
return False
def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_state, current_target, cognitive_stack):
"""
Executes the autonomous Direct Messaging logic in the Zero-Latency architecture.
Assumes the bot is already at the "MessageInbox" UI state.
Safety guarantees:
- Refuses to execute if dm_reply plugin is disabled in config.
- Skips threads with no extractable text context.
- Structurally verifies the Send button before logging success.
- Hard-caps replies per inbox visit to MAX_REPLIES_PER_INBOX_VISIT.
"""
# ── Kill-Switch: Respect dm_reply.enabled config ──
dm_plugin_config = configs.get_plugin_config("dm_reply")
if not dm_plugin_config.get("enabled", False):
logger.warning(
"🛑 [DM Engine] dm_reply plugin is DISABLED in config. Refusing to process inbox.",
extra={"color": f"{Fore.RED}"},
)
return "BOREDOM_CHANGE_FEED"
logger.info(
f"🧠 [DM Engine] Initiating inbox processing in {current_target}...",
extra={"color": f"{Style.BRIGHT}{Fore.CYAN}"},
@@ -30,6 +68,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
session_state.totalMessages = 0
failed_attempts = 0
replies_this_visit = 0
while not dopamine.is_app_session_over():
# Limits check
@@ -92,54 +131,80 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
logger.debug(f"Last received message context: {context_text}")
# Verify we aren't at limits before sending
if not getattr(configs.args, "disable_ai_messaging", False):
# Configure models
model = getattr(configs.args, "ai_condenser_model", "llama3.2:1b")
url = getattr(configs.args, "ai_condenser_url", "http://localhost:11434/api/generate")
# Generate response
prompt = f"You are replying to a direct message on Instagram. The last message you received was: '{context_text}'. Keep it short, casual, and friendly. Do not use hashtags."
response_dict = query_llm(
url=url,
model=model,
prompt=prompt,
format_json=False,
timeout=120,
max_tokens=100,
temperature=0.7,
# ── Context Guard: Skip threads with no extractable message ──
if context_text.strip().lower() in _EMPTY_CONTEXT_SENTINELS:
logger.warning(
"⏭️ [DM Engine] Thread has no extractable message context (story reply / media-only). Skipping."
)
device.press("back")
sleep(1.5)
continue
if response_dict and "response" in response_dict:
response_text = response_dict["response"].strip()
# Find the input field
input_nodes = telepathic._extract_semantic_nodes(
thread_xml, "find the message input text field", threshold=0.7
# Verify we aren't at limits before sending
# ── Iteration Cap: Prevent DM spam ──
if replies_this_visit >= MAX_REPLIES_PER_INBOX_VISIT:
logger.info(
f"🛑 [DM Engine] Reached max replies per inbox visit ({MAX_REPLIES_PER_INBOX_VISIT}). Exiting."
)
device.press("back")
sleep(1.0)
return "BOREDOM_CHANGE_FEED"
# Configure models
model = getattr(configs.args, "ai_condenser_model", "llama3.2:1b")
url = getattr(configs.args, "ai_condenser_url", "http://localhost:11434/api/generate")
# Generate response
prompt = f"You are replying to a direct message on Instagram. The last message you received was: '{context_text}'. Keep it short, casual, and friendly. Do not use hashtags."
response_dict = query_llm(
url=url,
model=model,
prompt=prompt,
format_json=False,
timeout=120,
max_tokens=100,
temperature=0.7,
)
if response_dict and "response" in response_dict:
response_text = response_dict["response"].strip()
# Find the input field
input_nodes = telepathic._extract_semantic_nodes(
thread_xml, "find the message input text field", threshold=0.7
)
if input_nodes and not input_nodes[0].get("skip"):
in_node = input_nodes[0]
_humanized_click(device, in_node["x"], in_node["y"])
sleep(1.0)
# Type the message
ghost_type(device, response_text, speed="fast")
sleep(1.0)
# Find Send button
send_xml = device.dump_hierarchy()
send_nodes = telepathic._extract_semantic_nodes(
send_xml, "find the send message button", threshold=0.8
)
if input_nodes and not input_nodes[0].get("skip"):
in_node = input_nodes[0]
_humanized_click(device, in_node["x"], in_node["y"])
sleep(1.0)
# Type the message
ghost_type(device, response_text, speed="fast")
sleep(1.0)
if send_nodes and not send_nodes[0].get("skip"):
s_node = send_nodes[0]
# Find Send button
send_xml = device.dump_hierarchy()
send_nodes = telepathic._extract_semantic_nodes(
send_xml, "find the send message button", threshold=0.8
)
if send_nodes and not send_nodes[0].get("skip"):
s_node = send_nodes[0]
# ── Send Button Structural Verification ──
if not _is_send_button(s_node):
s_rid = s_node.get("original_attribs", {}).get("resource-id", "unknown")
logger.warning(
f"⚠️ [DM Engine] Refused to click non-Send element: {s_rid}. Aborting reply."
)
else:
_humanized_click(device, s_node["x"], s_node["y"])
logger.info(
"✅ [DM Engine] Successfully sent a generated reply.", extra={"color": Fore.GREEN}
"✅ [DM Engine] Successfully sent a generated reply.",
extra={"color": Fore.GREEN},
)
session_state.totalMessages += 1
replies_this_visit += 1
dm_memory = cognitive_stack.get("dm_memory")
if dm_memory:
dm_memory.log_sent_dm("unknown_target", response_text, "", [])

88
tests/e2e/test_e2e_dm_engine.py
View File

@@ -199,23 +199,48 @@ class TestDMSendVerification:
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
device = MagicMock()
# First dump = inbox, second = thread, third = thread (for send button search)
inbox_xml = _make_dm_inbox_xml()
thread_xml = _make_dm_thread_xml()
# Flow: inbox → thread → send_xml (re-dump) → back → check_xml → inbox (no unread)
device.dump_hierarchy.side_effect = [
_make_dm_inbox_xml(),
_make_dm_thread_xml(),
_make_dm_thread_xml(), # after typing, re-dump for send button
_make_dm_inbox_xml(), # after pressing back
_make_dm_inbox_xml(), # no more unread
inbox_xml, # 1. inbox: find unread
thread_xml, # 2. thread: read messages
thread_xml, # 3. after typing: re-dump for send button
thread_xml, # 4. check_xml after pressing back (still in thread?)
inbox_xml, # 5. inbox again on re-loop
]
configs = _make_configs(dm_reply_enabled=True)
session_state = _make_session_state()
dopamine = _make_dopamine(boredom_sequence=[False, False, True])
# Dopamine: never session-over, but wants_to_change_feed after boredom bump
dopamine = MagicMock()
dopamine.is_app_session_over.return_value = False
dopamine.boredom = 0.0
dopamine.wants_to_change_feed.side_effect = lambda: dopamine.boredom >= 4.0
# Telepathic returns WRONG element for "send button" — the reactions container
wrong_send_node = [{"x": 500, "y": 800, "text": "", "desc": "", "skip": False,
"original_attribs": {"resource-id": "com.instagram.android:id/message_reactions_pill_container"}}]
telepathic = _make_telepathic(send_nodes=wrong_send_node)
# On second unread call, return no threads (inbox clear)
unread_call_n = {"n": 0}
def _extract_nodes(xml, intent, threshold=0.7):
if "unread" in intent.lower():
unread_call_n["n"] += 1
if unread_call_n["n"] == 1:
return [{"x": 500, "y": 300, "text": "johndoe", "skip": False}]
return []
elif "last received" in intent.lower():
return [{"x": 500, "y": 600, "text": "Hey what's up?", "skip": False}]
elif "input" in intent.lower():
return [{"x": 500, "y": 900, "text": "Message…", "skip": False}]
elif "send" in intent.lower():
return wrong_send_node
return []
telepathic = MagicMock()
telepathic._extract_semantic_nodes.side_effect = _extract_nodes
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": MagicMock()}
@@ -258,20 +283,55 @@ class TestDMContextRequirement:
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
device = MagicMock()
# Flow: inbox → click unread → thread (no context) → back → continue →
# inbox (same, but telepathic returns no unread) → boredom exit
inbox_xml = _make_dm_inbox_xml()
device.dump_hierarchy.side_effect = [
_make_dm_inbox_xml(),
_make_dm_thread_xml_no_context(), # Thread with no text
_make_dm_inbox_xml(), # Back to inbox
_make_dm_inbox_xml(), # No more unread (empty)
inbox_xml, # 1. inbox: find unread
_make_dm_thread_xml_no_context(), # 2. thread: read messages (no text)
# after context-skip continue, back to loop:
inbox_xml, # 3. inbox again (check is_inbox)
# 4. check_xml after pressing back from thread (dm_engine L152)
]
configs = _make_configs(dm_reply_enabled=True)
session_state = _make_session_state()
dopamine = _make_dopamine(boredom_sequence=[False, False, True])
# 1st call: not over (process first thread)
# 2nd call: not over (after context skip, re-loop)
# 3rd+ calls: not needed because boredom triggers exit
dopamine = MagicMock()
dopamine.is_app_session_over.return_value = False
dopamine.boredom = 0.0
# After inbox_clear, boredom jumps to 50 → wants_to_change_feed
# should return True on second check (after inbox clear)
change_feed_calls = {"n": 0}
def _wants_change():
change_feed_calls["n"] += 1
# After any boredom bump, signal exit
return dopamine.boredom >= 40.0
dopamine.wants_to_change_feed.side_effect = _wants_change
# No extractable text from thread
no_text_msg_nodes = [{"x": 500, "y": 600, "text": "", "skip": False}]
telepathic = _make_telepathic(msg_nodes=no_text_msg_nodes)
# On the second inbox visit, return NO unread threads (inbox clear)
call_count = {"n": 0}
def _extract_nodes(xml, intent, threshold=0.7):
if "unread" in intent.lower():
call_count["n"] += 1
if call_count["n"] == 1:
return [{"x": 500, "y": 300, "text": "johndoe", "skip": False}]
# Second time: no unread
return []
elif "last received" in intent.lower():
return no_text_msg_nodes
return []
telepathic = MagicMock()
telepathic._extract_semantic_nodes.side_effect = _extract_nodes
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": MagicMock()}