async def run_auto_loop(
sock,
state: dict,
broadcast_func,
interval: float = 10.0,
max_loops=math.inf,
benchmark: Optional[Benchmark] = None,
persistence=None,
run_state=None,
mgba_proc: Optional[Any] = None,
):
"""Main async loop: Get state, call LLM, send action, update/broadcast state.
Args:
mgba_proc: The mGBA subprocess - needed for auto-restart on failures for 24/7 operation.
"""
global \
action_count, \
tokens_used_session, \
start_time, \
SCREENSHOT_PATH, \
MINIMAP_PATH, \
SAVED_SCREENSHOT_PATH, \
SAVED_MINIMAP_PATH
cycle_count = 0
# Track if we're resuming a run - don't increment cycle on first iteration
# because the restored cycle was interrupted and should be completed first
is_first_cycle_after_continue = False
# Restore state from persistence if available
if run_state and run_state.action_count > 0:
cycle_count = run_state.cycle_count
action_count = run_state.action_count
tokens_used_session = run_state.tokens_used
# Restore elapsed time by adjusting start_time
start_time = datetime.datetime.now() - datetime.timedelta(
seconds=run_state.elapsed_seconds
)
# Restore chat history logic deferred
restored_history = (
run_state.chat_history if run_state and run_state.chat_history else []
)
if restored_history:
log.info(
f"🔄 Found persistent chat history: {len(restored_history)} messages"
)
log.info(
f"🔄 Restored from persistence: cycle={cycle_count}, actions={action_count}, tokens={tokens_used_session}"
)
# Flag to skip cycle increment on first loop iteration
is_first_cycle_after_continue = True
# Initialize memory manager - persist memories when continuing a run
is_continuing_run = run_state and run_state.action_count > 0
memory_manager = MemoryManager(reset_on_start=not is_continuing_run)
if is_continuing_run:
log.info("📝 Memory manager: Loaded existing memories (continuing run)")
else:
log.info("📝 Memory manager: Fresh start (new run)")
# Initialize goal tracker
goal_tracker = GoalTracker()
if not goal_tracker.goals:
goal_tracker.initialize_default_goals()
log.info("Goal tracker initialized")
# Initialize exploration tracker - persist when continuing a run
exploration_tracker = ExplorationTracker(reset_on_start=not is_continuing_run)
if is_continuing_run:
log.info(
f"🗺️ Exploration tracker: Loaded existing data ({len(exploration_tracker.maps)} maps)"
)
else:
log.info("🗺️ Exploration tracker: Fresh start")
# Initialize achievement tracker for tweet generation milestones
achievement_tracker = create_achievement_tracker(
storage_path="data/achievements.json",
reset_on_start=not is_continuing_run,
)
if is_continuing_run:
log.info(
f"🏆 Achievement tracker: Loaded {len(achievement_tracker.triggered)} achievements"
)
else:
log.info("🏆 Achievement tracker: Fresh start")
# Initialize map visit tracker for loop detection
map_visit_tracker = MapVisitTracker()
last_map_name = None # Track previous map for transition detection
log.info("🗺️ Map visit tracker initialized (loop detection enabled)")
# Initialize screenshot history tracker for file lifecycle management (cleanup)
screenshot_history = ScreenshotHistoryTracker(
snapshot_dir="snapshots", max_history=5
)
log.info("📸 Screenshot history tracker initialized (keeping last 5 snapshots)")
# Initialize coordinate tracker for loop detection and target tracking
coord_tracker = CoordinateTracker(
storage_path="data/coordinate_history.json",
max_history=10,
reset_on_start=not is_continuing_run,
)
log.info(
f"📍 Coordinate tracker initialized (history: {coord_tracker.get_context_summary()[:100] if coord_tracker.history else 'empty'})"
)
# Initialize navigation controller for goal-based navigation
navigation_controller = NavigationController(
coord_tracker=coord_tracker,
memory_manager=memory_manager,
map_visit_tracker=map_visit_tracker,
storage_path="data/navigation_state.json",
reset_on_start=not is_continuing_run,
)
log.info(
f"🧭 Navigation controller initialized ({len(navigation_controller.goal_stack)} goals)"
)
# Initialize target tracker for explicit pathfinding destinations
target_tracker = TargetTracker()
log.info("🎯 Target tracker initialized for explicit pathfinding")
# Initialize scenario manager for scripted overrides and goals
scenario_manager = ScenarioManager(navigation_controller)
scenario_manager.set_achievement_tracker(achievement_tracker)
log.info("Scenario manager initialized (with achievement tracking)")
# ═══════════════════════════════════════════════════════════════════════════
# PREPLANNED NAMES - Generate rival name at run start for unique each run
# ═══════════════════════════════════════════════════════════════════════════
name_planner = get_name_planner()
starter_planner = get_starter_planner()
if not is_continuing_run:
# Fresh run - generate new rival name via LLM
log.info("Generating preplanned rival name for new run...")
try:
rival_name = generate_rival_name_sync(client, MODEL, temperature=0.9)
name_planner.set_preplanned_rival_name(rival_name, source="llm")
log.info(f"Preplanned rival name: '{rival_name}' (LLM-generated)")
except Exception as e:
log.warning(f"Failed to generate rival name via LLM: {e}")
# Fallback to random from suggestions
import random
fallback_name = random.choice(RIVAL_NAME_SUGGESTIONS)
name_planner.set_preplanned_rival_name(fallback_name, source="fallback")
log.info(f"Using fallback rival name: '{fallback_name}'")
else:
# Continuing run - restore preplanned names from persistence
if run_state and run_state.preplanned_names:
rival_name = run_state.preplanned_names.get("rival_name")
if rival_name:
name_planner.set_preplanned_rival_name(rival_name, source="persisted")
log.info(f"Restored rival name from persistence: '{rival_name}'")
# Restore starter choice if available
if run_state and run_state.starter_choice:
species = run_state.starter_choice.get("species")
nickname = run_state.starter_choice.get("nickname")
if species and nickname:
starter_planner.set_choice(species, nickname)
# Restore state flags
starter_planner._starter_obtained = run_state.starter_choice.get(
"obtained", False
)
starter_planner._nickname_entered = run_state.starter_choice.get(
"nicknamed", False
)
log.info(
f"Restored starter choice from persistence: {species} '{nickname}'"
)
# Ensure a name exists even if not persisted
if not name_planner.rival_name:
import random
fallback_name = random.choice(RIVAL_NAME_SUGGESTIONS)
name_planner.set_preplanned_rival_name(fallback_name, source="continued")
log.info(f"Continuing run with fallback rival name: '{fallback_name}'")
# Capture the main event loop for thread-safe callbacks
# This MUST be done before defining callbacks that will run in executor threads
loop = asyncio.get_running_loop()
# Define thread-safe status callback for vision updates (called from executor threads)
# Uses run_coroutine_threadsafe because LLMController.stream_action runs in ThreadPoolExecutor
def status_callback(status: str):
state["processingStatus"] = status
# Guard against "Event loop is closed" during shutdown
if loop.is_running():
try:
asyncio.run_coroutine_threadsafe(
broadcast_func({"processingStatus": status}), loop
)
except RuntimeError:
pass # Loop closed during call - ignore
set_status_callback(status_callback)
log.info("📢 Processing status callback initialized (thread-safe)")
# Initialize LLM Controller - encapsulates all LLM interaction logic
# Re-read supports_vision from env since setup_llm_client doesn't return it yet
import os
llm_supports_vision_str = os.getenv("LLM_SUPPORTS_VISION", "true").lower()
llm_supports_vision = llm_supports_vision_str in ("true", "1", "yes", "on")
llm_config = {
"mode": CURRENT_MODE,
"model": MODEL,
"temperature": TEMPERATURE,
"max_tokens": MAX_TOKENS,
"is_local": IS_LOCAL,
"reasoning_enabled": REASONING_ENABLED,
"reasoning_effort": REASONING_EFFORT,
"uses_default_temp": USES_DEFAULT_TEMPERATURE,
"uses_max_completion_tokens": USES_MAX_COMPLETION_TOKENS,
"minimap_enabled": MINIMAP_ENABLED,
"minimap_2d": MINIMAP_2D,
"cleanup_window": CLEANUP_WINDOW,
"system_prompt_unsupported": SYSTEM_PROMPT_UNSUPPORTED,
"supports_reasoning": supports_reasoning,
"llm_supports_vision": llm_supports_vision,
}
controller = LLMController(client, vision_manager, memory_manager, llm_config)
controller.set_status_callback(status_callback)
# Vision result callback - broadcasts vision analysis immediately when complete
# Bundles processingStatus with vision_log for atomic UI update
def vision_callback(
vision_text: str, screenshot_b64: str = None, new_status: str = ""
):
log.info(
f"📸 Broadcasting vision result immediately ({len(vision_text)} chars)"
)
log_id = state.get("log_id_counter", 0)
vision_log = {
"id": log_id,
"text": vision_text,
"is_vision": True,
"timestamp": int(time.time() * 1000),
"screenshot_base64": screenshot_b64,
}
# Bundle vision_log and processingStatus in ONE broadcast for atomic UI update
# This ensures the screenshot is revealed exactly when vision analysis completes
broadcast_payload = {"vision_log": vision_log}
if new_status:
state["processingStatus"] = new_status
broadcast_payload["processingStatus"] = new_status
# Guard against "Event loop is closed" during shutdown
if loop.is_running():
try:
asyncio.run_coroutine_threadsafe(
broadcast_func(broadcast_payload), loop
)
except RuntimeError:
pass # Loop closed during call - ignore
controller.set_vision_callback(vision_callback)
log.info("🤖 LLM Controller initialized")
# Initialize Twitch chat service (optional - gracefully disabled if not configured)
twitch_service = create_twitch_service()
if twitch_service.is_available:
try:
await twitch_service.start()
log.info("📺 Twitch chat service started")
except Exception as e:
log.warning(f"Failed to start Twitch chat service: {e}")
else:
log.info(
"📺 Twitch chat service not configured (set TWITCH_* env vars to enable)"
)
# Initialize Pump.fun chat service (optional - for crypto livestream integration)
pumpfun_service = create_pumpfun_service()
if pumpfun_service.is_available:
try:
await pumpfun_service.start()
log.info("💎 Pump.fun chat service started")
except Exception as e:
log.warning(f"Failed to start Pump.fun chat service: {e}")
else:
log.info(
"💎 Pump.fun chat not configured (set PUMPFUN_TOKEN_ADDRESS to enable)"
)
# Initialize Zora chat service (optional - for onchain livestream integration)
zora_chat_service = create_zora_chat_service()
if zora_chat_service.is_available:
try:
await zora_chat_service.start()
log.info("🟣 Zora chat service started")
except Exception as e:
log.warning(f"Failed to start Zora chat service: {e}")
else:
log.info("🟣 Zora chat not configured (set ZORA_USERNAME to enable)")
# Initialize Zora Poster service (for minting achievement coins)
zora_poster_service = get_zora_poster_service()
try:
await zora_poster_service.start()
log.info("🖼️ Zora poster service started")
except Exception as e:
log.warning(f"Failed to start Zora poster service: {e}")
# Initialize specialized trackers
screenshot_manager = get_screenshot_manager()
zora_achievement_tracker = get_zora_achievement_tracker()
# Pre-fetch Zora post counter if needed (poster service does this internally or we can do it here)
# The tracker handles it on init/load.
# Initialize ComfyUI TTS service (optional - gracefully disabled if not configured)
# Create callback to notify UI when TTS starts playing (for synchronized typewriter)
async def on_tts_playback_start(
text: str, duration_ms: int, metadata: Dict[str, Any] = None
):
"""Called when TTS audio is about to start playing. Broadcasts to UI for sync."""
log.info(
f"🔊 Broadcasting TTS playback start: {len(text)} chars, {duration_ms}ms, meta={bool(metadata)}"
)
try:
payload = {
"tts_commentary": {
"text": text,
"duration_ms": duration_ms,
"playing": True,
}
}
# Inject metadata if present (e.g. reply_to info)
if metadata:
payload["tts_commentary"].update(metadata)
await broadcast_func(payload)
except Exception as e:
log.error(f"Failed to broadcast TTS start: {e}")
tts_service = create_tts_service(on_playback_start=on_tts_playback_start)
if tts_service.is_available:
is_connected = await tts_service.check_connection()
if is_connected:
log.info(f"🔊 ComfyUI TTS service connected: {tts_service.base_url}")
else:
log.warning(
f"🔊 ComfyUI TTS not reachable at {tts_service.base_url} (will retry when needed)"
)
else:
log.info(
"🔊 ComfyUI TTS service not configured (set COMFYUI_URL in .env to enable)"
)
# Initialize Chat Response service (uses Featherless AI / Alkahest for chat responses)
chat_response_service = create_chat_response_service()
if chat_response_service.is_available:
log.info(f"💬 Chat response service configured: {chat_response_service.model}")
else:
log.info(
"💬 Chat response service not configured (set LLM_CHAT_PROVIDER and keys to enable)"
)
# Initialize Tweet Generator service (Discord approval + X posting during intro)
tweet_generator = create_tweet_generator(llm_client=client)
if tweet_generator.enabled:
log.info("🐦 Tweet generator enabled (Discord approval + X posting)")
# Set up photo callback for scripted photo moments
def on_photo_moment(achievement):
"""Callback when a scripted photo moment is triggered."""
async def trigger_photo_tweet():
try:
outcome = await tweet_generator.generate_and_post_achievement_tweet(
achievement=achievement,
run_state=run_state,
game_state=state,
screenshot_path=SAVED_SCREENSHOT_PATH, # Include game screenshot
)
if outcome.result == TweetGeneratorResult.POSTED:
achievement_tracker.mark_tweet_posted(
achievement.achievement_type,
outcome.tweet_url,
)
log.info(f"📸 Photo tweet posted: {outcome.tweet_url}")
except Exception as e:
log.error(f"Photo tweet generation failed: {e}")
asyncio.create_task(trigger_photo_tweet())
scenario_manager.set_photo_callback(on_photo_moment)
else:
log.info(
"🐦 Tweet generator disabled (set TWITTER_ENABLED and DISCORD_ENABLED to enable)"
)
# Helper function to generate a chat response via the dedicated chat LLM
async def generate_chat_response(
username: str, message: str, is_past: bool = False
) -> str:
"""Generate a response to a Twitch chat message using the chat response service."""
# Get chatter memory context if available
chatter_context = ""
if twitch_service.is_available and hasattr(twitch_service, "engagement"):
chatter_context = twitch_service.engagement.memory.get_context(username)
if chat_response_service.is_available:
return await chat_response_service.generate_response(
username, message, is_past, chatter_context
)
# Fallback to main LLM if chat service not available
try:
prompt = get_chat_response_prompt(username, message, is_past)
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": prompt}],
max_tokens=150,
temperature=0.9,
)
if response.choices and response.choices[0].message.content:
return response.choices[0].message.content.strip()
return ""
except Exception as e:
log.error(f"Error generating chat response: {e}")
return ""
# Position history for stuck detection
position_history = []
# Track last action for failure replay
last_action = None
last_position = None
b64_mm = None
# Persistence save interval (save every N cycles)
PERSIST_INTERVAL = 5
cycles_since_persist = 0
# Track previous cycle time for UI display
prev_cycle_time_s = 0.0
cycle_times_history = [] # List of recent cycle times for average calculation
# mGBA timeout - if no response in this time, restart the cycle
# INCREASED from 5s to 15s - screenshot capture can take 4+ seconds after restart
# and thread race conditions caused false timeouts when old threads kept running
MGBA_TIMEOUT = 35 # seconds - must exceed socket timeout (30s) to let prep_llm complete/fail naturally
benchInstructions = ""
if benchmark is not None:
benchInstructions = benchmark.instructions
logging.info(f"Added bench instructions: {benchInstructions}")
# Always use fresh system prompt (in case prompts.py was updated)
fresh_system_prompt = build_system_prompt("", benchInstructions)
if run_state and run_state.chat_history:
# Get the old system prompt from persisted history
old_system_prompt = ""
if run_state.chat_history and run_state.chat_history[0].get("role") == "system":
old_system_prompt = run_state.chat_history[0].get("content", "")
# Check if prompt format has changed by looking for key structural changes
# We check for the COMMENTARY section format which changed from section 8/9 to section 7
old_has_commentary_7 = (
"7. COMMENTARY" in old_system_prompt and "REQUIRED" in old_system_prompt
)
new_has_commentary_7 = (
"7. COMMENTARY" in fresh_system_prompt and "REQUIRED" in fresh_system_prompt
)
prompt_format_changed = old_has_commentary_7 != new_has_commentary_7
if prompt_format_changed:
# Prompt format changed - start fresh to avoid LLM following old patterns
log.info(
"🔄 PROMPT FORMAT CHANGED - Clearing chat history to adopt new format"
)
controller.chat_history = [
{"role": "system", "content": fresh_system_prompt}
]
else:
# Restore chat history but replace the system prompt with fresh one
controller.chat_history = restored_history
if (
controller.chat_history
and controller.chat_history[0].get("role") == "system"
):
controller.chat_history[0] = {
"role": "system",
"content": fresh_system_prompt,
}
log.info("🔄 Updated system prompt to latest version")
else:
controller.chat_history = [{"role": "system", "content": fresh_system_prompt}]
# Track consecutive mGBA failures across cycle retries
# This persists across loop iterations so we can detect when mGBA is completely dead
consecutive_mgba_failures = 0
# State tracking for Twitch Predictions
was_in_battle = False
# Reduced to 3 - mGBA Lua socket callbacks may occasionally timeout
# but if it fails 3 times in a row, it's likely frozen
MAX_CONSECUTIVE_FAILURES = 3 # Restart mGBA after 3 consecutive failures
RECONNECT_THRESHOLD = 3 # Skip socket reconnection - go straight to restart
# Use mutable containers for socket and process so restarts update all references
sock_ref = {"socket": sock}
proc_ref = {"proc": mgba_proc}
# Socket lock to prevent race condition where old ThreadPoolExecutor threads
# continue using the socket after asyncio timeout, corrupting data for new threads.
# ThreadPoolExecutor doesn't cancel running threads on timeout - they keep running!
socket_lock = threading.Lock()
def prep_llm_locked(sock):
"""Wrapper that acquires lock before accessing socket.
This prevents race conditions where an old prep_llm thread (that didn't
get cancelled by asyncio timeout) corrupts socket data for new calls.
"""
with socket_lock:
return prep_llm(sock)
# Import config for mGBA paths
import config
def restart_mgba(port=8888):
"""
Kill and restart the entire mGBA process for 24/7 autonomous operation.
This is called when socket reconnection fails - mGBA Lua script is probably frozen.
"""
nonlocal sock_ref, proc_ref
log.warning("🔄 RESTARTING MGBA PROCESS (Lua script may be frozen)...")
# 1. Close old socket
try:
sock_ref["socket"].close()
log.info("🔌 Old socket closed")
except Exception as e:
log.warning(f"Error closing old socket: {e}")
# 2. Kill old process
if proc_ref["proc"] and proc_ref["proc"].poll() is None:
try:
proc_ref["proc"].terminate()
proc_ref["proc"].wait(timeout=5)
log.info("💀 Old mGBA process terminated")
except subprocess.TimeoutExpired:
log.warning("mGBA didn't terminate gracefully, killing...")
proc_ref["proc"].kill()
proc_ref["proc"].wait()
except Exception as e:
log.error(f"Error terminating mGBA: {e}")
# 3. Wait a moment for cleanup
time.sleep(2)
# 4. Start new mGBA process
rom_path = get_rom_path()
# Construct path to slot 1 save state (e.g. roms/red.gb -> roms/red.ss1)
# Using CLI load avoids the socket LOADSTATE pause/freeze issue
import os
ss1_path = os.path.splitext(rom_path)[0] + ".ss1"
cmd = [config.MGBA_EXE, "--script", config.LUA_SCRIPT]
# If save state exists, load it via CLI
if os.path.exists(ss1_path):
log.info(f"Using CLI to load save state: {ss1_path}")
cmd.extend(["-t", ss1_path])
else:
log.warning(f"Save state not found at {ss1_path}, starting fresh")
cmd.append(rom_path)
log.info(f"Starting new mGBA: {' '.join(cmd)}")
try:
proc_ref["proc"] = subprocess.Popen(
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True
)
except Exception as e:
log.error(f"Failed to start mGBA: {e}")
return False
# 5. Wait for mGBA to initialize
log.info("⏳ Waiting for mGBA to initialize...")
time.sleep(4)
# Check if mGBA started successfully
if proc_ref["proc"].poll() is not None:
log.error(
f"mGBA exited immediately with code {proc_ref['proc'].returncode}"
)
return False
# 6. Connect to new socket
import socket as sock_module
for attempt in range(5):
try:
new_sock = sock_module.create_connection(("localhost", port), timeout=5)
new_sock.setblocking(True)
new_sock.settimeout(30.0)
sock_ref["socket"] = new_sock
log.info(f"✅ Connected to new mGBA socket (fd={new_sock.fileno()})")
break
except Exception as e:
log.warning(f"Socket connection attempt {attempt + 1}/5 failed: {e}")
time.sleep(1)
else:
log.error("Failed to connect to new mGBA socket after 5 attempts")
return False
# 7. Load save state (SKIPPED - handled via CLI to prevent freeze)
# try:
# response = send_command(sock_ref["socket"], "LOADSTATE 1")
# if response and "OK" in response:
# log.info("✅ Save state loaded successfully!")
# else:
# log.warning(f"Save state load response: {response}")
# except Exception as e:
# log.error(f"Failed to load save state: {e}")
# 8. Enable input display
try:
send_command(sock_ref["socket"], "INPUT_DISPLAY_ON")
except:
pass
log.info("🎮 MGBA RESTART COMPLETE - Resuming game loop!")
return True
def reconnect_socket(port=8888):
"""Attempt to reconnect to mGBA socket. Updates sock_ref in place."""
nonlocal sock_ref
log.warning("🔌 Attempting to reconnect to mGBA socket...")
import socket as sock_module
try:
# Close old socket cleanly
try:
sock_ref["socket"].close()
except:
pass
# Try to create a new connection
time.sleep(1) # Give mGBA time to clean up
new_sock = sock_module.create_connection(("localhost", port), timeout=5)
new_sock.setblocking(True)
new_sock.settimeout(
30.0
) # Set default timeout to prevent indefinite blocking
sock_ref["socket"] = new_sock # Update the shared reference
log.info(
f"✅ Successfully reconnected to mGBA socket! (new fd={new_sock.fileno()})"
)
return True
except Exception as e:
log.error(f"❌ Failed to reconnect to mGBA: {e}")
return False
def check_socket_health():
"""Quick check if socket is still valid before operations."""
try:
fd = sock_ref["socket"].fileno()
if fd < 0:
log.warning(f"🔌 Socket fd is invalid ({fd})")
return False
return True
except Exception as e:
log.warning(f"🔌 Socket health check failed: {e}")
return False
# ═══════════════════════════════════════════════════════════════════════════
# STARTUP COUNTDOWN & INTRO CYCLE
# Wait for countdown duration (synced with frontend Starting Screen)
# Generate dynamic intro via LLM, preload TTS, play after chatbox appears
# ═══════════════════════════════════════════════════════════════════════════
# Import countdown duration from config
import config as cfg
countdown_seconds = cfg.STREAM_COUNTDOWN_SECONDS
# Get memory context for dynamic intro generation
memory_context = ""
if is_first_cycle_after_continue:
# Get previous session context from memory manager
try:
# Get recent narrative memories for context
recent_memories = memory_manager.get_context_for_llm("")
if recent_memories:
memory_context = recent_memories[:500] # Limit for prompt size
except Exception as e:
log.warning(f"Failed to get memory context for intro: {e}")
# Generate dynamic intro text via LLM
intro_text = ""
try:
if is_first_cycle_after_continue:
intro_prompt = f"""You are Lass, a young Pokemon trainer streaming your Pokemon Red adventure.
Generate a SHORT (1-2 sentences) stream intro message for returning after connection issues.
Context from your last session:
{memory_context if memory_context else "You were playing Pokemon Red."}
Rules:
- Be energetic and excited to continue
- Reference something specific from memory if possible (location, goal, recent event)
- Keep it under 30 words
- Use casual streaming language like "Hey chat!", "We're back!", etc.
- NO hashtags, NO emotes, NO special formatting
Example: "Hey everyone! Lass is back! Sorry about that, let's see if we can finally make it through Viridian Forest this time!"
Your intro message:"""
else:
intro_prompt = """You are Lass, a young Pokemon trainer starting your Pokemon Red stream.
Generate a SHORT (1-2 sentences) stream intro message for starting a new adventure.
Rules:
- Be energetic and excited to start
- Keep it under 30 words
- Use casual streaming language like "Hey chat!", "Let's go!", etc.
- NO hashtags, NO emotes, NO special formatting
Example: "Hey chat! It's Lass! Welcome to my Pokemon journey! Let's catch 'em all and become the very best!"
Your intro message:"""
log.info("🎬 Generating dynamic intro text via LLM...")
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": intro_prompt}],
max_tokens=100,
temperature=0.9,
)
if response.choices and response.choices[0].message.content:
intro_text = response.choices[0].message.content.strip()
# Clean up any quotes
intro_text = intro_text.strip("\"'")
log.info(f"✅ Generated intro: {intro_text}")
except Exception as e:
log.warning(f"Failed to generate dynamic intro: {e}")
# Fallback if LLM generation failed
if not intro_text:
if is_first_cycle_after_continue:
intro_text = "Hey everyone! Lass is back! Had some connection issues but I'm ready to continue my Pokemon adventure!"
else:
intro_text = "Hey chat! It's Lass! Welcome to my Pokemon Red stream! Let's catch some Pokemon and become the very best!"
# Broadcast that we're in the 'starting' phase with countdown duration FIRST
# Frontend will use countdownSeconds to set its timer immediately
await broadcast_func(
{
"introPhase": "starting",
"countdownSeconds": countdown_seconds,
"processingStatus": "",
}
)
log.info(f"⏱️ Stream countdown started: {countdown_seconds}s")
# --- TWEET GENERATION DURING COUNTDOWN ---
# Start tweet generation as a background task - runs Discord approval during countdown
tweet_task = None
tweet_outcome = None
if tweet_generator.enabled:
log.info("🐦 Starting tweet generation (runs during countdown)...")
async def run_tweet_generation():
"""Run tweet generation and return outcome."""
try:
# Build minimal game state from current state
game_state = {}
if state:
game_state = {
"map_name": state.get("map_name", ""),
"badges": state.get("badges", []),
"party": state.get("party", []),
}
return await tweet_generator.generate_and_post_tweet(
is_continuing_run=is_continuing_run,
run_state=run_state,
game_state=game_state,
)
except Exception as e:
log.error(f"Tweet generation failed: {e}")
return None
# Start as background task - will complete during countdown
tweet_task = asyncio.create_task(run_tweet_generation())
# Preload TTS audio DURING countdown (synthesize but don't play yet)
# Run this in background while countdown is active
preloaded_audio_path = None
if tts_service and tts_service.is_available:
# Clear any orphaned TTS requests from previous session
await tts_service.clear_queue()
log.info(f"🎬 Preloading intro TTS during countdown...")
try:
# Synthesize but don't play - returns the audio file path
preloaded_audio_path = await tts_service.synthesize_only(intro_text)
if preloaded_audio_path:
log.info(f"✅ Intro TTS preloaded: {preloaded_audio_path}")
else:
log.warning("⚠️ TTS preload returned None, will synthesize on-demand")
except Exception as e:
log.warning(f"⚠️ TTS preload failed, will synthesize on-demand: {e}")
# --- PRE-LOAD FIRST CYCLE DURING COUNTDOWN ---
# While the intro TTS and countdown are happening, we run the first cycle analysis
# so it's ready to display immediately when we transition to the game screen
log.info("🎮 Pre-loading first cycle analysis during countdown...")
first_cycle_vision = None
first_cycle_analysis = None
first_cycle_tts_path = None
try:
# Take screenshot of paused game (mGBA is paused but still provides screenshots)
await asyncio.sleep(0.5) # Let mGBA settle
loop = asyncio.get_event_loop()
current_socket = sock_ref["socket"]
try:
first_cycle_state = await asyncio.wait_for(
loop.run_in_executor(None, prep_llm_locked, current_socket), timeout=20
)
except asyncio.TimeoutError:
log.warning("⚠️ First cycle prep_llm timed out - will run on game start")
first_cycle_state = None
if first_cycle_state:
log.info("📸 Got first cycle screenshot from paused game")
# Broadcast that we're analyzing vision (even though we're in intro)
await broadcast_func({"processingStatus": "ANALYZING VISION..."})
# Run vision analysis on the paused screenshot
try:
first_cycle_vision = None
if vision_manager:
vision_result, _ = await vision_manager.analyze_image(
SAVED_SCREENSHOT_PATH
)
first_cycle_vision = vision_result
if first_cycle_vision:
log.info(
f"👁️ First cycle vision ready: {first_cycle_vision[:100]}..."
)
except Exception as ve:
log.warning(f"⚠️ First cycle vision failed: {ve}")
# Run LLM analysis for first cycle (skipped - would need controller)
if first_cycle_vision:
# Note: Full LLM analysis requires controller.call_with_timeout()
# which is not available during intro. Skipping preload.
log.info(
"📝 First cycle vision ready, LLM analysis will run on game start"
)
await broadcast_func({"processingStatus": ""})
except Exception as e:
log.warning(f"⚠️ First cycle preload failed: {e}")
await broadcast_func({"processingStatus": ""})
# Wait for remaining countdown duration (TTS preload + analysis took time)
elapsed_for_preload = 20 # Approximate seconds for TTS + first cycle preload
remaining_countdown = max(0, countdown_seconds - elapsed_for_preload)
if remaining_countdown > 0:
# Log progress for longer countdowns
for elapsed in range(0, remaining_countdown, 30):
remaining = remaining_countdown - elapsed
log.info(f"⏱️ Countdown: {remaining}s remaining...")
await asyncio.sleep(min(30, remaining))
# Countdown complete! Trigger color transition
log.info("🎨 Countdown complete! Starting color transition...")
await broadcast_func({"introPhase": "transitioning"})
# --- CHECK TWEET GENERATION RESULT ---
# If tweet task is still running, give it a few more seconds to complete
if tweet_task and not tweet_task.done():
log.info("🐦 Waiting for tweet generation to complete...")
try:
tweet_outcome = await asyncio.wait_for(tweet_task, timeout=30)
except asyncio.TimeoutError:
log.warning("🐦 Tweet generation timed out during transition")
tweet_task.cancel()
elif tweet_task:
# Task already completed
try:
tweet_outcome = tweet_task.result()
except Exception as e:
log.warning(f"🐦 Tweet task error: {e}")
# Log tweet outcome
if tweet_outcome:
if tweet_outcome.result == TweetGeneratorResult.POSTED:
log.info(f"🐦 Tweet posted successfully: {tweet_outcome.tweet_url}")
elif tweet_outcome.result == TweetGeneratorResult.MAX_REGENERATIONS:
log.info("🐦 Tweet skipped: max regenerations reached")
elif tweet_outcome.result == TweetGeneratorResult.ERROR:
log.warning(f"🐦 Tweet error: {tweet_outcome.error}")
# Wait for color transition animation (~3 seconds on frontend)
await asyncio.sleep(3)
# Transition to Just Chatting phase
# Character slides in over 6s, chatbox appears at 4.8s
log.info("🎥 Entering Just Chatting phase...")
await broadcast_func(
{
"introPhase": "just-chatting",
"introText": intro_text, # Send the dynamic intro text for the chatbox
"processingStatus": "STARTING STREAM...",
}
)
# Wait for character animation and chatbox to appear (5s)
log.info("⏳ Waiting for character entrance animation...")
await asyncio.sleep(5)
# Broadcast intro text for chatbox typewriter (just-chatting phase)
await broadcast_func(
{
"response_log": {
"id": 0,
"text": intro_text,
"is_response": True,
"timestamp": int(time.time() * 1000),
}
}
)
# Play the preloaded intro TTS
if tts_service and tts_service.is_available:
try:
if preloaded_audio_path:
await tts_service.play_preloaded(preloaded_audio_path, intro_text)
else:
await tts_service.synthesize_and_play(
intro_text, priority=tts_service.PRIORITY_COMMENTARY, wait=True
)
log.info("✅ Intro TTS complete")
except Exception as e:
log.warning(f"Intro TTS playback failed: {e}")
# Clear processing status
await broadcast_func({"processingStatus": ""})
# --- TRANSITION TO GAME ---
# 1. Clear intro text and show first cycle analysis instead
# 2. Continue game (no restart needed - game was running with audio during intro)
# 3. Start the session with preloaded first cycle analysis
log.info("🎮 Transitioning to game screen - first cycle analysis ready!")
# Broadcast session start with first cycle analysis (clearing intro text)
session_start_ms = int(time.time() * 1000)
game_transition_payload = {
"introPhase": "game",
"sessionStartTime": session_start_ms,
"cyclesEnabled": True,
}
# If we have first cycle analysis, send it to replace intro text
if first_cycle_analysis:
commentary = first_cycle_analysis.get("commentary", "")
if commentary:
# This will replace the intro text in the UI
game_transition_payload["response_log"] = {
"id": 1,
"text": commentary,
"is_response": True,
"timestamp": session_start_ms,
}
# Also update vision analysis if available
if first_cycle_vision:
game_transition_payload["vision_analysis"] = first_cycle_vision
# Update LLM analysis
llm_text = first_cycle_analysis.get(
"reasoning", ""
) or first_cycle_analysis.get("thoughts", "")
if llm_text:
game_transition_payload["llm_analysis"] = llm_text
await broadcast_func(game_transition_payload)
log.info(f"📊 Session started at {session_start_ms}")
# Play first cycle TTS if available
if first_cycle_tts_path and tts_service and tts_service.is_available:
try:
first_commentary = (
first_cycle_analysis.get("commentary", "")
if first_cycle_analysis
else ""
)
await tts_service.play_preloaded(first_cycle_tts_path, first_commentary)
log.info("✅ First cycle TTS played")
except Exception as e:
log.warning(f"First cycle TTS playback failed: {e}")
# Small delay to let frontend complete wipe transition
await asyncio.sleep(2)
while action_count < max_loops:
loop_start_time = time.time()
# Skip cycle increment on first iteration when continuing a run
# The restored cycle was interrupted and should be completed first
if is_first_cycle_after_continue:
is_first_cycle_after_continue = False
log.info(f"🔄 Resuming interrupted cycle {cycle_count}")
else:
cycle_count += 1
current_cycle = cycle_count
log.info(f"--- Loop Cycle {current_cycle} ---")
# CRITICAL: Always sync cycle/action count to run_state for graceful shutdown
# This ensures Ctrl+C saves the correct state even between PERSIST_INTERVAL saves
if run_state:
run_state.cycle_count = cycle_count
run_state.action_count = action_count
# Broadcast cycle count immediately
update_payload = {"cycle": current_cycle}
await broadcast_func({"cycle": current_cycle}) # Actually broadcast immediately
action_payload = {}
# Performance metrics container
cycle_metrics = {
"mGBA": 0.0,
"vision": 0.0,
"diff": 0.0,
"llm": 0.0,
"cycle": 0.0,
}
# Track true cycle start time (persists across restarts within same cycle)
true_cycle_start = time.time()
# Initialize result to prevent NameError if loop breaks early
result = None
try:
log.info("Requesting game state from mGBA...")
t_mgba_start = time.time()
# Check socket health before attempting prep_llm
if not check_socket_health():
log.warning(
"🔌 Socket unhealthy before prep_llm, attempting reconnection..."
)
if reconnect_socket():
log.info("✅ Socket reconnected before prep_llm")
else:
consecutive_mgba_failures += 1
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(
f"🔄 mGBA unresponsive after {MAX_CONSECUTIVE_FAILURES} failures - attempting full restart..."
)
if restart_mgba():
consecutive_mgba_failures = (
0 # Reset counter after successful restart
)
log.info("✅ mGBA restarted! Continuing 24/7 operation.")
else:
log.error(
"❌ mGBA restart failed! Waiting 30s before retry..."
)
await asyncio.sleep(30)
cycle_count -= 1
await asyncio.sleep(2)
continue
# Wrap prep_llm in async timeout to prevent indefinite blocking
# Use sock_ref["socket"] to ensure we use the current socket (may be reconnected)
# Use prep_llm_locked to prevent race condition with old threads
loop = asyncio.get_event_loop()
current_socket = sock_ref["socket"] # Capture current socket for executor
try:
current_mGBA_state = await asyncio.wait_for(
loop.run_in_executor(None, prep_llm_locked, current_socket),
timeout=MGBA_TIMEOUT,
)
except asyncio.TimeoutError:
mgba_duration = time.time() - t_mgba_start
consecutive_mgba_failures += 1
log.error(
f"⏱️ mGBA TIMEOUT after {mgba_duration:.1f}s (limit: {MGBA_TIMEOUT}s) - Retrying same cycle ({consecutive_mgba_failures}/{MAX_CONSECUTIVE_FAILURES})"
)
cycle_metrics["mGBA"] = (
mgba_duration * 1000
) # Store in ms for consistency
# Try to reconnect socket after RECONNECT_THRESHOLD failures
if consecutive_mgba_failures == RECONNECT_THRESHOLD:
log.warning(
f"🔌 Socket may be dead after {RECONNECT_THRESHOLD} failures. Attempting reconnection..."
)
if reconnect_socket():
log.info("✅ Socket reconnected! Continuing cycle retries...")
else:
log.error("❌ Socket reconnection failed. Will keep trying...")
# Check if mGBA is completely dead - trigger restart
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(
f"🔄 mGBA unresponsive after {MAX_CONSECUTIVE_FAILURES} failures - attempting full restart..."
)
if restart_mgba():
consecutive_mgba_failures = (
0 # Reset counter after successful restart
)
log.info("✅ mGBA restarted! Continuing 24/7 operation.")
else:
log.error("❌ mGBA restart failed! Waiting 30s before retry...")
await asyncio.sleep(30)
# Decrement cycle_count to retry the same cycle number (it was incremented at loop start)
cycle_count -= 1
await asyncio.sleep(2) # Brief pause before retry
continue
mgba_duration = time.time() - t_mgba_start
cycle_metrics["mGBA"] = mgba_duration * 1000 # Store in ms
log.info(f"⏱️ mGBA Response: {mgba_duration:.2f}s")
# Reset failure counter on success
consecutive_mgba_failures = 0
if benchmark is not None:
# check if we complted the bench
if benchmark.validation(current_mGBA_state):
break
# print(str(current_mGBA_state))
if not current_mGBA_state:
consecutive_mgba_failures += 1
log.error(
f"Failed to get state from mGBA (prep_llm returned None). Retrying same cycle ({consecutive_mgba_failures}/{MAX_CONSECUTIVE_FAILURES})."
)
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(f"🔄 mGBA unresponsive - attempting full restart...")
if restart_mgba():
consecutive_mgba_failures = 0
log.info("✅ mGBA restarted! Continuing 24/7 operation.")
else:
await asyncio.sleep(30)
cycle_count -= 1 # Retry same cycle number
await asyncio.sleep(max(0, interval - (time.time() - loop_start_time)))
continue
log.info("Received game state from mGBA.")
# --- ATOMIC SNAPSHOT LOGIC ---
t_snapshot_start = time.time()
# Create a unique snapshot for this cycle to prevent race conditions (Vision vs UI sync)
os.makedirs("snapshots", exist_ok=True)
snapshot_path = f"snapshots/cycle_{current_cycle}.png"
# Ensure we are copying the fresh capture
try:
shutil.copyfile(SAVED_SCREENSHOT_PATH, snapshot_path)
SCREENSHOT_PATH = snapshot_path
log.info(f"🔒 Atomic snapshot locked: {snapshot_path}")
# Register with history tracker - this keeps N and N-1, auto-cleans N-2+
screenshot_history.add_screenshot(current_cycle, snapshot_path)
# Capture for Zora gallery (keeps last 5 for progress posts)
if screenshot_manager:
screenshot_manager.capture(
source_path=snapshot_path,
cycle_number=current_cycle,
action_number=action_count,
map_name=current_mGBA_state.get("map_name", ""),
)
except Exception as e:
log.error(
f"Failed to create atomic snapshot: {e}. Falling back to global path."
)
SCREENSHOT_PATH = SAVED_SCREENSHOT_PATH
cycle_metrics["snapshot"] = (time.time() - t_snapshot_start) * 1000
except socket.timeout:
consecutive_mgba_failures += 1
log.error(
f"Socket timeout getting state from mGBA ({consecutive_mgba_failures}/{MAX_CONSECUTIVE_FAILURES}). Retrying..."
)
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(f"🔄 mGBA unresponsive - attempting full restart...")
if restart_mgba():
consecutive_mgba_failures = 0
else:
await asyncio.sleep(30)
cycle_count -= 1 # Retry same cycle
await asyncio.sleep(2) # Brief pause before retry
continue
except socket.error as se:
consecutive_mgba_failures += 1
log.error(
f"Socket error getting state from mGBA: {se} ({consecutive_mgba_failures}/{MAX_CONSECUTIVE_FAILURES}). Retrying..."
)
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(f"🔄 mGBA unresponsive - attempting full restart...")
if restart_mgba():
consecutive_mgba_failures = 0
else:
await asyncio.sleep(30)
cycle_count -= 1 # Retry same cycle
await asyncio.sleep(2) # Brief pause before retry
continue
except Exception as e:
consecutive_mgba_failures += 1
log.error(
f"Error getting state from mGBA: {e} - Retrying same cycle ({consecutive_mgba_failures}/{MAX_CONSECUTIVE_FAILURES})",
exc_info=True,
)
if consecutive_mgba_failures >= MAX_CONSECUTIVE_FAILURES:
log.warning(f"🔄 mGBA unresponsive - attempting full restart...")
if restart_mgba():
consecutive_mgba_failures = 0
else:
await asyncio.sleep(30)
cycle_count -= 1 # Retry same cycle number
await asyncio.sleep(max(0, interval - (time.time() - loop_start_time)))
continue
t_deepcopy_start = time.time()
llm_input_state = copy.deepcopy(current_mGBA_state)
cycle_metrics["deepcopy"] = (time.time() - t_deepcopy_start) * 1000
# Initialize minimap_analysis to None (will be set later if minimap data exists)
minimap_analysis = None
# REMOVE raw minimap_2d - LLM should use pre-computed minimap_analysis instead
# This prevents LLM from doing its own buggy parsing
# if "minimap_2d" in llm_input_state:
# del llm_input_state["minimap_2d"]
#
t_context_start = time.time()
cycle_metrics["context"] = 0
state_update_start = time.time()
# Track position for stuck detection
current_pos = current_mGBA_state.get("position")
current_map = current_mGBA_state.get("map_id")
stuck_info = {
"is_stuck": False,
"suggestion": "",
} # Default if no position data
if current_pos and current_map:
position_history.append((current_map, tuple(current_pos)))
# Keep only last 10 positions
position_history = position_history[-10:]
# Check if stuck and record failure
stuck_info = memory_manager.detect_stuck(position_history)
if stuck_info["is_stuck"]:
log.warning(f"🔄 STUCK DETECTED: {stuck_info['suggestion']}")
llm_input_state["stuck_warning"] = stuck_info["suggestion"]
# Record failure for replay context
if last_action and last_position:
goal_tracker.record_failure(
action=last_action,
position=last_position,
reason="Position unchanged - movement blocked",
)
# Use the stuck_position from detection (more accurate than current_pos for patterns)
map_name_for_decay = current_mGBA_state.get("map_name", "")
stuck_pos = stuck_info.get("stuck_position")
if stuck_pos and map_name_for_decay:
# stuck_pos is a tuple like (map_id, (x, y)), extract coordinates
if isinstance(stuck_pos, tuple) and len(stuck_pos) >= 2:
coords_to_decay = (
list(stuck_pos[1])
if isinstance(stuck_pos[1], tuple)
else list(stuck_pos)
)
else:
coords_to_decay = list(current_pos) if current_pos else None
if coords_to_decay:
# NEW: record_failed_exit_attempt now returns dict with status
failure_result = memory_manager.record_failed_exit_attempt(
map_name_for_decay, coords_to_decay
)
# Append the suggestion to stuck_warning
if failure_result.get("suggestion"):
llm_input_state["stuck_warning"] += (
f" {failure_result['suggestion']}"
)
# If there are untried directions, emphasize them
untried = failure_result.get("untried_directions", [])
if (
untried
and failure_result.get("status")
== "try_different_direction"
):
llm_input_state["stuck_warning"] += (
f" 🧭 TRY APPROACHING FROM: {untried[0]}"
)
# Add memory context to LLM input
map_name = current_mGBA_state.get("map_name", "")
memory_context = memory_manager.get_context_for_llm(map_name)
if memory_context:
llm_input_state["memory_context"] = memory_context
log.info(f"📝 Memory context: {memory_context[:100]}...")
# Add NPC avoidance context (only if map_name is available)
if map_name:
npc_context = memory_manager.get_npc_interaction_context(map_name)
if npc_context:
llm_input_state["npc_warning"] = npc_context
log.info(f"🚫 NPC context: {npc_context}")
# Add learned strategies context
# Build situation string from current state for relevance matching
situation_keywords = []
party = current_mGBA_state.get("party", [])
if party:
total_hp = sum(p.get("hp", 0) for p in party if isinstance(p, dict))
max_hp = sum(p.get("max_hp", 1) for p in party if isinstance(p, dict))
if max_hp > 0 and total_hp / max_hp < 0.3:
situation_keywords.append("low HP")
if all(p.get("hp", 0) == 0 for p in party if isinstance(p, dict)):
situation_keywords.append("fainted")
if stuck_info.get("is_stuck"):
situation_keywords.append("stuck")
situation_keywords.append("lost")
situation_str = " ".join(situation_keywords) if situation_keywords else map_name
strategy_context = memory_manager.get_strategy_context_for_llm(situation_str)
if strategy_context:
llm_input_state["strategy_hints"] = strategy_context
log.info(f"💡 Strategy context: {strategy_context[:80]}...")
# Track exploration and add context
map_id = current_mGBA_state.get("map_id", 0)
pos = current_mGBA_state.get("position", [0, 0])
minimap_2d = current_mGBA_state.get("minimap_2d", "")
if pos and len(pos) >= 2:
# Record this tile as visited
# NEW: Calculate total walkable tiles in the local minimap/window to allow % calculation
total_walkable = 0
if minimap_2d:
# Count walkable chars found in typical minimap string
# is_walkable(tile) usually includes: W, P, O, D, E, >, <, ^, v
# We simply count chars that are in this set
for char in minimap_2d:
if char in ["W", "P", "O", "D", "E", ">", "<", "^", "v"]:
total_walkable += 1
exploration_tracker.record_visit(
map_id, map_name, pos[0], pos[1], total_walkable=total_walkable
)
# Add exploration context to LLM input
exploration_context = exploration_tracker.get_context_for_llm(
map_id, map_name, pos[0], pos[1], minimap_2d
)
if exploration_context:
llm_input_state["exploration_context"] = exploration_context
log.info(f"🗺️ Exploration: {exploration_context.split(chr(10))[0]}")
# Track coordinates for loop detection and navigation progress
coord_tracker.add_position(current_cycle, pos[0], pos[1], map_name)
# Check for loops and add warning to LLM
loop_warning = coord_tracker.detect_loop()
if loop_warning:
llm_input_state["loop_warning"] = loop_warning
log.warning(f"⚠️ {loop_warning}")
# Add target progress context
target_progress = coord_tracker.get_progress_toward_target()
if target_progress.get("has_target"):
llm_input_state["navigation_target"] = target_progress
log.info(
f"🎯 Target: {target_progress.get('progress', 'unknown')} - {target_progress.get('recommendation', '')[:60]}"
)
# Check if target reached
if coord_tracker.check_target_reached():
log.info("✅ Navigation target reached!")
# Add full coordinate history context to help LLM understand movement patterns
coord_context = coord_tracker.get_context_summary()
if coord_context:
llm_input_state["coordinate_history"] = coord_context
log.info(
f"📍 Coord context: {len(coord_tracker.history)} positions tracked"
)
# ═══════════════════════════════════════════════════════════════════════════
# NAVIGATION CONTROLLER UPDATE - Goal-based navigation with computed paths
# ═══════════════════════════════════════════════════════════════════════════
# Pause context timing
cycle_metrics["context"] += (time.time() - t_context_start) * 1000
t_nav_start = time.time()
try:
nav_state = navigation_controller.update(
current_cycle=current_cycle,
current_pos=(pos[0], pos[1]),
current_map_name=map_name,
current_map_id=map_id,
position_history=position_history,
)
# Add navigation context to LLM input
nav_context = navigation_controller.get_context_for_llm(map_name)
if nav_context:
llm_input_state["navigation_goal"] = nav_context
log.info(
f"🧭 Nav: {nav_state.goal_description[:60]} (status: {nav_state.stuck_status})"
)
# Add computed path suggestion if available
if nav_state.next_moves:
llm_input_state["computed_path"] = (
f"FOLLOW THIS PATH: {nav_state.next_moves}"
)
log.info(f"🛤️ Path: {nav_state.next_moves}")
# Add navigation suggestion
if nav_state.suggestion:
llm_input_state["navigation_suggestion"] = nav_state.suggestion
except Exception as e:
log.warning(f"Navigation controller update failed: {e}")
cycle_metrics["nav"] = (time.time() - t_nav_start) * 1000
# Resume context timing
t_context_start = time.time()
# ═══════════════════════════════════════════════════════════════════════════
# DEV MARKERS CONTEXT - Special locations like starter Pokémon
# ═══════════════════════════════════════════════════════════════════════════
try:
from core.dev_markers import get_dev_marker_registry
dev_marker_registry = get_dev_marker_registry()
# Add dev marker context if on a map with special locations
dev_marker_context = dev_marker_registry.get_llm_context_for_map(
map_name=map_name, current_world_x=pos[0], current_world_y=pos[1]
)
if dev_marker_context:
llm_input_state["special_locations"] = dev_marker_context
# Special handling for starter Pokémon choice
starters = dev_marker_registry.get_starter_pokemon_choices(map_name)
party_size = len(current_mGBA_state.get("pokemon", []))
if starters and len(starters) > 0 and party_size == 0:
# We're in Oak's Lab with 0 Pokemon - starter selection time!
log.info(
f"STARTER POKEMON AVAILABLE: {len(starters)} choices, party size: {party_size}"
)
# Check if we've already made a choice
if not starter_planner.has_plan:
# Generate starter choice via LLM
log.info("Generating starter choice via LLM...")
try:
from core.name_generator import (
generate_starter_choice_sync,
)
choice = generate_starter_choice_sync(
client, MODEL, temperature=0.9
)
starter_planner.set_choice(
choice.species, choice.nickname
)
log.info(
f"Starter choice made: {choice.species} nicknamed '{choice.nickname}'"
)
# Set navigation goal to the starter
scenario_manager.set_starter_navigation_goal(
{
"map_name": map_name,
"map_id": map_id,
"world_x": pos[0],
"world_y": pos[1],
},
cycle_count,
)
except Exception as e:
log.error(f"Failed to generate starter choice: {e}")
# Fallback - let LLM choose manually
starter_names = [s.label for s in starters]
llm_input_state["starter_choice_available"] = (
f"CHOOSE YOUR STARTER POKEMON! Available: {', '.join(starter_names)}\n"
f"Navigate to the Pokeball and press A."
)
# Add context about the preplanned choice for AI commentary
if starter_planner.has_plan:
nav_context = starter_planner.get_nav_context(
pos[0], pos[1]
)
llm_input_state["starter_choice_context"] = (
f"STARTER POKEMON SELECTION (PREPLANNED)\n"
f"You chose: {starter_planner.species} nicknamed '{starter_planner.nickname}'!\n"
f"{nav_context}\n"
f"Express your excitement about this choice in your commentary!"
)
elif starters and len(starters) > 0 and party_size > 0:
# Already have a starter - mark as obtained if not already
if (
starter_planner.has_plan
and not starter_planner._starter_obtained
):
starter_planner.mark_starter_obtained()
log.info(f"Starter {starter_planner.species} obtained!")
except Exception as e:
log.warning(f"Dev markers context failed: {e}")
# ═══════════════════════════════════════════════════════════════════════════
# MAP OSCILLATION DETECTION - Prevent bouncing between same maps
# ═══════════════════════════════════════════════════════════════════════════
current_map_name = current_mGBA_state.get("map_name", "")
map_changed_this_cycle = False # Track for enhanced target prompt
if current_map_name and current_map_name != last_map_name:
# Map changed! Record the transition
map_changed_this_cycle = True
entry_pos = current_mGBA_state.get("position")
map_visit_tracker.record_map_entry(
map_name=current_map_name,
cycle=cycle_count,
position=tuple(entry_pos) if entry_pos else None,
from_map=last_map_name,
action=last_action,
)
last_map_name = current_map_name
# Prepare state for LLM
# This includes formatting the inventory, pokemon, etc.
# It updates the 'state' dictionary in-place
# --- TWITCH PREDICTIONS LOGIC ---
# Detect battle start/end
current_in_battle = state.get("battle_type", "None") != "None"
if twitch_service.is_available and hasattr(twitch_service, "engagement"):
# START BATTLE
if current_in_battle and not was_in_battle:
opponent = state.get("opponent_name", "Wild Pokemon")
is_gym = "Gym" in state.get("map_name", "") or "Leader" in opponent
# Only create predictions for significant battles or random chance
import random
if is_gym or random.random() < 0.3:
# Fire a prediction task without blocking loop
asyncio.create_task(
twitch_service.engagement.predictions.create_prediction(
title=f"Will Lass win against {opponent}?",
outcomes=["WIN", "LOSE/RUN"],
duration=60,
)
)
# END BATTLE
elif was_in_battle and not current_in_battle:
# Resolve prediction if one is active
# We assume "WIN" if we didn't black out?
# This is tricky without more state.
# For now, let's just resolve as WIN (index 0) if party is healthy?
# Or maybe we just close it.
# Actually, let's assume WIN for now to complete the flow.
# Ideally we check 'won_battle' flag if it existed.
# For now, if we are NOT in battle anymore and map didn't change to "Pokemon Center", maybe win?
asyncio.create_task(
twitch_service.engagement.predictions.resolve_prediction(
winning_outcome_index=0
)
)
was_in_battle = current_in_battle
# -------------------------------
# Check for oscillation patterns
oscillation = map_visit_tracker.detect_oscillation()
if oscillation:
llm_input_state["map_loop_warning"] = oscillation["warning"]
log.warning(f"🔄 {oscillation['warning']}")
# Add map visit frequency context
visit_frequency = map_visit_tracker.get_visit_frequency_context()
if visit_frequency:
llm_input_state["map_visit_frequency"] = visit_frequency
# Add pre-computed minimap analysis to reduce LLM hallucination
if minimap_2d:
# Pass world position for coordinate conversion
world_pos = current_mGBA_state.get("position", [])
minimap_analysis = parse_minimap(minimap_2d, world_position=world_pos)
if minimap_analysis and "error" not in minimap_analysis:
# Format as VERY CLEAR human-readable string for LLM
# Make blocked directions VERY obvious
blocked = minimap_analysis["blocked_directions"]
walkable = minimap_analysis["walkable_directions"]
exits = minimap_analysis["all_exit_tiles"]
adj = minimap_analysis["adjacent_tiles"]
passages = minimap_analysis.get("passages", [])
npcs = minimap_analysis.get("npc_tiles", [])
# Retrieve current exploration percentage for frontend display
exp_map = exploration_tracker.maps.get(map_id)
if exp_map:
# Set in state for persistence
state["explorationPct"] = exp_map.exploration_pct
# Add to update payload for frontend
update_payload["explorationPct"] = exp_map.exploration_pct
log.info(
f"🌍 Exploration: {exp_map.exploration_pct:.1f}% ({len(exp_map.visited_tiles)}/{exp_map.total_walkable})"
)
# Get target context for LLM (includes both meta-goal and tile target)
player_grid_x = minimap_analysis.get("player_col", 10)
player_grid_y = minimap_analysis.get("player_row", 10)
target_context = target_tracker.get_full_navigation_context(
player_grid_x, player_grid_y, map_name
)
analysis_str = (
f"⚠️ MINIMAP DATA (Viewport centered on you - usually 21x21, smaller at map edges!) ⚠️\n"
f"Grid: {minimap_analysis['grid_size']} | Player: {minimap_analysis['player_position']}\n"
f"═══════════════════════════════════════\n"
f"NORTH: {adj['north']} {'❌ BLOCKED!' if 'NORTH (U)' in blocked else '✓ walkable'}\n"
f"SOUTH: {adj['south']} {'❌ BLOCKED!' if 'SOUTH (D)' in blocked else '✓ walkable'}\n"
f"EAST: {adj['east']} {'❌ BLOCKED!' if 'EAST (R)' in blocked else '✓ walkable'}\n"
f"WEST: {adj['west']} {'❌ BLOCKED!' if 'WEST (L)' in blocked else '✓ walkable'}\n"
f"═══════════════════════════════════════\n"
f"🚫 BLOCKED MOVES: {blocked if blocked else 'NONE - all directions open'}\n"
f"✅ ALLOWED MOVES: {walkable}\n"
f"🚪 EXIT TILES: {exits if exits else 'None visible in current view'}\n"
f"🚶 NPCs: {npcs if npcs else 'None visible'}\n"
f"🌉 PASSAGES (paths through walls - might lead somewhere!): {chr(10).join(passages) if passages else 'None detected'}\n"
f"{target_context}"
)
llm_input_state["minimap_data"] = (
"See 'minimap_2d' field for ASCII grid layout."
)
log.info(
f"🗺️ Minimap: Player at {minimap_analysis['player_position']}, "
f"blocked: {blocked}, npcs: {len(npcs)}, passages: {len(passages)}"
)
# Store grid dimensions for lassMarkings overlay positioning
# Parse "21x19" format into separate width/height
try:
grid_dims = minimap_analysis["grid_size"].split("x")
state["minimapGridSize"] = {
"width": int(grid_dims[0]),
"height": int(grid_dims[1]),
}
update_payload["minimapGridSize"] = state["minimapGridSize"]
except (ValueError, IndexError):
pass # Keep existing or default
# === INVISIBLE OBSTACLE DETECTION ===
# If stuck but minimap shows walkable directions, there might be an invisible NPC/object
# BUT: Only suggest A-press if NOT at a map boundary (black edge)
if stuck_info.get("is_stuck") and walkable:
# Get the direction(s) we've been trying to move
# Check if any walkable direction is actually blocked (invisible obstacle)
stuck_warning_lower = llm_input_state.get(
"stuck_warning", ""
).lower()
# Check if player is facing a map boundary (black area)
# If so, don't suggest A-press - it's just the edge of the map
facing = current_mGBA_state.get("facing", "down")
is_at_map_boundary = False
# Check if the blocked direction matches a map edge
# Map boundaries are indicated by 'X' tiles in the minimap
if minimap_analysis and "blocked" in minimap_analysis:
blocked_directions = minimap_analysis.get("blocked", [])
facing_to_dir = {
"up": "N",
"down": "S",
"left": "W",
"right": "E",
}
if facing_to_dir.get(facing) in blocked_directions:
is_at_map_boundary = True
log.info(
f"📍 At map boundary facing {facing} - NOT an invisible obstacle"
)
# Suggest A-press to interact with potential NPC/object ONLY if not at map boundary
if (
"position unchanged" in stuck_warning_lower
or "movement blocked" in stuck_warning_lower
) and not is_at_map_boundary:
log.warning(
f"👻 INVISIBLE OBSTACLE: Minimap shows walkable tiles but movement blocked. Try pressing A to interact!"
)
llm_input_state["invisible_obstacle_hint"] = (
"👻 INVISIBLE OBSTACLE DETECTED: The minimap shows walkable tiles ahead, "
"but you can't move. There may be an NPC or object blocking you that isn't shown. "
"TRY PRESSING 'A' to interact with the invisible obstacle. If dialogue appears, "
"press B to close it and try moving again."
)
# Mark this location as potential NPC for Lass overlay
# The actual confirmation happens when vision detects dialogue
current_pos = current_mGBA_state.get("position", [])
if current_pos and len(current_pos) >= 2:
# Mark the tile in front based on facing direction
dx, dy = 0, 0
if facing == "up":
dy = -1
elif facing == "down":
dy = 1
elif facing == "left":
dx = -1
elif facing == "right":
dx = 1
potential_npc_coords = [
current_pos[0] + dx,
current_pos[1] + dy,
]
# Store as potential NPC (will be confirmed if dialogue detected)
memory_manager.add_lass_marking(
map_name, potential_npc_coords, "N", confidence=0.5
)
# ═══════════════════════════════════════════════════════════════════════════
# SCENARIO MANAGER - AUTO-GOALS & OVERRIDES (e.g. Pallet Town Loop)
# ═══════════════════════════════════════════════════════════════════════════
scenario_manager.check_for_override(llm_input_state, current_cycle)
# ═══════════════════════════════════════════════════════════════════════════
# AUTO-SET NAVIGATION GOAL FROM DETECTED EXITS
# ═══════════════════════════════════════════════════════════════════════════
# If no navigation goal is active and we can see exits, pick the nearest one
if exits and not navigation_controller.goal_stack:
try:
# Parse exit coordinates from format like "Grid[5,10] = World[12,8] (SOUTH EXIT)"
# Note: re module is imported at module level
nearest_exit = None
nearest_dist = float("inf")
current_pos = current_mGBA_state.get("position", [])
for exit_str in exits:
# Extract World coordinates from "World[x,y]" format
world_match = re.search(
r"World\[(\d+),\s*(\d+)\]", exit_str
)
if world_match:
exit_x, exit_y = (
int(world_match.group(1)),
int(world_match.group(2)),
)
if current_pos and len(current_pos) >= 2:
dist = abs(exit_x - current_pos[0]) + abs(
exit_y - current_pos[1]
)
if dist < nearest_dist:
nearest_dist = dist
# Extract direction hint (NORTH/SOUTH/EAST/WEST EXIT)
dir_match = re.search(
r"\((\w+)\s+EXIT\)", exit_str
)
direction = (
dir_match.group(1)
if dir_match
else "unknown"
)
nearest_exit = (exit_x, exit_y, direction)
if (
nearest_exit and nearest_dist <= 15
): # Only set goal if exit is reasonably close
navigation_controller.set_exit_goal(
exit_coords=(nearest_exit[0], nearest_exit[1]),
map_name=map_name,
map_id=map_id,
destination=f"{nearest_exit[2]} exit",
current_cycle=current_cycle,
)
log.info(
f"🎯 Auto-set navigation goal to nearest exit: {nearest_exit}"
)
except Exception as e:
log.debug(f"Exit goal auto-set failed: {e}")
# Add battle context using game memory (more accurate than vision)
try:
inventory = current_mGBA_state.get("inventory", [])
battle_context = get_battle_context(sock_ref["socket"], inventory=inventory)
if battle_context:
llm_input_state["battle_context"] = battle_context
log.info(f"⚔️ Battle detected: {battle_context[:80]}...")
except Exception as e:
log.debug(f"Battle context error (not in battle): {e}")
new_team = current_mGBA_state.get("party")
prev_team = state.get("currentTeam", []) or []
prev_team_size = len(prev_team) if prev_team else 0
new_team_size = len(new_team) if new_team else 0
# 🎉 MILESTONE DETECTION: First Pokemon obtained
if prev_team_size == 0 and new_team_size > 0:
log.info("🎉 MILESTONE DETECTED: First Pokemon obtained!")
# Complete the starter goal
if goal_tracker.complete_goal_by_keyword("first Pokemon"):
log.info("✅ Goal auto-completed: Get first Pokemon")
# Record in memory as a milestone event
pokemon_name = new_team[0].get("name", "Unknown") if new_team else "Unknown"
memory_manager.add_gameplay_memory(
location=current_mGBA_state.get("map_name", "unknown"),
description=f"MILESTONE: Received starter Pokemon {pokemon_name} from Professor Oak",
event_type="milestone",
outcome="obtained_first_pokemon",
pokemon_involved=[pokemon_name],
)
log.info(f"📝 Memory recorded: Obtained starter {pokemon_name}")
# Record outcome for strategy learning
memory_manager.record_outcome(
"goal_complete",
{"goal": "first_pokemon", "pokemon": pokemon_name},
{"location": current_mGBA_state.get("map_name", "unknown")},
)
# 🩺 HEALTH MONITORING: Detect blackouts and heals for strategy learning
prev_party_hp = getattr(memory_manager, "_last_party_hp", None)
if new_team:
current_hp_total = sum(
p.get("hp", 0) for p in new_team if isinstance(p, dict)
)
current_hp_max = sum(
p.get("max_hp", 1) for p in new_team if isinstance(p, dict)
)
all_fainted = all(
p.get("hp", 0) == 0 for p in new_team if isinstance(p, dict)
)
# Detect blackout (all Pokemon fainted, then respawned with full health)
if prev_party_hp is not None:
was_all_fainted = prev_party_hp.get("all_fainted", False)
prev_hp_total = prev_party_hp.get("total", 0)
# Blackout detected: was fainted, now have HP
if was_all_fainted and current_hp_total > 0:
memory_manager.record_outcome(
"blacked_out",
{
"respawn_location": current_mGBA_state.get(
"map_name", "unknown"
),
"hp_restored": current_hp_total,
},
{"was_lost": stuck_info.get("is_stuck", False)},
)
log.info("💀 Blackout detected - recording as potential strategy")
# Heal detected: significant HP increase without blackout
elif current_hp_total > prev_hp_total * 1.5 and not was_all_fainted:
memory_manager.record_outcome(
"healed",
{
"location": current_mGBA_state.get("map_name", "unknown"),
"hp_before": prev_hp_total,
"hp_after": current_hp_total,
},
{},
)
# Store for next cycle
memory_manager._last_party_hp = {
"total": current_hp_total,
"max": current_hp_max,
"all_fainted": all_fainted,
}
# Build team context for LLM awareness (every cycle)
team_pokemon_names = []
if new_team and new_team_size > 0:
for mon in new_team:
name = mon.get("name", "Unknown")
level = mon.get("level", "?")
team_pokemon_names.append(name)
# Create detailed team summary for LLM
team_details = []
for mon in new_team:
name = mon.get("name", "Unknown")
level = mon.get("level", "?")
hp = mon.get("current_hp", "?")
max_hp = mon.get("max_hp", "?")
team_details.append(f"{name} Lv{level} ({hp}/{max_hp}HP)")
llm_input_state["pokemon_team"] = (
f"YOUR TEAM ({new_team_size}/6): " + ", ".join(team_details)
)
log.info(f"🎮 Team context added: {new_team_size} Pokemon")
# Update goal context with team awareness
goal_context = goal_tracker.get_context_for_llm(
team_size=new_team_size, team_pokemon=team_pokemon_names
)
if goal_context:
llm_input_state["goal_context"] = goal_context
# Log first line only to avoid spam
log.info(f"🎯 Goals: {goal_context.split(chr(10))[0]}...")
# Standard team state update
if new_team is not None and json.dumps(new_team) != json.dumps(
state.get("currentTeam")
):
state["currentTeam"] = new_team
update_payload["currentTeam"] = state["currentTeam"]
log.info("State Update: currentTeam")
badge_data = current_mGBA_state.get("badges")
current_state_badges = state.get("badges")
# Compare the new list with the stored list
if badge_data != current_state_badges:
log.info(
f"State Update: Badges changed from {current_state_badges} to {badge_data}"
)
state["badges"] = badge_data
update_payload["badges"] = badge_data
# ═══════════════════════════════════════════════════════════════════════════
# ACHIEVEMENT DETECTION & TWEET TRIGGERING
# Check for new achievements and trigger tweet generation in background
# ═══════════════════════════════════════════════════════════════════════════
# Check Zora Minor Achievements
if zora_achievement_tracker:
try:
zora_ach = zora_achievement_tracker.check_for_achievements(
current_party=new_team or [],
current_map=map_name,
pokedex_caught=current_mGBA_state.get("pokedex_owned", 0),
game_state=current_mGBA_state,
)
if zora_ach:
log.info(f"🟣 Zora Minor Achievement: {zora_ach.get_title()}")
except Exception as e:
log.warning(f"Zora achievement detection error: {e}")
if tweet_generator.enabled:
try:
# Check for new achievements based on current game state
new_achievement = achievement_tracker.check_for_achievements(
current_party=new_team or [],
current_badges=badge_data or [],
current_map=map_name,
game_state=current_mGBA_state,
)
if new_achievement:
log.info(
f"🏆 ACHIEVEMENT UNLOCKED: {new_achievement.achievement_type.value}"
)
# Also trigger for Zora major achievement
if zora_achievement_tracker:
zora_achievement_tracker.trigger_from_major_achievement(
new_achievement.achievement_type.value,
context=new_achievement.context,
)
# Trigger tweet generation as background task
async def trigger_achievement_tweet(achievement):
"""Background task to generate and post achievement tweet."""
try:
outcome = await tweet_generator.generate_and_post_achievement_tweet(
achievement=achievement,
run_state=run_state,
game_state=current_mGBA_state,
screenshot_path=SAVED_SCREENSHOT_PATH, # Include game screenshot
)
if outcome.result == TweetGeneratorResult.POSTED:
achievement_tracker.mark_tweet_posted(
achievement.achievement_type,
outcome.tweet_url,
)
log.info(
f"🐦 Achievement tweet posted: {outcome.tweet_url}"
)
else:
log.info(
f"🐦 Achievement tweet result: {outcome.result.value}"
)
except Exception as e:
log.error(f"Achievement tweet generation failed: {e}")
# Start as background task (non-blocking)
asyncio.create_task(trigger_achievement_tweet(new_achievement))
except Exception as e:
log.warning(f"Achievement detection error: {e}")
pos = current_mGBA_state.get("position")
map_id = current_mGBA_state.get("map_id", "N/A")
# Ensure map_name is never None (handles missing key AND explicit None value)
map_name = current_mGBA_state.get("map_name") or ""
loc_str = "Unknown"
if pos:
loc_str = (
f"{map_name} (Map {map_id}) ({pos[0]}, {pos[1]})"
if map_name
else f"Map {map_id} ({pos[0]}, {pos[1]})"
)
if loc_str != state.get("minimapLocation"):
state["minimapLocation"] = loc_str
update_payload["minimapLocation"] = state["minimapLocation"]
log.info(f"State Update: minimapLocation -> {loc_str}")
# Always update minimap timestamp to trigger UI refresh (minimap image changes each cycle)
minimap_ts = int(time.time() * 1000)
state["minimapTimestamp"] = minimap_ts
update_payload["minimapTimestamp"] = minimap_ts
# Broadcast exploration percentage for current map
if map_id in exploration_tracker.maps:
exp_pct = exploration_tracker.maps[map_id].exploration_pct
state["explorationPct"] = round(exp_pct, 1)
update_payload["explorationPct"] = state["explorationPct"]
# Broadcast Lass markings for current map (N=NPC, O=Opening)
lass_marks = memory_manager.get_lass_markings_for_map(map_name)
if lass_marks:
lass_marks = list(lass_marks) # Make mutable copy
# CRITICAL FIX: Transform absolute world coordinates to relative grid coordinates
# The frontend displays a cropped minimap, so absolute coords will be off-screen
# We calculate the offset based on player position (world vs relative grid)
if minimap_analysis and current_pos:
player_world_x, player_world_y = current_pos
# Default to center (10,10) if analysis missing
player_grid_x = minimap_analysis.get("player_col", 10)
player_grid_y = minimap_analysis.get("player_row", 10)
# Offset = World - Grid
# e.g. World(50) - Grid(10) = Offset(40)
# Then Marker(52) - Offset(40) = Relative(12)
offset_x = player_world_x - player_grid_x
offset_y = player_world_y - player_grid_y
# Apply offset to all memory markers
valid_marks = []
grid_w = state.get("minimapGridSize", {}).get("width", 21)
grid_h = state.get("minimapGridSize", {}).get("height", 21)
for mark in lass_marks:
rel_x = mark["x"] - offset_x
rel_y = mark["y"] - offset_y
# Only keep markers that are visible on the current minimap crop
if 0 <= rel_x < grid_w and 0 <= rel_y < grid_h:
mark["x"] = rel_x
mark["y"] = rel_y
valid_marks.append(mark)
lass_marks = valid_marks
else:
lass_marks = []
# === TARGET TRACKING ===
# Check meta-goal progress first
if target_tracker.has_meta_goal:
meta_reached = target_tracker.check_meta_goal_reached(map_name)
if meta_reached:
llm_input_state["meta_goal_reached"] = (
f"🎯 META-GOAL REACHED! You arrived at {map_name}! Set a new meta-goal for your next destination."
)
else:
# Increment meta-goal cycle counter
if target_tracker.meta_goal:
target_tracker.meta_goal.cycles_active += 1
# Log progress periodically
if target_tracker.meta_goal.cycles_active % 10 == 0:
log.info(
f"🎯 Meta-goal progress: {target_tracker.meta_goal.cycles_active} cycles toward {target_tracker.meta_goal.destination_map}"
)
# Check if current tile target is reached
# Note: check_reached() also clears target when map changes (returns False in that case)
target_was_active = target_tracker.has_target
if target_tracker.has_target and current_pos:
player_world_x, player_world_y = current_pos
target_reached = target_tracker.check_reached(
player_world_x, player_world_y, map_id
)
if target_reached:
log.info("🎯 TILE TARGET REACHED! Notifying LLM to set a new target.")
if target_tracker.has_meta_goal:
llm_input_state["target_reached"] = (
f"🎯 TILE TARGET REACHED! Set a new tile target toward meta-goal: {target_tracker.meta_goal.destination_map}"
)
else:
llm_input_state["target_reached"] = (
"🎯 TILE TARGET REACHED! Set a new tile target. Also consider setting a meta-goal for your destination map."
)
# Enhanced prompt for map change (target auto-cleared)
# Target was active before check_reached(), but now cleared due to map change
if (
map_changed_this_cycle
and target_was_active
and not target_tracker.has_target
):
log.info(
"MAP TRANSITION: Target was cleared, providing enhanced prompt with exits"
)
# Build enhanced map transition prompt
prompt_parts = []
prompt_parts.append("\nCRITICAL: MAP TRANSITION DETECTED")
prompt_parts.append(
f"Map changed from {last_map_name or 'Unknown'} to {current_map_name}."
)
prompt_parts.append(
"Your tile target was automatically cleared (targets are map-specific)."
)
if target_tracker.has_meta_goal:
prompt_parts.append(
f"Meta-goal remains active: {target_tracker.meta_goal.destination_map}."
)
prompt_parts.append("\nIMMEDIATE ACTION REQUIRED:")
prompt_parts.append(
"Set a new tile target in this response using: <target_destination>x,y</target_destination>"
)
# Add available exits if minimap_analysis exists
if minimap_analysis:
adjacent_exits = minimap_analysis.get("adjacent_exits", [])
all_exits = minimap_analysis.get("all_exit_tiles", [])
if adjacent_exits:
prompt_parts.append(
"\nAdjacent exits (recommend setting target to one):"
)
for exit_str in adjacent_exits:
prompt_parts.append(f" - {exit_str}")
if all_exits:
prompt_parts.append("\nAll visible exits:")
# Limit to 5 exits to prevent prompt bloat
for exit_tile in all_exits[:5]:
prompt_parts.append(f" - {exit_tile}")
if len(all_exits) > 5:
prompt_parts.append(
f" ({len(all_exits) - 5} additional exits not shown)"
)
llm_input_state["target_reached"] = "\n".join(prompt_parts)
# Log warning if no targets set
if not target_tracker.has_target and not target_tracker.has_meta_goal:
log.warning(
"🎯 NO TARGETS SET - LLM should set both meta-goal and tile target for navigation!"
)
elif not target_tracker.has_target:
log.info("🎯 No tile target - LLM should set one toward meta-goal")
# Update target grid position based on viewport shift
# NOTE: Target marker should always be added if it exists, even without minimap_analysis
if target_tracker.has_target:
if minimap_analysis and current_pos:
# Update grid position based on viewport (normal case with minimap)
player_world_x, player_world_y = current_pos
player_grid_x = minimap_analysis.get("player_col", 10)
player_grid_y = minimap_analysis.get("player_row", 10)
target_tracker.update_grid_position(
player_world_x, player_world_y, player_grid_x, player_grid_y
)
# Add target marker to lassMarkings for overlay display
# This should work even without minimap_analysis (fallback positioning)
target_marker = target_tracker.get_marker_for_overlay()
if target_marker:
lass_marks.append(target_marker)
log.debug(
f"🎯 Added target marker to overlay at grid ({target_marker['x']}, {target_marker['y']})"
)
# Increment cycle counter for tile target
target_tracker.increment_cycle()
# === DEV MARKERS (STATIC SPECIAL LOCATIONS) ===
# Add developer-defined markers (e.g., starter Pokémon in Oak's lab)
from core.dev_markers import get_dev_marker_registry
dev_marker_registry = get_dev_marker_registry()
if minimap_analysis and current_pos:
player_world_x, player_world_y = current_pos
player_grid_x = minimap_analysis.get("player_col", 10)
player_grid_y = minimap_analysis.get("player_row", 10)
grid_w = state.get("minimapGridSize", {}).get("width", 21)
grid_h = state.get("minimapGridSize", {}).get("height", 21)
dev_markers = dev_marker_registry.get_overlay_markers_for_map(
map_name,
player_world_x,
player_world_y,
player_grid_x,
player_grid_y,
grid_w,
grid_h,
)
# Log dev marker detection for debugging
log.info(
f"[DEV_MARKERS] Map: {map_name}, Found {len(dev_markers)} dev markers"
)
if dev_markers:
log.debug(f"[DEV_MARKERS] Markers: {dev_markers}")
lass_marks.extend(dev_markers)
# Broadcast markings (with target if present)
if lass_marks:
state["lassMarkings"] = lass_marks
update_payload["lassMarkings"] = lass_marks
# --- DYNAMIC AVATAR STATE UPDATES ---
# Update Battle State (camelCase for frontend)
battle_state = current_mGBA_state.get("battle_state")
if battle_state:
state["inBattle"] = battle_state.get("in_battle", False)
state["battleType"] = battle_state.get("battle_type", None)
update_payload["inBattle"] = state["inBattle"]
update_payload["battleType"] = state["battleType"]
if state["inBattle"] and state["battleType"]:
# Only log change to avoid spam
if state.get("_last_battle_type") != state["battleType"]:
log.info(f"⚔️ Battle State: {state['battleType']}")
state["_last_battle_type"] = state["battleType"]
# Update Text/Dialog State for Speaking Avatar
text_state = current_mGBA_state.get("text_state")
if text_state:
state["textState"] = text_state
update_payload["textState"] = state["textState"]
# Update Menu State (simple heuristic for now)
menu_state = current_mGBA_state.get("menu_state")
if menu_state:
# If menu item count > 0, we can assume a menu is active?
# Or use explicit flag if available. For now using item count > 0 logic as fallback
# But relying on vision might be safer for "inMenu".
# However, memory is faster. Let's send inMenu if item_count > 0 AND text is NOT printing (menus often overlay text)
# Actually, let's just pass raw menu state if needed, or set inMenu
is_menu = menu_state.get("menu_item_count", 0) > 0
state["inMenu"] = is_menu
update_payload["inMenu"] = is_menu
# Update Movement State for biking/surfing avatar switching
movement_state = current_mGBA_state.get("movement_state")
if movement_state:
state["movementState"] = movement_state
update_payload["movementState"] = movement_state
if movement_state.get("movement_mode") != "walking":
log.info(f"🚴 Movement mode: {movement_state.get('movement_mode')}")
# Update Name Entry State (when on character naming screen)
name_entry_state = current_mGBA_state.get("name_entry_state")
forced_auto_action = None
# CRITICAL: Skip name entry if dialog text is present (menu state can persist from earlier)
dialog_text = text_state.get("text", "") if text_state else ""
has_dialog = bool(dialog_text and len(dialog_text.strip()) > 0)
if name_entry_state and not has_dialog:
state["nameEntryState"] = name_entry_state
update_payload["nameEntryState"] = name_entry_state
# Get the name planner
name_planner = get_name_planner()
# Detect what type of name we're entering
dialog_text = text_state.get("text", "") if text_state else ""
readable_text = current_mGBA_state.get("readable_text", "")
# Prioritize detection from tile map if available (from state.py)
detected_type = name_entry_state.get("name_type")
if detected_type and detected_type != "unknown":
name_type = detected_type
name_planner.name_type = name_type # Sync planner
log.info(f"Name Entry: Trusting state-detected type '{name_type}'")
else:
name_type = name_planner.detect_name_type(dialog_text, readable_text)
# Check if this is PRESET MENU or KEYBOARD
is_preset_menu = name_entry_state.get("is_preset_menu", False)
is_keyboard = name_entry_state.get("is_keyboard", False)
if is_preset_menu:
# ═══════════════════════════════════════════════════════════════
# STAGE 1: PRESET MENU - Enter keyboard mode to type custom name
# ═══════════════════════════════════════════════════════════════
cursor_option = name_entry_state.get("cursor_option", "NEW NAME")
cursor_index = name_entry_state.get("cursor_index", 0)
# Always enter keyboard mode to type custom names
recommended_action = "A;"
forced_auto_action = recommended_action
# Determine which name we're typing
if name_type == "player":
target_name = "LASS"
name_explanation = "your name 'LASS'"
elif name_type == "rival":
# Use preplanned rival name
target_name = name_planner.rival_name or "BUTT"
name_explanation = (
f"rival's name '{target_name}'\n"
f" 📢 COMMENTARY INSTRUCTION: Explicitly tell the chat you plan to name him '{target_name}'!"
)
elif name_type == "pokemon":
# Check for starter nickname
if (
starter_planner.waiting_for_nickname
and starter_planner.nickname
):
target_name = starter_planner.nickname
name_explanation = (
f"your {starter_planner.species}'s nickname '{target_name}'"
)
else:
target_name = "SKIP"
name_explanation = "press START to skip nickname"
else:
target_name = "custom name"
name_explanation = "a custom name"
name_entry_context = (
f"🎮 NAME SELECTION MENU\n"
f"══════════════════════════════════════\n"
f"📍 CURSOR: '{cursor_option}' (Row {cursor_index})\n"
f"\n"
f"⭐ ENTERING KEYBOARD MODE TO TYPE '{target_name}'!\n"
f"\n"
f"▶️ ACTION: {recommended_action} (AUTO-EXECUTED)\n"
f" This selects 'NEW NAME' and enters keyboard mode.\n"
f"\n"
f"💡 Typing {name_explanation}\n"
f" The keyboard sequence is auto-executed!\n"
f"\n"
f"⚠️ TRUST THIS CONTEXT, NOT VISION - Vision is unreliable for name entry!\n"
)
log.info(
f"📝 Preset menu: cursor on '{cursor_option}', entering keyboard mode to type '{target_name}'"
)
elif is_keyboard:
# ═══════════════════════════════════════════════════════════════
# STAGE 2: KEYBOARD MODE - Use pre-computed sequences
# ═══════════════════════════════════════════════════════════════
# Activate keyboard tracker
kb_tracker = get_keyboard_tracker()
kb_tracker.activate()
# Get tracked position (more reliable than memory reads)
tracked_state = kb_tracker.get_state_dict()
row = tracked_state.get("row", 1) # 1-indexed for display
col = tracked_state.get("col", 1) # 1-indexed for display
selected_char = tracked_state.get("selected_char", "A")
# Build context based on name type
# NOTE: Name entry is AUTO-EXECUTED - LLM just needs context for commentary
if name_type == "player":
# Type "LASS" for player
target_name = "LASS"
if not name_planner.current_name:
name_planner.start_typing(target_name)
next_step = name_planner.get_current_step()
progress = name_planner.get_progress_string()
if next_step:
# Calculate target position for context (1-indexed for display)
target_row = next_step["to_pos"][0] + 1
target_col = next_step["to_pos"][1] + 1
step_num = name_planner.current_char_index + 1
total_steps = len(name_planner.typing_sequence)
name_entry_context = (
f"⌨️ KEYBOARD MODE - TYPING 'LASS'\n"
f"══════════════════════════════════════\n"
f"Auto-executing keyboard input for player name.\n"
f"\n"
f"📝 Progress: {progress} (Step {step_num}/{total_steps})\n"
f"\n"
f"🎯 Next character: '{next_step['char']}'\n"
f"▶️ Action: {next_step['path']} (AUTO-EXECUTED)\n"
f"\n"
f"📍 Current: '{selected_char}' (Row {row}, Col {col})\n"
f" Target: '{next_step['char']}' (Row {target_row}, Col {target_col})\n"
)
forced_auto_action = next_step["path"]
elif (
name_planner.is_done_typing()
and not name_planner.confirmation_sent
):
# All letters typed, need to confirm with START
forced_auto_action = "START;"
name_entry_context = (
f"✅ TYPING COMPLETE: '{target_name}'\n"
f"══════════════════════════════════════\n"
f"All letters typed successfully!\n"
f"\n"
f"▶️ ACTION: START; (AUTO-EXECUTED)\n"
f"\n"
f"🔴 CRITICAL - MEMORY_WRITE REQUIRED:\n"
f" After confirming, you MUST write in section 12:\n"
f' "Named myself {target_name}"\n'
f"\n"
f"Example MEMORY_WRITE entry:\n"
f'**12. MEMORY_WRITE**: "Named myself {target_name}"\n'
)
name_planner.set_confirmation_sent(True)
elif name_planner.confirmation_sent:
# Wait for state transition
name_entry_context = (
"✅ Name confirmed. Waiting for transition..."
)
forced_auto_action = None
elif name_type == "rival":
# Use preplanned rival name (set at run start)
if not name_planner.rival_name:
# Fallback if somehow not set
import random
fallback = random.choice(RIVAL_NAME_SUGGESTIONS)
name_planner.set_preplanned_rival_name(
fallback, source="fallback"
)
log.warning(f"Rival name not set, using fallback: {fallback}")
target_name = name_planner.rival_name
if (
not name_planner.current_name
or name_planner.current_name != target_name
):
name_planner.start_typing(target_name)
next_step = name_planner.get_current_step()
progress = name_planner.get_progress_string()
if next_step:
step_num = name_planner.current_char_index + 1
total_steps = len(name_planner.typing_sequence)
name_entry_context = (
f"⌨️ KEYBOARD MODE - TYPING RIVAL NAME '{target_name}'\n"
f"══════════════════════════════════════\n"
f"📢 COMMENTARY ACTION: Tell the chat you've decided to name him '{target_name}'!\n"
f" e.g. 'I know exactly what to call him... {target_name}!'\n"
f"\n"
f"Auto-executing keyboard input...\n"
f"\n"
f"📝 Progress: {progress} (Step {step_num}/{total_steps})\n"
f"\n"
f"🎯 Next character: '{next_step['char']}'\n"
f"▶️ Action: {next_step['path']} (AUTO-EXECUTED)\n"
f"\n"
f"📍 Current cursor: '{selected_char}' (Row {row}, Col {col})\n"
)
forced_auto_action = next_step["path"]
elif (
name_planner.is_done_typing()
and not name_planner.confirmation_sent
):
forced_auto_action = "START;"
name_entry_context = (
f"✅ TYPING COMPLETE: '{target_name}'\n"
f"══════════════════════════════════════\n"
f"All letters typed successfully!\n"
f"\n"
f"▶️ ACTION: START; (AUTO-EXECUTED)\n"
f"\n"
f"🔴 CRITICAL - MEMORY_WRITE REQUIRED:\n"
f" After confirming, you MUST write in section 12:\n"
f' "Named rival {target_name}"\n'
f"\n"
f"Example MEMORY_WRITE entry:\n"
f'**12. MEMORY_WRITE**: "Named rival {target_name}"\n'
)
name_planner.set_confirmation_sent(True)
elif name_planner.confirmation_sent:
name_entry_context = (
"✅ Name confirmed. Waiting for transition..."
)
forced_auto_action = None
elif name_type == "pokemon":
# Pokemon nickname entry
# Check if this is a starter nickname (preplanned)
if (
starter_planner.waiting_for_nickname
and starter_planner.nickname
):
target_name = starter_planner.nickname
if (
not name_planner.current_name
or name_planner.current_name != target_name
):
name_planner.set_current_starter_nickname(target_name)
name_planner.start_typing(target_name)
next_step = name_planner.get_current_step()
progress = name_planner.get_progress_string()
if next_step:
step_num = name_planner.current_char_index + 1
total_steps = len(name_planner.typing_sequence)
name_entry_context = (
f"KEYBOARD MODE - TYPING '{target_name}'\n"
f"==================================\n"
f"Auto-executing keyboard input for starter nickname.\n"
f"\n"
f"Progress: {progress} (Step {step_num}/{total_steps})\n"
f"\n"
f"Next character: '{next_step['char']}'\n"
f"Action: {next_step['path']} (AUTO-EXECUTED)\n"
)
forced_auto_action = next_step["path"]
elif (
name_planner.is_done_typing()
and not name_planner.confirmation_sent
):
forced_auto_action = "START;"
name_entry_context = (
f"TYPING COMPLETE: '{target_name}'\n"
f"==================================\n"
f"Confirming nickname for your {starter_planner.species}!\n"
f"\n"
f"ACTION: START; (AUTO-EXECUTED)\n"
)
name_planner.set_confirmation_sent(True)
starter_planner.mark_nickname_entered()
elif name_planner.confirmation_sent:
name_entry_context = (
"Nickname confirmed. Waiting for transition..."
)
forced_auto_action = None
else:
# No preplanned nickname - skip
name_entry_context = (
f"POKEMON NICKNAME (SKIPPING)\n"
f"==================================\n"
f"Cursor: '{selected_char}' (Row {row}, Col {col})\n"
f"\n"
f"Skipping nickname to continue faster.\n"
f"\n"
f"ACTION: START; (Skip nickname)\n"
)
forced_auto_action = "START;"
else:
# UNKNOWN NAME TYPE - Do NOT auto-execute
# This happens during confirmation screens or ambiguous states
name_entry_context = (
f"❓ UNKNOWN NAME ENTRY STATE\n"
f"══════════════════════════════════════\n"
f"Could not determine if Player, Rival, or Pokemon.\n"
f"Auto-execution DISABLED. Manual control active.\n"
)
forced_auto_action = None
log.info("Name Entry: Unknown type - yielding control to agent.")
else:
# Unknown state - shouldn't happen
name_entry_context = (
f"🎮 NAME ENTRY DETECTED\n"
f"══════════════════════════════════════\n"
f"Processing name entry state...\n"
)
llm_input_state["name_entry_context"] = name_entry_context
log.info(
f"✅ Added name_entry_context: type={name_type}, preset={is_preset_menu}, keyboard={is_keyboard}"
)
else:
# Deactivate keyboard tracker when not in name entry
kb_tracker = get_keyboard_tracker()
if kb_tracker.active:
kb_tracker.deactivate()
# Reset name planner for next name entry session
name_planner = get_name_planner()
name_planner.reset()
# Debug: Log menu state to trace why name entry wasn't detected
menu_state = current_mGBA_state.get("menu_state", {})
if menu_state:
log.info(
f"name_entry_state=None, menu_state: item_count={menu_state.get('menu_item_count')}, "
f"cursor=({menu_state.get('cursor_x')},{menu_state.get('cursor_y')}), "
f"selected={menu_state.get('selected_item')}"
)
# Stop context timing
cycle_metrics["context"] += (time.time() - t_context_start) * 1000
# Start Image Prep timing
t_img_start = time.time()
# Default: Analysis uses the clean snapshot unless combined
ANALYSIS_IMAGE_PATH = SCREENSHOT_PATH
UI_IMAGE_PATH = SCREENSHOT_PATH
if ONE_IMAGE_PER_PROMPT and MINIMAP_ENABLED:
try:
# Load images
# VALIDATION: Use SCREENSHOT_PATH (the atomic snapshot) instead of global SAVED_SCREENSHOT_PATH
ss_img = Image.open(SCREENSHOT_PATH)
mm_img = Image.open(SAVED_MINIMAP_PATH)
# Resize minimap to match screenshot height
mm_ratio = ss_img.height / mm_img.height
new_mm_width = int(mm_img.width * mm_ratio)
mm_img = mm_img.resize((new_mm_width, ss_img.height), Image.LANCZOS)
# Create a new canvas wide enough for both
combined_width = ss_img.width + mm_img.width
combined = Image.new("RGB", (combined_width, ss_img.height))
# Paste screenshot at (0,0), minimap at (ss.width, 0)
combined.paste(ss_img, (0, 0))
combined.paste(mm_img, (ss_img.width, 0))
# Save combined image and override SCREENSHOT_PATH
# VALIDATION: Derive combined path from atomic snapshot path to keep it unique per cycle
combined_path = (
os.path.splitext(SCREENSHOT_PATH)[0] + "_with_minimap.png"
)
combined.save(combined_path)
# CRITICAL: AI gets the combined image, but UI gets the clean original
ANALYSIS_IMAGE_PATH = combined_path
# UI_IMAGE_PATH remains SCREENSHOT_PATH (the clean snapshot)
log.info(f"Combined screenshot + minimap saved to {combined_path}")
except Exception as e:
log.error(f"Failed to combine minimap: {e}")
# Fallback: Analysis path remains SCREENSHOT_PATH
# Handle image processing based on provider
if CURRENT_MODE == "ZAI" and vision_manager:
# For Z.AI MCP, use the CLEAN screenshot (UI_IMAGE_PATH) per user request
# The minimap context is no longer sent to vision - user prefers clean image
llm_input_state["screenshot_path"] = UI_IMAGE_PATH
if not ONE_IMAGE_PER_PROMPT and MINIMAP_ENABLED:
llm_input_state["minimap_path"] = MINIMAP_PATH
# Also create base64 versions for fallback
# CRITICAL: Encode the UI_IMAGE_PATH (clean snapshot) for the frontend
b64_ss = encode_image_base64(UI_IMAGE_PATH)
if b64_ss:
# Note: We send the CLEAN screenshot to the LLM's vision input here as valid base64
# But the MCP tool will use 'screenshot_path' (ANALYSIS_IMAGE_PATH)
llm_input_state["screenshot"] = {
"image_url": {
"url": f"data:image/png;base64,{b64_ss}",
"detail": IMAGE_DETAIL,
}
}
# CRITICAL: Also set top-level screenshot_base64 for vision callback to use
# This ensures the UI gets the screenshot when vision analysis completes
llm_input_state["screenshot_base64"] = b64_ss
else:
llm_input_state["screenshot"] = None
# Note: Explicitly attaching the clean screenshot base64 for the UI later
# This happens in the log creation step using 'b64_ss'
if not ONE_IMAGE_PER_PROMPT and MINIMAP_ENABLED:
b64_mm = encode_image_base64(MINIMAP_PATH)
if b64_mm:
llm_input_state["minimap"] = {
"image_url": {
"url": f"data:image/png;base64,{b64_mm}",
"detail": IMAGE_DETAIL,
}
}
else:
# Standard base64 image processing for other providers
# Use ANALYSIS path for standard models if they can't handle separate tools?
# Actually, standard models viewing the image directly should probably see the combined one if enabled.
# But the user specifically asked for "render the cycle not the one with the minimap" on the UI.
# So we must differentiate what we send to LLM vs what we send to UI.
# For standard LLM input (e.g. GPT-4o), we want it to see the Minimap if enabled.
b64_llm = encode_image_base64(ANALYSIS_IMAGE_PATH)
# But for UI, we want clean.
b64_ss = encode_image_base64(UI_IMAGE_PATH)
# Store raw base64 for UI screenshot display (used by vision callback)
llm_input_state["screenshot_base64"] = b64_ss
# CRITICAL: Set screenshot_path for Vision Manager (ComfyUI)
# Use ANALYSIS_IMAGE_PATH (with minimap) if available, otherwise UI_IMAGE_PATH
# This ensures Vision Manager has a file path to process!
llm_input_state["screenshot_path"] = (
ANALYSIS_IMAGE_PATH
if os.path.exists(ANALYSIS_IMAGE_PATH)
else UI_IMAGE_PATH
)
if b64_llm:
llm_input_state["screenshot"] = {
"image_url": {
"url": f"data:image/png;base64,{b64_llm}",
"detail": IMAGE_DETAIL,
}
}
else:
llm_input_state["screenshot"] = None
if not ONE_IMAGE_PER_PROMPT and MINIMAP_ENABLED:
b64_mm = encode_image_base64(MINIMAP_PATH)
if b64_mm:
llm_input_state["minimap"] = {
"image_url": {
"url": f"data:image/png;base64,{b64_mm}",
"detail": IMAGE_DETAIL,
}
}
else:
llm_input_state["minimap"] = None
cycle_metrics["img_prep"] = (time.time() - t_img_start) * 1000
log.info(
f"Pre-LLM state update & image prep took {time.time() - state_update_start:.2f}s. SS:{bool(b64_ss)}, MM:{bool(b64_mm)}"
)
# NEW: Get Contextual Area Hint
area_hint = get_area_hint(llm_input_state)
if area_hint:
llm_input_state["area_hint"] = area_hint
log.info(
f"💡 AREA HINT: {area_hint.splitlines()[0] if area_hint else 'None'}"
)
log_id_counter = state.get("log_id_counter", 0) + 1
state["log_id_counter"] = log_id_counter
# Broadcast "ANALYZING..." status AND screenshot before starting vision+LLM processing
# This allows the UI to show the new screenshot with CRT animation during the vision phase
state["processingStatus"] = "ANALYZING VISION..."
await broadcast_func(
{
"processingStatus": "ANALYZING VISION...",
"vision_log": {
"id": log_id_counter,
"text": "Analyzing screenshot...",
"is_vision": True,
"timestamp": int(time.time() * 1000),
"screenshot_base64": b64_ss, # Send screenshot immediately so CRT animation plays now
},
}
)
# LOG ALL KEY STATE FIELDS BEING SENT TO LLM (for debugging)
key_fields = [
f"map_name={llm_input_state.get('map_name', 'None')}",
f"position={llm_input_state.get('position', 'None')}",
f"dialog_text={llm_input_state.get('dialog_text', 'None')[:30] if llm_input_state.get('dialog_text') else 'None'}...",
f"in_battle={llm_input_state.get('battle_state', {}).get('in_battle', False)}",
f"name_entry_context={'YES' if llm_input_state.get('name_entry_context') else 'NO'}",
f"text_state.is_printing={llm_input_state.get('text_state', {}).get('is_printing', False)}",
f"menu_state.item_count={llm_input_state.get('menu_state', {}).get('menu_item_count', 0)}",
]
log.info(f"📊 LLM INPUT STATE: {' | '.join(key_fields)}")
# ═══════════════════════════════════════════════════════════════════════════
# PARALLEL CHAT + LLM PROCESSING
# Run chat responses during LLM wait to keep stream active
# ═══════════════════════════════════════════════════════════════════════════
# ═══════════════════════════════════════════════════════════════════════════
# PARALLEL CHAT + LLM PROCESSING
# Run chat responses during LLM wait to keep stream active
# ═══════════════════════════════════════════════════════════════════════════
chat_stop_event = asyncio.Event()
parallel_chat_task = asyncio.create_task(
run_chat_background_task(
chat_stop_event,
tts_service,
twitch_service,
cycle_count,
pumpfun_service, # Pump.fun service for crypto chat
zora_chat_service, # Zora service for onchain chat
chat_response_service, # For Featherless LLM responses
)
)
try:
# Run the main LLM call via Controller
(
action,
game_analysis,
summary_json,
vision_analysis_for_ui,
) = await controller.call_with_timeout(
llm_input_state,
STREAM_TIMEOUT,
LLM_TOTAL_TIMEOUT,
benchmark,
cycle_metrics,
)
# ------------------------------------------------------------------
# INTRO PACING CONTROL (User Request)
# ------------------------------------------------------------------
# Prevent 'A' spam during intro to allow name entry detection to catch up
# and prevent accidental skipping of the name entry screen.
# Heuristic: Party 0 AND not in Players House 2F (0x26 = 38).
if action and not forced_auto_action:
_map_id = current_mGBA_state.get("map_id", 0)
_party_size = current_mGBA_state.get("party_size", 0)
if _party_size == 0 and _map_id != 38: # 38 is PLAYERS_HOUSE_2F
# Check if action is purely 'A' presses (e.g. "A;", "A;A;", "A;A;A;")
_commands = [cmd for cmd in action.split(";") if cmd.strip()]
if all(cmd == "A" for cmd in _commands) and len(_commands) > 1:
log.info(
f"Intro Pacing: Throttling '{action}' to single 'A;' to prevent skipping"
)
action = "A;"
# OVERRIDE ACTION IF AUTO-EXECUTION IS PENDING
if forced_auto_action:
log.info(f"🤖 Auto-executing name entry action: {forced_auto_action}")
action = forced_auto_action
tokens_used_session = controller.tokens_used_session
# ═══════════════════════════════════════════════════════════════
# FRESH STATE CHECK & SCENARIO OVERRIDE (Fix Race Conditions)
# ═══════════════════════════════════════════════════════════════
fresh_state = get_fast_game_state(sock)
override_action = scenario_manager.check_for_override(
fresh_state, cycle_count
)
if override_action:
log.info(
f"⚠️ SCENARIO OVERRIDE: {override_action} (based on fresh state)"
)
action = override_action
# Clear analysis to avoid confusion in UI
game_analysis = "Scenario Override (Auto-Executed)"
finally:
# Signal chat task to stop and cancel if still running
chat_stop_event.set()
# Cancel current chat TTS (so it doesn't talk over game commentary)
if tts_service and tts_service.is_available:
tts_service.cancel_pending_chat_responses()
t_chat_wait_start = time.time()
try:
await parallel_chat_task
except asyncio.CancelledError:
pass
except Exception as e:
log.error(f"Error awaiting parallel chat task: {e}")
cycle_metrics["chat_wait"] = (time.time() - t_chat_wait_start) * 1000
# Clear processing status after completion
state["processingStatus"] = ""
await broadcast_func({"processingStatus": ""})
if summary_json is not None:
tmp = {
"log_entry": {
"id": log_id_counter,
"text": "🔎 Chat history cleaned up.",
}
}
await broadcast_func(tmp)
required = ("primaryGoal", "secondaryGoal", "tertiaryGoal", "otherNotes")
if isinstance(summary_json, dict):
# summary_json is dict, safe to check for keys
missing = [k for k in required if k not in summary_json]
if not missing:
state["goals"] = {
"primary": summary_json["primaryGoal"],
"secondary": summary_json["secondaryGoal"],
"tertiary": summary_json["tertiaryGoal"],
}
state["otherGoals"] = summary_json["otherNotes"]
update_payload["goals"] = state["goals"]
update_payload["otherGoals"] = state["otherGoals"]
else:
logging.error(
f"Missing required goal keys in summary_json: {missing!r}"
)
else:
logging.error(
f"Expected summary_json to be dict, but got {type(summary_json).__name__!r}"
)
if vision_analysis_for_ui:
update_payload["vision_analysis"] = vision_analysis_for_ui
# Extract screen_type from vision analysis for UI and chat context
detected_screen_type = ""
try:
vision_json = json.loads(vision_analysis_for_ui)
detected_screen_type = vision_json.get("screen_type", "")
except json.JSONDecodeError:
# Fallback: regex extraction
match = re.search(
r'"screen_type"\s*:\s*"([^"]+)"', vision_analysis_for_ui
)
if match:
detected_screen_type = match.group(1)
if detected_screen_type:
update_payload["screen_type"] = detected_screen_type
state["screen_type"] = detected_screen_type
log.debug(f"📺 Detected screen type: {detected_screen_type}")
action_to_send = None
log_action_text = "No action taken (LLM failed)."
# ═══════════════════════════════════════════════════════════════════════════
# EARLY LOG CREATION & BROADCAST (before action execution for sync)
# This ensures UI shows reasoning/action BEFORE mGBA executes it
# ═══════════════════════════════════════════════════════════════════════════
# Calculate button counts EARLY (needed for action log)
buttons_in_action = 0
action_start = action_count
action_end = action_count
if action:
action_buttons = [
c for c in action.replace(";", "").replace(" ", "") if c in "UDLRABST"
]
buttons_in_action = len(action_buttons)
action_start = action_count + 1
action_end = action_count + buttons_in_action
log.info(
f"📊 Calculated action counts: #{action_start}-#{action_end} ({buttons_in_action} buttons)"
)
# ═══════════════════════════════════════════════════════════════════════════
# PARSE TARGET DESTINATION from LLM response
# Format: <target_destination>[x,y] reason: "description"</target_destination>
# ═══════════════════════════════════════════════════════════════════════════
if game_analysis:
# Parse target_destination tag
target_match = re.search(
r'<target_destination>\s*\[(\d+)\s*,\s*(\d+)\]\s*(?:reason:\s*["\']?(.+?)["\']?)?\s*</target_destination>',
game_analysis,
re.IGNORECASE | re.DOTALL,
)
if target_match:
try:
target_grid_x = int(target_match.group(1))
target_grid_y = int(target_match.group(2))
target_reason = (
target_match.group(3).strip()
if target_match.group(3)
else "Navigation target"
)
# Convert grid to world coordinates
if current_pos and minimap_analysis:
player_world_x, player_world_y = current_pos
player_grid_x = minimap_analysis.get("player_col", 10)
player_grid_y = minimap_analysis.get("player_row", 10)
target_world_x, target_world_y = grid_to_world(
target_grid_x,
target_grid_y,
player_grid_x,
player_grid_y,
player_world_x,
player_world_y,
)
target_tracker.set_target(
grid_x=target_grid_x,
grid_y=target_grid_y,
world_x=target_world_x,
world_y=target_world_y,
map_id=map_id,
map_name=map_name,
reason=target_reason,
)
# Sync to NavigationController to enable BFS pathfinding
navigation_controller.set_goal(
goal_type=GoalType.COORDINATE,
target_x=target_world_x,
target_y=target_world_y,
map_name=map_name,
map_id=map_id,
reason=target_reason,
current_cycle=current_cycle,
)
except (ValueError, TypeError) as e:
log.warning(f"🎯 Failed to parse target_destination: {e}")
# Also check for <clear_target/>
if re.search(r"<clear_target\s*/?\s*>", game_analysis, re.IGNORECASE):
target_tracker.clear_target("llm_cleared")
# Parse <meta_goal> tag for high-level navigation objectives
# Format: <meta_goal>MAP_NAME reason: "description"</meta_goal>
meta_match = re.search(
r'<meta_goal>\s*(\w+)\s*(?:reason:\s*["\']?(.+?)["\']?)?\s*</meta_goal>',
game_analysis,
re.IGNORECASE | re.DOTALL,
)
if meta_match:
try:
dest_map = meta_match.group(1).strip().upper()
meta_reason = (
meta_match.group(2).strip()
if meta_match.group(2)
else "Navigation goal"
)
target_tracker.set_meta_goal(dest_map, meta_reason)
except (ValueError, TypeError) as e:
log.warning(f"🎯 Failed to parse meta_goal: {e}")
# Check for <clear_meta_goal/>
if re.search(r"<clear_meta_goal\s*/?\s*>", game_analysis, re.IGNORECASE):
target_tracker.clear_meta_goal("llm_cleared")
# Create log entries BEFORE action execution
log_entries = []
cycle_timestamp = int(time.time() * 1000)
# Vision log
if vision_analysis_for_ui:
vision_log = {
"id": log_id_counter,
"text": vision_analysis_for_ui,
"is_vision": True,
"timestamp": cycle_timestamp,
"screenshot_base64": b64_ss,
}
log_entries.append(vision_log)
if b64_ss:
state["latest_screenshot_base64"] = b64_ss
# Response log (LLM reasoning)
if game_analysis and game_analysis.strip():
response_log = {
"id": log_id_counter,
"text": game_analysis.strip(),
"is_response": True,
"timestamp": cycle_timestamp,
}
log_entries.append(response_log)
# Action log (intended action - shows BEFORE execution)
if action and vision_analysis_for_ui:
action_log = {
"id": log_id_counter,
"text": f"Action: {action}",
"is_action": True,
"action_start": action_start,
"action_end": action_end,
"button_count": buttons_in_action,
"timestamp": cycle_timestamp,
}
log_entries.append(action_log)
# BROADCAST LOGS IMMEDIATELY (before action execution)
if "log_history" not in state:
state["log_history"] = []
for log_entry in log_entries:
state["log_history"].insert(0, log_entry)
if log_entry.get("is_vision"):
update_payload["vision_log"] = log_entry
log.info(
f"🖼️ Broadcasting vision log: {log_entry.get('text', '')[:50]}..."
)
elif log_entry.get("is_response"):
update_payload["response_log"] = log_entry
log.info(
f"💭 Broadcasting response log: {log_entry.get('text', '')[:50]}..."
)
elif log_entry.get("is_action"):
update_payload["log_entry"] = (
log_entry # For action_payload compatibility
)
log.info(
f"🎮 Broadcasting action log: {log_entry.get('text', '')[:50]}..."
)
state["log_history"] = state["log_history"][:50]
# Include state counts in the early broadcast (with PREDICTED action count)
predicted_action_count = action_count + buttons_in_action
update_payload["cycle"] = current_cycle
update_payload["actions"] = (
predicted_action_count # Show what it WILL be after execution
)
update_payload["tokensUsed"] = tokens_used_session
state["actions"] = predicted_action_count
state["tokensUsed"] = tokens_used_session
# BROADCAST NOW - before action execution
if update_payload:
await broadcast_func(update_payload)
log.info(f"📡 Broadcast complete: {list(update_payload.keys())}")
# ═══════════════════════════════════════════════════════════════════════════
# TTS COMMENTARY PLAYBACK (BEFORE action execution for proper sync)
# Order: UI shows reasoning/action → TTS plays → mGBA executes
# ═══════════════════════════════════════════════════════════════════════════
if game_analysis and tts_service and tts_service.is_available:
# Extract commentary from the LLM response
commentary_match = re.search(
r"<commentary>([\s\S]*?)</commentary>", game_analysis, re.IGNORECASE
)
if not commentary_match:
# Fallback: various numbered formats
commentary_match = re.search(
r'(?:7|8|9|10|11)\.\s*\*{0,2}COMMENTARY\*{0,2}[:\s]*["\'\(]?(.+?)["\'\)]?(?=\n\d+\.|$|\n\n|</game_analysis>)',
game_analysis,
re.IGNORECASE | re.DOTALL,
)
if commentary_match:
commentary_text = commentary_match.group(1).strip()
# Strip <think> tags that sometimes appear in commentary
commentary_text = re.sub(
r"<think>[\s\S]*?</think>",
"",
commentary_text,
flags=re.IGNORECASE | re.DOTALL,
).strip()
# Strip meta_goal and target_destination tags that leak into commentary
commentary_text = re.sub(
r"<meta_goal>[\s\S]*?</meta_goal>",
"",
commentary_text,
flags=re.IGNORECASE | re.DOTALL,
)
commentary_text = re.sub(
r"<target_destination>[\s\S]*?</target_destination>",
"",
commentary_text,
flags=re.IGNORECASE | re.DOTALL,
)
# Handle short tags <t>...</t> and malformed/shorthand tags
commentary_text = re.sub(
r"<t>.*?</t>", "", commentary_text, flags=re.DOTALL | re.IGNORECASE
)
# Remove standalone <t>, </t>, </ t>, </t > tags
commentary_text = re.sub(
r"</?t\s*>", "", commentary_text, flags=re.IGNORECASE
)
commentary_text = re.sub(r"^[-–•]\s*", "", commentary_text)
commentary_text = re.sub(
r"\n.*$", "", commentary_text
) # Take first line only
commentary_text = commentary_text.strip().strip("\"'")
if commentary_text and len(commentary_text) > 5:
log.info(f"🔊 Playing TTS: {commentary_text[:60]}...")
# Update chat response service context with FULL game state
# This ensures Lass can give factual answers about her team, location, etc.
if chat_response_service.is_available:
# Build rich team summary for chat context
team_summary = ""
if new_team and len(new_team) > 0:
team_parts = []
for mon in new_team:
name = mon.get("name", "Unknown")
level = mon.get("level", "?")
hp = mon.get("hp", mon.get("current_hp", "?"))
max_hp = mon.get("max_hp", "?")
team_parts.append(f"{name} Lv{level} ({hp}/{max_hp}HP)")
team_summary = ", ".join(team_parts)
else:
team_summary = "No Pokemon yet!"
# Build rich game context
game_status = (
f"In {current_mGBA_state.get('map_name', 'unknown')}"
)
if (
current_mGBA_state.get("battle_type")
and current_mGBA_state.get("battle_type") != "None"
):
game_status = f"In battle! ({current_mGBA_state.get('battle_type', 'wild')} battle)"
# Include recent actions/history
# Note: History tracking could be added later if needed
history_text = ""
# Fetch $LASS token info for chat context (uses DexScreener, cached 60s)
token_info_text = ""
t_token_start = time.time()
try:
token_service = get_token_service()
if token_service.is_available:
token_info = await token_service.get_token_info()
if token_info:
token_info_text = token_info.format_summary()
except Exception as ti_err:
log.debug(
f"Token info fetch failed (non-critical): {ti_err}"
)
cycle_metrics["token"] = (time.time() - t_token_start) * 1000
chat_response_service.update_context(
game_context=game_status,
commentary=commentary_text,
location=current_mGBA_state.get("map_name", "unknown"),
team=team_summary,
history=history_text,
memory=memory_manager.get_narrative_context(),
token_info=token_info_text,
screen_type=state.get("screen_type", ""),
)
# Synthesize and play TTS - WAIT for it to complete
t_tts_start = time.time()
try:
await tts_service.synthesize_and_play(
commentary_text,
priority=tts_service.PRIORITY_COMMENTARY,
wait=True,
)
log.info(f"✅ TTS playback complete")
except Exception as tts_err:
log.warning(f"🔊 TTS error: {tts_err}")
finally:
cycle_metrics["tts"] = (time.time() - t_tts_start) * 1000
# ═══════════════════════════════════════════════════════════════════════════
# ACTION EXECUTION (now happens AFTER TTS completes)
# ═══════════════════════════════════════════════════════════════════════════
t_action_start = time.time()
if action:
action_to_send = action
log_action_text = f"Action: {action}"
log.info(f"LLM proposed action: {action}")
try:
# Broadcast action_execute to trigger button animation in UI
# This is sent right before mGBA executes so animation syncs with game
await broadcast_func(
{
"action_execute": True,
"action_buttons": action_to_send.replace(";", " ")
.strip()
.split(),
"buttons_in_action": buttons_in_action,
}
)
log.info(f"🎮 Broadcasted action_execute for animation trigger")
# Wait 1s before sending action to let game state settle
time.sleep(1)
sock.sendall((action_to_send + "\n").encode("utf-8"))
log.info(f"Action '{action_to_send}' sent to mGBA.")
# Update keyboard tracker with the action (for name entry screens)
kb_tracker = get_keyboard_tracker()
if kb_tracker.active:
# Update tracker with full action (handles D/U/L/R navigation)
kb_tracker.apply_action(action_to_send)
log.info(
f"🎹 Keyboard tracker updated with action: {action_to_send}"
)
# Advance name planner step if action contains 'A' (typing a character)
name_planner = get_name_planner()
if "A" in action_to_send.upper() and name_planner.current_name:
# Only advance if we're actively typing (not just pressing A on menu)
if not name_planner.is_done_typing():
name_planner.advance()
log.info(
f"🎹 Name planner advanced: {name_planner.get_progress_string()}"
)
else:
# Done typing, pressing START will reset the planner
log.info(
f"🎹 Name planner complete, ready for START confirmation"
)
# Wait 6s AFTER sending action to let screen fully render before next screenshot
# This prevents cut-off/partial screenshots and ensures dialog text is captured
time.sleep(6)
log.info(
"Post-action delay complete (6s), ready for next cycle screenshot."
)
# Track action execution time (includes 1s pre-wait + send + 6s post-wait)
cycle_metrics["action"] = (time.time() - t_action_start) * 1000
# Track for failure replay
last_action = action_to_send
last_position = current_pos
# Track NPC interactions when A button is pressed
if "A" in action_to_send.upper() and current_pos and map_name:
npc_warning = memory_manager.record_npc_interaction(
map_name, list(current_pos), npc_name="NPC"
)
if npc_warning:
# Add warning to stuck_warning for next cycle
llm_input_state["stuck_warning"] = (
llm_input_state.get("stuck_warning", "") + f" {npc_warning}"
)
# VALIDATION: Check if name was just confirmed and MEMORY_WRITE was forgotten
name_planner = get_name_planner()
if "S" in action_to_send.upper() or "START" in action_to_send.upper():
# Check if we just completed a name entry
if name_planner.name_type and game_analysis:
expected_memory = None
if (
name_planner.name_type == "player"
and name_planner.player_name
):
expected_memory = f"Named myself {name_planner.player_name}"
elif (
name_planner.name_type == "rival"
and name_planner.rival_name
):
expected_memory = f"Named rival {name_planner.rival_name}"
if expected_memory:
# Check if MEMORY_WRITE section contains the expected entry
if "MEMORY_WRITE" in game_analysis:
# Found MEMORY_WRITE section, check if name is mentioned
if expected_memory not in game_analysis:
# MEMORY_WRITE exists but name not mentioned - WARNING!
warning_msg = (
f"⚠️ CRITICAL WARNING: You just confirmed your "
f"{'name' if name_planner.name_type == 'player' else 'rival name'} "
f"but didn't write it to MEMORY_WRITE! "
f"You MUST include: '{expected_memory}' in section 12 or you'll forget the name!"
)
llm_input_state["stuck_warning"] = (
llm_input_state.get("stuck_warning", "")
+ f" {warning_msg}"
)
log.warning(f"🔴 {warning_msg}")
else:
# No MEMORY_WRITE section at all - CRITICAL ERROR!
warning_msg = (
f"🔴 CRITICAL ERROR: You just confirmed your "
f"{'name' if name_planner.name_type == 'player' else 'rival name'} "
f"but completely forgot section 12 (MEMORY_WRITE)! "
f"You MUST write: '{expected_memory}' in section 12 immediately or you'll forget the name forever!"
)
llm_input_state["stuck_warning"] = (
llm_input_state.get("stuck_warning", "")
+ f" {warning_msg}"
)
log.error(f"🔴 {warning_msg}")
except socket.error as se:
log.error(
f"Socket error sending action '{action_to_send}': {se}. Stopping loop."
)
break
except Exception as e:
log.error(
f"Unexpected error sending action '{action_to_send}': {e}",
exc_info=True,
)
else:
log.error("No valid action from LLM. Cannot send command.")
# User request: Stay on same cycle number if failure occurs
cycle_count -= 1
log.info(
"Decremented cycle count to retry same cycle number on next attempt."
)
# Update the GLOBAL action_count variable after execution
# (state was already updated in the early broadcast with predicted value)
if action:
action_count += buttons_in_action
log.info(
f"📊 Actions #{action_start}-#{action_end} ({buttons_in_action} buttons) - EXECUTED"
)
# Update elapsed time for status display
elapsed = datetime.datetime.now() - start_time
elapsed_seconds = elapsed.total_seconds()
game_status_str = f"{int(elapsed_seconds // 3600)}h {int((elapsed_seconds % 3600) // 60)}m {int(elapsed_seconds % 60)}s"
state["gameStatus"] = game_status_str
state["modelName"] = MODEL
# Force memory recording for important location transitions
# IMPORTANT: This runs ALWAYS, not just when analysis_text is empty
try:
# MAP TRANSITION LOGIC - use current_mGBA_state which has correct keys
current_map = current_mGBA_state.get("map_name", "unknown")
current_map_id = current_mGBA_state.get("map_id", -1)
current_pos = current_mGBA_state.get("position", [])
# We need to track the PREVIOUS map/pos to record a link
# use attributes on memory_manager to persist across loops efficiently
last_map = getattr(memory_manager, "last_map", None)
last_map_id = getattr(memory_manager, "last_map_id", None)
last_pos = getattr(memory_manager, "last_pos", None)
# Check if map changed (and neither is unknown)
if (
current_map != "unknown"
and last_map != "unknown"
and current_map_id != -1
and last_map_id != -1
and (current_map != last_map or current_map_id != last_map_id)
):
# We moved between maps! exact position we left FROM is last_pos
# exact position we arrived AT is current_pos
if last_pos and current_pos:
# Check if we were on an O tile before transition
# This helps determine if this was a natural transition or cutscene
last_minimap = getattr(memory_manager, "last_minimap_2d", "")
was_on_o_tile = False
minimap_had_exit = False
if last_minimap:
# Check if there was an 'O' tile near player position in last minimap
# 'O' tiles indicate exits/entrances
rows = last_minimap.split(";")
for row in rows:
if "O" in row:
minimap_had_exit = True
break
# Check if player marker 'P' was adjacent to or on an 'O'
for row_idx, row in enumerate(rows):
p_idx = row.find("P")
if p_idx >= 0:
# Check adjacent tiles for 'O'
if (p_idx > 0 and row[p_idx - 1] == "O") or (
p_idx < len(row) - 1 and row[p_idx + 1] == "O"
):
was_on_o_tile = True
if (
row_idx > 0
and len(rows[row_idx - 1]) > p_idx
and rows[row_idx - 1][p_idx] == "O"
):
was_on_o_tile = True
if (
row_idx < len(rows) - 1
and len(rows[row_idx + 1]) > p_idx
and rows[row_idx + 1][p_idx] == "O"
):
was_on_o_tile = True
break
new_links = memory_manager.record_transition(
from_map=last_map,
from_pos=list(last_pos)
if isinstance(last_pos, tuple)
else last_pos,
to_map=current_map,
to_pos=list(current_pos)
if isinstance(current_pos, tuple)
else current_pos,
was_on_o_tile=was_on_o_tile,
minimap_had_exit=minimap_had_exit,
)
if new_links:
log.info(
f"🔗 TRANSITION RECORDED: {last_map} -> {current_map} (O-tile: {was_on_o_tile})"
)
transition_memory_payload = {
"memory_write": {
"text": f"Mapped connection: {last_map} -> {current_map}"
}
}
await broadcast_func(transition_memory_payload)
# Reset failed attempts and boost confidence since this exit WORKED
memory_manager.reset_failed_attempts(
last_map,
list(last_pos) if isinstance(last_pos, tuple) else last_pos,
)
# Mark this as a Lass-discovered exit for the minimap overlay
memory_manager.mark_lass_exit(
last_map,
list(last_pos) if isinstance(last_pos, tuple) else last_pos,
destination=current_map,
confidence=0.95 if was_on_o_tile else 0.8,
)
# Update history for next loop
memory_manager.last_map = current_map
memory_manager.last_map_id = current_map_id
memory_manager.last_pos = current_pos
memory_manager.last_minimap_2d = current_mGBA_state.get("minimap_2d", "")
# Extract memories from LLM response and vision analysis
# CRITICAL FIX: Use current_mGBA_state which has map_name and position keys
# Also guard against None analysis_text (occurs when vision fails)
extracted_memories = []
if game_analysis:
extracted_memories = memory_manager.extract_memories_from_response(
analysis_text=game_analysis,
game_state=current_mGBA_state, # Fixed: was 'state' which has different keys
vision_analysis=vision_analysis_for_ui,
)
if extracted_memories:
log.info(
f"📝 Extracted {len(extracted_memories)} memories from LLM response"
)
for memory in extracted_memories:
log.debug(f"Memory: {memory.type} - {memory.description[:50]}...")
# ═══════════════════════════════════════════════════════════════════
# QUEST GOAL CREATION - When we detect quest items, create goals
# Quest goals are CRITICAL priority - they override healing
# ═══════════════════════════════════════════════════════════════════
if hasattr(memory, "quest_id") and memory.quest_id:
# Check if we already have this quest goal
if not goal_tracker.has_quest_goal(memory.quest_id):
goal_id = goal_tracker.add_quest_goal(
quest_id=memory.quest_id,
description=memory.description,
target_location=getattr(
memory, "target_location", None
),
target_npc=getattr(memory, "target_npc", None),
)
if goal_id:
log.info(f"🎯 CREATED QUEST GOAL: {memory.description}")
log.info(
f" Quest ID: {memory.quest_id} | Goal ID: {goal_id}"
)
# Broadcast quest detection to UI immediately
quest_memory_payload = {
"memory_write": {
"text": f"🎯 NEW QUEST: {memory.description}"
}
}
await broadcast_func(quest_memory_payload)
# Verify pending vision claims against minimap data
minimap_2d = current_mGBA_state.get("minimap_2d", "")
if minimap_2d and current_pos:
unverified_claims = memory_manager.get_unverified_claims(limit=3)
for claim in unverified_claims:
was_correct = memory_manager.verify_vision_claim(
claim=claim,
minimap_2d=minimap_2d,
player_pos=list(current_pos) if current_pos else [],
)
if was_correct:
log.info(f"✅ Vision claim VERIFIED: {claim.description}")
else:
log.warning(f"❌ Vision claim WRONG: {claim.description}")
# Log vision accuracy periodically
vision_stats = memory_manager.get_vision_accuracy()
if (
vision_stats.get("verified", 0) > 0
and vision_stats.get("verified", 0) % 5 == 0
):
log.info(f"👁️ {vision_stats['message']}")
# Always get latest memory for broadcasting
latest_memory = memory_manager.get_latest_memory()
if latest_memory:
# Add memory directly to update_payload for frontend compatibility
memory_write_payload = {
"memory_write": {"text": latest_memory.description}
}
log.info(
f"📝 Broadcasting memory_write to UI: {latest_memory.description[:100]}..."
)
# BROADCAST IMMEDIATELY - the main update_payload was already broadcast earlier
# We need a separate broadcast here for memory_write to reach the UI
await broadcast_func(memory_write_payload)
# Also add to memory_updates array for potential future use
if "memory_updates" not in update_payload:
update_payload["memory_updates"] = []
update_payload["memory_updates"].append(memory_write_payload)
except Exception as e:
log.error(f"Error extracting memories: {e}", exc_info=True)
latest_memory = None
# Persist run state periodically (after memory extraction so latest_memory is available)
cycles_since_persist += 1
if persistence and run_state and cycles_since_persist >= PERSIST_INTERVAL:
run_state.cycle_count = cycle_count
run_state.action_count = action_count
run_state.tokens_used = tokens_used_session
run_state.elapsed_seconds = elapsed_seconds
run_state.goals = state.get("goals", run_state.goals)
run_state.other_goals = state.get("otherGoals", run_state.other_goals)
run_state.chat_history = controller.chat_history[
-20:
] # Keep last 20 messages
run_state.recent_actions.append(action if action else "NONE")
run_state.recent_actions = run_state.recent_actions[-50:] # Keep last 50
if latest_memory:
run_state.latest_memory = latest_memory.description
# Save preplanned names for persistence across restarts
run_state.preplanned_names = name_planner.to_dict()
run_state.starter_choice = starter_planner.to_dict()
persistence.save_run_state(run_state)
cycles_since_persist = 0
log.info(
f"Persisted run state: cycle={cycle_count}, actions={action_count}, tokens={tokens_used_session}"
)
# Log action to database
if persistence and run_state and action:
try:
persistence.log_action(
run_id=run_state.run_id,
action=action,
screenshot_b64=b64_ss[:1000]
if b64_ss
else None, # Truncate for storage
llm_analysis=game_analysis[:2000] if game_analysis else None,
vision_analysis=vision_analysis_for_ui[:2000]
if vision_analysis_for_ui
else None,
position=current_pos,
map_name=map_name,
metrics=cycle_metrics,
)
except Exception as pe:
log.warning(f"Failed to log action to database: {pe}")
# CRITICAL: Check for system failure
if update_payload and update_payload.get("system_halt"):
log.critical(
"🛑 SYSTEM HALT DETECTED - Terminating agent operation immediately"
)
break # Exit the main loop immediately
# Auto-save game state at end of each cycle
try:
# AUTOSAVE enabled: save to slot 1 every cycle for safety
save_game_state(sock_ref["socket"], slot=1) # Use slot 1 for regular saves
except Exception as e:
log.warning(f"⚠️ Save operation failed: {e} - continuing cycle")
elapsed_loop_time = time.time() - loop_start_time
# ORIGINAL llmdriver.py used max(10, ...).
# Changed to 10s minimum wait as requested.
wait_time = max(
0, interval - elapsed_loop_time
) # No minimum wait - proceed immediately if under interval
if result and result.get("stats", {}).get("action_count", 0) > 0:
log.info(f"💾 Cycle {current_cycle} action execution successful")
t_cycle_end = time.time()
cycle_duration_s = (
t_cycle_end - loop_start_time
) # Processing time (from successful mGBA response)
true_cycle_duration_s = (
t_cycle_end - true_cycle_start
) # True wall clock time (includes mGBA retries)
cycle_metrics["cycle"] = true_cycle_duration_s * 1000 # Store TRUE cycle time
log.info(f"⏱️ Total Cycle Time: {true_cycle_duration_s:.2f}s")
# Detailed timing breakdown
mgba_s = cycle_metrics.get("mGBA", 0) / 1000
snapshot_s = cycle_metrics.get("snapshot", 0) / 1000
deepcopy_s = cycle_metrics.get("deepcopy", 0) / 1000
context_s = cycle_metrics.get("context", 0) / 1000
nav_s = cycle_metrics.get("nav", 0) / 1000
img_prep_s = cycle_metrics.get("img_prep", 0) / 1000
vision_s = cycle_metrics.get("vision", 0) / 1000
llm_s = cycle_metrics.get("llm", 0) / 1000
chat_wait_s = cycle_metrics.get("chat_wait", 0) / 1000
token_s = cycle_metrics.get("token", 0) / 1000
tts_s = cycle_metrics.get("tts", 0) / 1000 if "tts" in cycle_metrics else 0
action_s = (
cycle_metrics.get("action", 0) / 1000 if "action" in cycle_metrics else 0
)
# Calculate accounted vs unaccounted time
accounted_s = (
mgba_s
+ snapshot_s
+ deepcopy_s
+ context_s
+ nav_s
+ img_prep_s
+ vision_s
+ llm_s
+ chat_wait_s
+ token_s
+ tts_s
+ action_s
)
other_s = max(0, true_cycle_duration_s - accounted_s)
log.info(
f"⏱️ Breakdown: mGBA={mgba_s:.1f}s | Snap={snapshot_s:.1f}s | DeepCP={deepcopy_s:.1f}s | "
f"Context={context_s:.1f}s | Nav={nav_s:.1f}s | Img={img_prep_s:.1f}s | "
f"Vision={vision_s:.1f}s | LLM={llm_s:.1f}s | ChatW={chat_wait_s:.1f}s | "
f"Token={token_s:.1f}s | TTS={tts_s:.1f}s | Action={action_s:.1f}s | "
f"Unaccounted={other_s:.1f}s"
)
log.info(
f"Cycle {current_cycle} took {elapsed_loop_time:.2f}s. Waiting {wait_time:.2f}s..."
)
# ═══════════════════════════════════════════════════════════════════
# Broadcast cycle timing to UI
# ═══════════════════════════════════════════════════════════════════
cycle_timing_str = f"{elapsed_loop_time:.1f}s | wait {wait_time:.1f}s"
# We need to update the log_action call (it happened inside llm_stream_action or executed via persistence object?)
# Actually persistence.log_action is called where? It hasn't been called yet for this cycle in this scope?
# Aah, log_action is usually called inside the execute loop or we need to pass these metrics to where it IS called.
# But wait, run_persistence.log_action is called in `run_auto_loop`? No, it's not.
# Looking at previous code, `persistence.log_action` was NOT called in `run_auto_loop` explicitly in the visible code.
# It must be called elsewhere or I missed it.
# Let's check `llm_stream_action` returns `summary_json`?
# Ah, `persistence` is global or passed?
# I need to find where `log_action` is called.
pass # Placeholder to allow finding the callsite in next step if needed, or simply patching likely location.
if broadcast_func:
try:
# Track cycle time for average calculation - use TRUE time
cycle_times_history.append(true_cycle_duration_s)
# Keep only last 20 cycles for average
if len(cycle_times_history) > 20:
cycle_times_history.pop(0)
# Calculate average
avg_cycle_time = (
sum(cycle_times_history) / len(cycle_times_history)
if cycle_times_history
else 0
)
# Broadcast enhanced cycle metrics: current cycle, previous cycle, average, and detailed breakdown
cycle_metrics_payload = {
"cycleTiming": cycle_timing_str,
"currentCycleTime": round(
true_cycle_duration_s, 1
), # TRUE wall clock time
"prevCycleTime": round(prev_cycle_time_s, 1),
"avgCycleTime": round(avg_cycle_time, 1),
"cycleMetrics": {
"mGBA": round(
cycle_metrics.get("mGBA", 0) / 1000, 1
), # Convert ms to s
"snapshot": round(cycle_metrics.get("snapshot", 0) / 1000, 1),
"deepcopy": round(cycle_metrics.get("deepcopy", 0) / 1000, 1),
"context": round(cycle_metrics.get("context", 0) / 1000, 1),
"nav": round(cycle_metrics.get("nav", 0) / 1000, 1),
"img_prep": round(cycle_metrics.get("img_prep", 0) / 1000, 1),
"vision": round(cycle_metrics.get("vision", 0) / 1000, 1),
"llm": round(cycle_metrics.get("llm", 0) / 1000, 1),
"chat_wait": round(cycle_metrics.get("chat_wait", 0) / 1000, 1),
"token": round(cycle_metrics.get("token", 0) / 1000, 1),
"tts": round(cycle_metrics.get("tts", 0) / 1000, 1),
"action": round(cycle_metrics.get("action", 0) / 1000, 1),
"total": round(true_cycle_duration_s, 1),
},
}
await broadcast_func(cycle_metrics_payload)
# Also update the shared state so new clients get the values
state["currentCycleTime"] = round(true_cycle_duration_s, 1)
state["prevCycleTime"] = round(prev_cycle_time_s, 1)
state["avgCycleTime"] = round(avg_cycle_time, 1)
except Exception as e:
log.warning(f"Failed to broadcast cycle timing: {e}")
# Store current cycle time as previous for next iteration - use TRUE time
prev_cycle_time_s = true_cycle_duration_s
# ═══════════════════════════════════════════════════════════════════════════
# CHAT RESPONSE PROCESSING (Twitch + Pump.fun during wait period)
# ═══════════════════════════════════════════════════════════════════════════
# Mark the current time as when game commentary was sent
if twitch_service and twitch_service.is_available:
twitch_service.mark_commentary_timestamp()
if pumpfun_service and pumpfun_service.is_available:
pumpfun_service.mark_commentary_timestamp()
# Process chat messages during wait period
wait_start = time.time()
chat_response_count = 0
max_chat_responses = 3 # Limit responses per cycle to avoid overwhelming
# Flag to detect if new cycle is starting (for interruption)
cycle_interrupted = False
while time.time() - wait_start < wait_time and not cycle_interrupted:
remaining_wait = wait_time - (time.time() - wait_start)
# Check if Twitch service is available and we haven't hit response limit
if (
not twitch_service.is_available
or chat_response_count >= max_chat_responses
):
await asyncio.sleep(min(remaining_wait, 2.0))
continue
# Check if we're in test mode first (handles its own message generation)
from services.twitch_chat_service import CHAT_TEST_MODE, CHAT_TEST_LLM
if CHAT_TEST_MODE:
# Test mode: Generate message immediately, then wait for TTS
import random
if chat_response_count >= max_chat_responses:
await asyncio.sleep(1.0)
continue
# Check time remaining - only need 1 second to start a response
remaining = wait_time - (time.time() - wait_start)
if remaining < 1:
continue
# Generate one random test message FIRST (no pre-delay!)
# Randomly choose between Twitch and Pump.fun format for test messages
if random.random() < 0.5:
test_msg = twitch_service.generate_single_test_message()
else:
# Use Pump.fun test message if available, otherwise fallback to Twitch
if pumpfun_service:
test_msg = pumpfun_service.generate_single_test_message()
else:
test_msg = twitch_service.generate_single_test_message()
if test_msg:
# Decide: spam = skip, else = respond
is_spam = any(
spam in test_msg["message"].lower()
for spam in [
"kekw",
"lul",
"kappa",
"!gamble",
"first",
"asdf",
"zzz",
"spam",
]
)
if is_spam or random.random() < 0.25: # 25% skip rate
log.info(
f"⏭️ [TEST] Skipping spam from @{test_msg['display_name']}: {test_msg['message'][:40]}..."
)
await asyncio.sleep(0.5) # Brief pause before next message
else:
# Generate response - use LLM if CHAT_TEST_LLM=true, otherwise mock
llm_start = time.time()
if CHAT_TEST_LLM and chat_response_service.is_available:
# Use actual LLM (Featherless) for realistic latency testing
try:
response_text = (
await chat_response_service.generate_response(
test_msg["display_name"],
test_msg["message"],
is_past=False,
)
)
llm_time = time.time() - llm_start
log.info(
f'💬 [TEST+LLM] @{test_msg["display_name"]}: "{test_msg["message"]}"'
)
log.info(
f"🎤 [TEST+LLM] Lass responds (LLM took {llm_time:.2f}s): {response_text}"
)
except Exception as e:
log.warning(
f"[TEST+LLM] LLM error, falling back to mock: {e}"
)
response_text = f"Hehe thanks @{test_msg['display_name']}! You're awesome!"
else:
# Use mock responses for speed testing
mock_responses = [
f"Hehe, thanks for watching @{test_msg['display_name']}! You're making this adventure more fun!",
f"Omg hi @{test_msg['display_name']}! I'm so happy you're here with me!",
f"@{test_msg['display_name']} Aww you're so sweet! Let's catch 'em all together!",
f"Thanks @{test_msg['display_name']}! I may get lost sometimes but at least we're lost together!",
f"@{test_msg['display_name']} You're the Pikachu to my Ash! Well... once I actually GET a Pikachu...",
f"Haha @{test_msg['display_name']}! True, but hey, every Pokemon master started somewhere!",
f"Aww @{test_msg['display_name']} that's so nice! You're giving me all the encouragement I need!",
f"@{test_msg['display_name']} Good question! I'm working on it, I promise! Maybe...",
]
response_text = random.choice(mock_responses)
log.info(
f'💬 [TEST] @{test_msg["display_name"]}: "{test_msg["message"]}"'
)
log.info(f"🎤 [TEST] Lass responds: {response_text}")
chat_response_count += 1
# Queue TTS (non-blocking) - if queue full, response will just show in UI
tts_queued = False
if tts_service.is_available:
try:
# Include reply_to metadata for test messages
test_reply_metadata = {
"reply_to": {
"username": test_msg["display_name"],
"platform": "Twitch",
"message": test_msg["message"],
}
}
request = await tts_service.queue_and_start_synthesis(
response_text,
priority=tts_service.PRIORITY_CHAT_RESPONSE,
cycle_id=current_cycle,
metadata=test_reply_metadata,
)
tts_queued = request is not None
if tts_queued:
log.info(f"✅ [TEST] TTS queued (non-blocking)")
else:
log.info(
f"📝 [TEST] TTS queue full, response will only show in UI"
)
except Exception as tts_err:
log.warning(f"🔊 [TEST] TTS error: {tts_err}")
# Broadcast to UI (always - even if TTS queue full)
chat_response_payload = {
"chat_response": {
"username": test_msg["display_name"],
"message": test_msg["message"],
"response": response_text,
"is_test": True,
"tts_queued": tts_queued, # UI can show indicator
"timestamp": int(time.time() * 1000),
}
}
await broadcast_func(chat_response_payload)
# Brief delay before next potential message
await asyncio.sleep(random.uniform(0.5, 1.5))
continue # Continue the wait loop
# Real mode below - get actual messages from ALL configured sources
messages_for_cycle = []
# Get Twitch messages
if twitch_service and twitch_service.is_available:
twitch_msgs = twitch_service.get_messages_for_cycle_or_test()
messages_for_cycle.extend(twitch_msgs)
# Get Pump.fun messages
if pumpfun_service and pumpfun_service.is_available:
pumpfun_msgs = pumpfun_service.get_messages_for_cycle_or_test()
messages_for_cycle.extend(pumpfun_msgs)
# Get Zora messages
if zora_chat_service and zora_chat_service.is_available:
zora_msgs = zora_chat_service.get_messages_for_cycle_or_test()
messages_for_cycle.extend(zora_msgs)
# ═══════════════════════════════════════════════════════════════════
# WHALE DETECTION - Query token balances
# ═══════════════════════════════════════════════════════════════════
# Solana/Pump.fun Whales
token_service = get_token_service()
if token_service.is_available:
# Get unique pump.fun wallet addresses that need whale check
pumpfun_wallets = [
msg.get("user_address")
for msg in messages_for_cycle
if msg.get("source") == "pumpfun"
and msg.get("user_address")
and not msg.get("is_test", False) # Skip test messages
]
if pumpfun_wallets:
# Batch check whale status (uses cache, very efficient)
whale_status = await token_service.check_whale_status_batch(
pumpfun_wallets
)
# Update messages with whale status
for msg in messages_for_cycle:
if msg.get("source") == "pumpfun":
wallet = msg.get("user_address")
if wallet and wallet in whale_status:
msg["is_whale"] = whale_status[wallet]
# Base/Zora Whales
base_token_service = get_base_token_service()
if base_token_service.is_available:
zora_wallets = [
msg.get("user_address")
for msg in messages_for_cycle
if msg.get("source") == "zora"
and msg.get("user_address")
and not msg.get("is_test", False)
]
if zora_wallets:
base_status = await base_token_service.check_holder_status_batch(
zora_wallets
)
for msg in messages_for_cycle:
if msg.get("source") == "zora":
wallet = msg.get("user_address")
if wallet and wallet in base_status:
msg["is_whale"] = base_status[wallet]
# ═══════════════════════════════════════════════════════════════════
# PRIORITY SORTING - Higher priority messages get responded to first
# ═══════════════════════════════════════════════════════════════════
# Priority hierarchy (highest first):
# 1. 👑 WHALE holders (100k+ tokens) - Priority 5
# 2. 🟣 Zora subscribers/holders - Priority 4
# 3. 💎 Twitch subscribers - Priority 3
# 4. 🪙 Regular pump.fun / Zora chatters - Priority 2
# 5. 👤 Regular Twitch chatters - Priority 1
def get_priority(msg: dict) -> int:
source = msg.get("source", "twitch")
is_whale = msg.get("is_whale", False)
is_subscriber = msg.get("is_subscriber", False)
if is_whale:
return 5 # Whale - highest priority (any chain)
elif source == "zora" and is_subscriber:
return 4 # Zora subscriber
elif source == "twitch" and is_subscriber:
return 3 # Twitch subscriber
elif source in ("pumpfun", "zora"):
return 2 # Regular crypto chatter
else:
return 1 # Regular Twitch chatter
# Sort by priority (descending), then by timestamp (ascending/oldest first)
messages_for_cycle.sort(
key=lambda m: (-get_priority(m), m.get("timestamp", 0))
)
if not messages_for_cycle:
await asyncio.sleep(min(remaining_wait, 2.0))
continue
# Real mode: Use chat response service for decisions
if not chat_response_service.is_available:
await asyncio.sleep(min(remaining_wait, 2.0))
continue
# Decide SKIP/RESPOND for all messages at once
decisions = await chat_response_service.decide_skip_or_respond(
messages_for_cycle
)
# Process RESPOND messages oldest first
for decided in decisions:
# Find original message object and metadata
msg_data = next(
(
m
for m in messages_for_cycle
if m["timestamp"] == decided.timestamp
),
None,
)
original_msg = msg_data.get("_original") if msg_data else None
msg_source = msg_data.get("source", "twitch") if msg_data else "twitch"
if decided.decision == MessageDecision.SKIP:
log.info(
f"⏭️ Skipping message from @{decided.display_name} ({msg_source})"
)
# Mark as responded so we don't process again
if original_msg:
if msg_source == "twitch" and twitch_service:
twitch_service.mark_responded(original_msg)
elif hasattr(original_msg, "responded"):
original_msg.responded = True
continue
# Check if we should stop for new cycle
if time.time() - wait_start >= wait_time:
log.info("🔄 Cycle time up - interrupting chat responses")
tts_service.cancel_current() # Cancel any playing audio
cycle_interrupted = True
break
# Generate and send response
response_text = await chat_response_service.generate_response(
decided.display_name, decided.message, is_past=False
)
if response_text:
# Format response with @ mention if not already present
if not response_text.startswith("@"):
response_text = f"@{decided.display_name} {response_text}"
# Mark original message as responded
if original_msg:
if msg_source == "twitch" and twitch_service:
twitch_service.mark_responded(original_msg)
elif hasattr(original_msg, "responded"):
original_msg.responded = True
chat_response_count += 1
# Queue TTS for the response (non-blocking) - if full, text still goes to chat
tts_queued = False
if tts_service.is_available:
# Pass context about what we are replying to
reply_metadata = {
"reply_to": {
"username": decided.display_name,
"platform": msg_source.capitalize(),
"message": decided.message,
}
}
request = await tts_service.queue_and_start_synthesis(
response_text,
priority=tts_service.PRIORITY_CHAT_RESPONSE,
cycle_id=current_cycle,
metadata=reply_metadata,
)
tts_queued = request is not None
# Send response to appropriate platform
if msg_source == "twitch" and twitch_service:
await twitch_service.send_response(
decided.display_name,
response_text.replace(f"@{decided.display_name} ", ""),
)
elif msg_source == "pumpfun" and pumpfun_service:
await pumpfun_service.send_message(response_text)
elif msg_source == "zora":
# Zora is currently read-only (responses shown in OBS overlay/TTS only)
pass
# Broadcast chat response to OBS widget
chat_response_payload = {
"chat_response": {
"username": decided.display_name,
"response": response_text,
"source": msg_source,
"is_past_message": False,
"timestamp": int(time.time() * 1000),
}
}
await broadcast_func(chat_response_payload)
log.info(
f"✅ Chat response sent ({msg_source}): {response_text[:50]}..."
)
# If we processed all messages, sleep briefly
await asyncio.sleep(min(remaining_wait, 1.0))
log.info("Auto loop terminated.")
if benchmark is not None:
benchmark.finalize(current_mGBA_state, MODEL)