refactor(app): extract helper functions for manual action updates

README.md

@@ -10,6 +10,7 @@ pinned: true
 license: mit
 short_description: Multi-stage MCP-driven advisor fights fires autonomously
 tags:
+- mcp-1st-birthday
 - mcp-in-action-track-creative
 - agent
 - mcp
@@ -188,6 +189,9 @@ uv run python app.py
 
 # Or plain Python
 python -m app
+
+# Or use the helper restart script for local runs
+./restart.sh
 ```
 
 Visit http://localhost:7860 to start simulating.

app.py

@@ -775,6 +775,44 @@ def prepare_reset(
     return None, None
 
 
+def _ensure_display_cache_initialized(display_cache: Optional[dict]) -> dict:
+    """Guarantee we always have a mutable display cache."""
+    return display_cache if display_cache is not None else _create_display_cache()
+
+
+def _render_manual_action_update(service: SimulationService, display_cache: dict):
+    """Return the standard UI updates after a manual grid action."""
+    state = service.get_state()
+    updates = get_all_button_updates(state)
+    is_thinking = service.is_thinking()
+    thinking_stage = service.get_thinking_stage()
+    return [
+        render_game_result(state.get("status", ""), state.get("after_action_report")),
+        _get_combined_advisor_messages(service),
+        service.get_event_log_text(),
+        render_status_html(state, is_thinking, thinking_stage),
+    ] + updates + [service, display_cache]
+
+
+def _normalize_manual_selection(selection: str) -> str:
+    """Map flexible text inputs to the radio labels used by the UI."""
+    if not selection:
+        return "🚒 Truck"
+    normalized = selection.strip().lower()
+    mappings = {
+        "truck": "🚒 Truck",
+        "fire_truck": "🚒 Truck",
+        "firetruck": "🚒 Truck",
+        "🚒": "🚒 Truck",
+        "heli": "🚁 Heli",
+        "helicopter": "🚁 Heli",
+        "🚁": "🚁 Heli",
+        "fire": "🔥 Fire",
+        "🔥": "🔥 Fire",
+    }
+    return mappings.get(normalized, selection if selection in ["🚒 Truck", "🚁 Heli", "🔥 Fire"] else "🚒 Truck")
+
+
 def deploy_at_cell(
     x: int,
     y: int,
@@ -784,24 +822,12 @@ def deploy_at_cell(
 ):
     """Deploy unit or fire at specific cell, or remove if unit already exists there."""
     service = get_or_create_service(service)
-    if display_cache is None:
-        display_cache = _create_display_cache()
-    
-    # Get thinking state
-    is_thinking = service.is_thinking()
-    thinking_stage = service.get_thinking_stage()
+    display_cache = _ensure_display_cache_initialized(display_cache)
     
     # Only allow deployment when simulation is actively running (not paused)
     if not service.is_running():
         gr.Warning("⚠️ Please start the simulation first!")
-        state = service.get_state()
-        updates = get_all_button_updates(state)
-        return [
-            render_game_result(state.get("status", ""), state.get("after_action_report")),
-            _get_combined_advisor_messages(service),
-            service.get_event_log_text(),
-            render_status_html(state, is_thinking, thinking_stage),
-        ] + updates + [service, display_cache]
+        return _render_manual_action_update(service, display_cache)
     
     # Handle fire placement
     if selection == "🔥 Fire":
@@ -823,19 +849,96 @@ def deploy_at_cell(
             error_msg = result.get("message", "Unknown error")
             gr.Warning(f"⚠️ {error_msg}")
     
-    state = service.get_state()
-    updates = get_all_button_updates(state)
+    return _render_manual_action_update(service, display_cache)
+
+
+def handle_map_deploy(
+    selection: str,
+    x: int,
+    y: int,
+    service: Optional[SimulationService],
+    display_cache: Optional[dict],
+):
+    """Public API handler to deploy units/fires at a coordinate."""
+    normalized = _normalize_manual_selection(selection)
+    return deploy_at_cell(
+        int(x),
+        int(y),
+        normalized,
+        service,
+        display_cache,
+    )
+
+
+def handle_map_remove(
+    x: int,
+    y: int,
+    service: Optional[SimulationService],
+    display_cache: Optional[dict],
+):
+    """Public API handler to remove a unit at coordinates."""
+    service = get_or_create_service(service)
+    display_cache = _ensure_display_cache_initialized(display_cache)
     
-    # Refresh thinking state after action
-    is_thinking = service.is_thinking()
-    thinking_stage = service.get_thinking_stage()
+    if not service.is_running():
+        gr.Warning("⚠️ Please start the simulation first!")
+        return _render_manual_action_update(service, display_cache)
     
-    return [
-        render_game_result(state.get("status", ""), state.get("after_action_report")),
-        _get_combined_advisor_messages(service),
-        service.get_event_log_text(),
-        render_status_html(state, is_thinking, thinking_stage),
-    ] + updates + [service, display_cache]
+    x = int(x)
+    y = int(y)
+    
+    if not service.has_unit_at(x, y):
+        gr.Warning("⚠️ No unit found at the requested coordinates.")
+        return _render_manual_action_update(service, display_cache)
+    
+    result = service.remove_unit(x, y)
+    if result.get("status") != "ok":
+        error_msg = result.get("message", "Unknown error")
+        gr.Warning(f"⚠️ {error_msg}")
+    
+    return _render_manual_action_update(service, display_cache)
+
+
+def handle_map_move(
+    source_x: int,
+    source_y: int,
+    target_x: int,
+    target_y: int,
+    service: Optional[SimulationService],
+    display_cache: Optional[dict],
+):
+    """Public API handler to move a unit between coordinates."""
+    service = get_or_create_service(service)
+    display_cache = _ensure_display_cache_initialized(display_cache)
+    
+    if not service.is_running():
+        gr.Warning("⚠️ Please start the simulation first!")
+        return _render_manual_action_update(service, display_cache)
+    
+    sx, sy = int(source_x), int(source_y)
+    tx, ty = int(target_x), int(target_y)
+    
+    if not service.has_unit_at(sx, sy):
+        gr.Warning("⚠️ No unit found at the source coordinates.")
+        return _render_manual_action_update(service, display_cache)
+    
+    removal = service.remove_unit(sx, sy)
+    if removal.get("status") != "ok":
+        error_msg = removal.get("message", "Failed to remove unit before moving.")
+        gr.Warning(f"⚠️ {error_msg}")
+        return _render_manual_action_update(service, display_cache)
+    
+    removed_unit = removal.get("unit") or {}
+    unit_type = removed_unit.get("type") or removal.get("removed_unit_type") or "fire_truck"
+    
+    deploy_result = service.deploy_unit(unit_type, tx, ty, "player_move")
+    if deploy_result.get("status") != "ok":
+        # Attempt to restore the original placement
+        service.deploy_unit(unit_type, sx, sy, "player_move_rollback")
+        error_msg = deploy_result.get("message", "Failed to deploy unit at target location.")
+        gr.Warning(f"⚠️ {error_msg}")
+    
+    return _render_manual_action_update(service, display_cache)
 
 
 def poll_after_action_report(service: Optional[SimulationService]):
@@ -2021,12 +2124,28 @@ def create_app() -> gr.Blocks:
                                 )
                                 grid_buttons.append((x, y, btn))
         
+        # Hidden components that only exist to expose MCP-friendly APIs (kept out of UI)
+        with gr.Column(visible=False):
+            manual_deploy_selection = gr.Textbox(label="deploy_selection", value="🚒 Truck")
+            manual_deploy_x = gr.Number(label="deploy_x", value=0, precision=0)
+            manual_deploy_y = gr.Number(label="deploy_y", value=0, precision=0)
+            manual_move_source_x = gr.Number(label="move_source_x", value=0, precision=0)
+            manual_move_source_y = gr.Number(label="move_source_y", value=0, precision=0)
+            manual_move_target_x = gr.Number(label="move_target_x", value=0, precision=0)
+            manual_move_target_y = gr.Number(label="move_target_y", value=0, precision=0)
+            manual_remove_x = gr.Number(label="remove_x", value=0, precision=0)
+            manual_remove_y = gr.Number(label="remove_y", value=0, precision=0)
+            manual_deploy_trigger = gr.Button("manual_deploy_trigger", visible=False)
+            manual_move_trigger = gr.Button("manual_move_trigger", visible=False)
+            manual_remove_trigger = gr.Button("manual_remove_trigger", visible=False)
+        
         # Timers: main simulation refresh + after-action poller
         timer = gr.Timer(value=1.0, active=False)
         report_timer = gr.Timer(value=1.0, active=False)
         
         # Collect all button outputs for updates
         all_buttons = [btn for (_, _, btn) in grid_buttons]
+        manual_action_outputs = [result_popup, advisor_display, event_log_display, status_display] + all_buttons + [service_state, display_cache_state]
         
         # Event handlers for simulation controls
         start_btn.click(
@@ -2111,6 +2230,38 @@ def create_app() -> gr.Blocks:
             inputs=[model_selector, service_state],
             outputs=[service_state],
         )
+        
+        # Public MCP endpoints for manual deploy/move/remove without exposing all grid buttons
+        manual_deploy_trigger.click(
+            fn=handle_map_deploy,
+            inputs=[manual_deploy_selection, manual_deploy_x, manual_deploy_y, service_state, display_cache_state],
+            outputs=manual_action_outputs,
+            api_name="deploy_unit_manual",
+            api_visibility="public",
+        )
+        
+        manual_move_trigger.click(
+            fn=handle_map_move,
+            inputs=[
+                manual_move_source_x,
+                manual_move_source_y,
+                manual_move_target_x,
+                manual_move_target_y,
+                service_state,
+                display_cache_state,
+            ],
+            outputs=manual_action_outputs,
+            api_name="move_unit_manual",
+            api_visibility="public",
+        )
+        
+        manual_remove_trigger.click(
+            fn=handle_map_remove,
+            inputs=[manual_remove_x, manual_remove_y, service_state, display_cache_state],
+            outputs=manual_action_outputs,
+            api_name="remove_unit_manual",
+            api_visibility="public",
+        )
         app.load(
             fn=_initialize_session_defaults,
             inputs=[service_state, display_cache_state],