feat(api): Add GUI state manipulation hooks with thread-safe queueing

api_hooks.py

@@ -49,6 +49,19 @@ class HookHandler(BaseHTTPRequestHandler):
                 self.send_header('Content-Type', 'application/json')
                 self.end_headers()
                 self.wfile.write(json.dumps({'status': 'updated'}).encode('utf-8'))
+            elif self.path == '/api/gui':
+                if not hasattr(app, '_pending_gui_tasks'):
+                    app._pending_gui_tasks = []
+                if not hasattr(app, '_pending_gui_tasks_lock'):
+                    app._pending_gui_tasks_lock = threading.Lock()
+                    
+                with app._pending_gui_tasks_lock:
+                    app._pending_gui_tasks.append(data)
+                    
+                self.send_response(200)
+                self.send_header('Content-Type', 'application/json')
+                self.end_headers()
+                self.wfile.write(json.dumps({'status': 'queued'}).encode('utf-8'))
             else:
                 self.send_response(404)
                 self.end_headers()

gui.py

@@ -473,6 +473,10 @@ class App:
         self._pending_history_adds: list[dict] = []
         self._pending_history_adds_lock = threading.Lock()
         
+        # API GUI Hooks Queue
+        self._pending_gui_tasks: list[dict] = []
+        self._pending_gui_tasks_lock = threading.Lock()
+        
         # Blink state
         self._trigger_blink = False
         self._is_blinking = False
@@ -2085,6 +2089,27 @@ class App:
                     # Force scroll to bottom using a very large number
                     dpg.set_y_scroll("disc_scroll", 99999)
 
+            # Process queued API GUI tasks
+            with self._pending_gui_tasks_lock:
+                gui_tasks = self._pending_gui_tasks[:]
+                self._pending_gui_tasks.clear()
+            for task in gui_tasks:
+                try:
+                    action = task.get("action")
+                    if action == "set_value":
+                        item = task.get("item")
+                        val = task.get("value")
+                        if item and dpg.does_item_exist(item):
+                            dpg.set_value(item, val)
+                    elif action == "click":
+                        item = task.get("item")
+                        if item and dpg.does_item_exist(item):
+                            cb = dpg.get_item_callback(item)
+                            if cb:
+                                cb()
+                except Exception as e:
+                    print(f"Error executing GUI hook task: {e}")
+
             # Handle retro arcade blinking effect
             if self._trigger_script_blink:
                 self._trigger_script_blink = False

tests/test_hooks.py

@@ -69,5 +69,15 @@ def test_ipc_server_starts_and_responds():
             assert response.status == 200
             assert app_mock.disc_entries == [{"role": "User", "content": "hi"}]
 
+        # Test GUI queue hook
+        req = urllib.request.Request("http://127.0.0.1:8999/api/gui", method="POST", data=json.dumps({"action": "set_value", "item": "test_item", "value": "test_value"}).encode("utf-8"), headers={'Content-Type': 'application/json'})
+        with urllib.request.urlopen(req) as response:
+            assert response.status == 200
+            # Instead of checking DPG (since we aren't running the real main loop in tests),
+            # check if it got queued in app_mock
+            assert hasattr(app_mock, '_pending_gui_tasks')
+            assert len(app_mock._pending_gui_tasks) == 1
+            assert app_mock._pending_gui_tasks[0] == {"action": "set_value", "item": "test_item", "value": "test_value"}
+            
     finally:
         server.stop()