| 1 | from helpers.ws import WsHandler |
| 2 | from helpers import extension |
| 3 | |
| 4 | |
| 5 | class WsWebui(WsHandler): |
| 6 | """State synchronisation handler — the primary WebSocket endpoint for the UI.""" |
| 7 | |
| 8 | async def on_connect(self, sid: str) -> None: |
| 9 | await extension.call_extensions_async( |
| 10 | "webui_ws_connect", agent=None, instance=self, sid=sid |
| 11 | ) |
| 12 | |
| 13 | async def on_disconnect(self, sid: str) -> None: |
| 14 | await extension.call_extensions_async( |
| 15 | "webui_ws_disconnect", agent=None, instance=self, sid=sid |
| 16 | ) |
| 17 | |
| 18 | async def process(self, event: str, data: dict, sid: str) -> dict | None: |
| 19 | response_data: dict = {} |
| 20 | |
| 21 | await extension.call_extensions_async( |
| 22 | "webui_ws_event", |
| 23 | agent=None, |
| 24 | instance=self, |
| 25 | sid=sid, |
| 26 | event_type=event, |
| 27 | data=data, |
| 28 | response_data=response_data, |
| 29 | ) |
| 30 | |
| 31 | # Return None (fire-and-forget) when no extension populated the response. |
| 32 | return response_data if response_data else None |