feat: add configurable throttle parameter to update() method for granular UI broadcast control

Author: manatlan

.agent/skills/htag-development/SKILL.md

@@ -121,7 +121,7 @@ Declare individual state variables when you only need one or are passing it as a
 - Use the `.text` property to quickly replace all text content of a tag: `self.my_label.text = "New Status"`. This completely clears existing children and replaces them with a single string.
 **Background Tasks & `update()`**:
 - **Automatic Reactivity**: When using `State`, mutations from background tasks (started via `asyncio.create_task`) automatically trigger UI synchronization.
-- **Manual Synchronization**: For non-state changes or complex manual updates, every component exposes an `.update()` method (e.g., `self.update()`) which schedules a throttled UI broadcast. 
+- **Manual Synchronization**: For non-state changes or complex manual updates, every component exposes an `.update()` method (e.g., `self.update()`) which schedules a throttled UI broadcast. By default, updates are throttled at 50ms (configurable). You can adjust this on the fly: use `self.update(throttle=0.1)` for 100ms throttle, or `self.update(throttle=0)` to force an immediate broadcast.
 - **Pattern**: Use `on_mount` to start background tasks and `on_unmount` to cancel them. (This pattern is now robust against page refreshes in `WebApp`).
 
 ```python

docs/components.md

@@ -63,7 +63,7 @@ class Clock(Tag.div):
 ### Background Tasks & `update()`
 
 - **Automatic Reactivity**: `State` mutations from background tasks (started via `asyncio.create_task`) now automatically trigger UI updates.
-- **Manual Synchronization**: For non-state changes, use `self.update()` to schedule a throttled UI sync.
+- **Manual Synchronization**: For non-state changes, use `self.update()` to schedule a throttled UI sync. By default, it is throttled at 50ms. You can adjust it on the fly (e.g. `self.update(throttle=0.1)` for a 100ms delay, or `self.update(throttle=0)` to force an immediate broadcast).
 
 ```python
     def on_mount(self):

htag/core.py

@@ -800,14 +800,14 @@ def call_js(self, script: str) -> "GTag":
         self.__js_calls.append(script)
         return self
 
-    def update(self) -> None:
+    def update(self, throttle: float | None = None) -> None:
         """
         Schedules a UI synchronization.
         Useful when modifying state from background tasks or outside regular event handlers.
         """
         root = self.root
         if root and hasattr(root, "update"):
-            root.update()
+            root.update(throttle=throttle)
 
     # --- Public API for server-side access (avoids name-mangled access) ---
 

htag/runner.py

@@ -146,6 +146,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.websockets: set[WebSocket] = set()
         self.sse_queues: set[asyncio.Queue] = set()  # Queues for active SSE connections
         self.sent_statics: set[str] = set()  # Track assets already in browser
+        self.throttle_delay: float = 0.05
+        self._last_broadcast_time: float = 0.0
 
     @property
     def app(self) -> Any:
@@ -566,24 +568,48 @@ async def _consume_generator(self, gen: Any, ws: WebSocket | None = None) -> Non
             if not has_sent:
                 self._push_to_fallback(err_payload)
 
-    def update(self) -> None:
+    def update(self, throttle: float | None = None) -> None:
         """
-        Schedules a broadcast of all pending updates in the next event loop iteration.
-        This provides a throttled, asynchronous update mechanism that can be safely
-        triggered from both synchronous and asynchronous code.
+        Schedules a broadcast of all pending updates after a throttle delay.
         """
-        if not getattr(self, "_update_scheduled", False):
-            try:
-                loop = asyncio.get_running_loop()
-                self._update_scheduled = True
-                loop.call_soon(self._do_update)
-            except RuntimeError:
-                # No running loop (might happen during early initialization)
-                pass
+        try:
+            loop = asyncio.get_running_loop()
+            now = loop.time()
+            elapsed = now - getattr(self, "_last_broadcast_time", 0.0)
+            throttle_delay = throttle if throttle is not None else getattr(self, "throttle_delay", 0.05)
+            
+            target_delay = 0.0 if elapsed >= throttle_delay else (throttle_delay - elapsed)
+            
+            if getattr(self, "_update_scheduled", False):
+                scheduled_at = getattr(self, "_update_scheduled_at", 0.0)
+                remaining = scheduled_at - now
+                if target_delay < remaining:
+                    if getattr(self, "_update_handle", None) is not None:
+                        self._update_handle.cancel()
+                else:
+                    return
+            
+            self._update_scheduled = True
+            self._update_scheduled_at = now + target_delay
+            
+            if target_delay == 0.0:
+                self._update_handle = loop.call_soon(self._do_update)
+            else:
+                self._update_handle = loop.call_later(target_delay, self._do_update)
+        except RuntimeError:
+            # No running loop
+            pass
 
     def _do_update(self) -> None:
-        """Internal callback to handle the actual broadcast after call_soon."""
+        """Internal callback to handle the actual broadcast after call_soon/call_later."""
         self._update_scheduled = False
+        self._update_handle = None
+        try:
+            loop = asyncio.get_running_loop()
+            self._last_broadcast_time = loop.time()
+        except RuntimeError:
+            import time
+            self._last_broadcast_time = time.time()
         asyncio.create_task(self.broadcast_updates())
 
     async def broadcast_updates(

tests/test_reactivity_edge_cases.py

@@ -177,3 +177,43 @@ def test_eval_child_non_stringify():
     assert val is True
     val_str = t._eval_child(True, stringify=True)
     assert val_str == "True"
+
+
+@pytest.mark.asyncio
+async def test_update_throttle():
+    """Test that Runner.update() properly throttles high-frequency updates"""
+    import time
+    class MyApp(App):
+        def init(self):
+            self.val = 0
+
+    app = MyApp()
+    broadcasts = []
+    
+    async def mock_broadcast(*args, **kwargs):
+        broadcasts.append(asyncio.get_running_loop().time())
+        
+    app.broadcast_updates = mock_broadcast
+    app.throttle_delay = 0.1  # 100ms throttle
+    
+    # 1. Trigger multiple updates in a row
+    app.update()
+    app.update()
+    app.update()
+    
+    # Give the event loop a tick to run call_soon
+    await asyncio.sleep(0.01)
+    assert len(broadcasts) == 1
+    
+    # 2. Trigger updates while throttled
+    app.update()  # should be scheduled for +100ms
+    app.update()
+    
+    await asyncio.sleep(0.05)
+    # Still only 1 broadcast (since we are at 50ms and delay is 100ms)
+    assert len(broadcasts) == 1
+    
+    await asyncio.sleep(0.06)
+    # Now we reached 110ms, the throttled update should have executed
+    assert len(broadcasts) == 2
+