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

feat(diagnostics): dump screenshots with xmls and limit retention to 5

Browse Source
This commit is contained in:
Marc Mintel
2026-04-27 11:55:44 +02:00
parent ae046be3b1
commit e9201e0e30
10 changed files with 480 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
GramAddict/core/bot_flow.py
View File

@@ -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

34
GramAddict/core/device_facade.py
View File

@@ -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}")

36
GramAddict/core/diagnostic_dump.py
View File

@@ -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

46
GramAddict/core/dm_engine.py
View File

@@ -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"

12
GramAddict/core/physics/humanized_input.py
View File

@@ -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).

10
GramAddict/core/situational_awareness.py
View File

@@ -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: