refactor: extract shared utilities, fix send_data signature & plugins.py bug
- Consolidate ConnectionIdentity, _ws_debug_enabled(), ws_debug() into ws.py as single-source exports, removing duplicate definitions in ws_manager.py and state_monitor.py - Make send_data() optional args keyword-only to prevent positional argument confusion with the instance method signature - Fix clear_plugin_cache in plugins.py: wrong parameter name (event_name → event_type) and stale namespace (/webui → /ws)
keyboardstaff committed
Mar 28, 2026 at 00:38 UTC
04d930ab02c43b9b9ff6db7ead83f85ce6cf3113
4 files changed
+38
-40
helpers/plugins.py
+3
-3
@@ -206,9 +206,9 @@ def clear_plugin_cache(plugin_names: list[str] | None = None):
206
207
DeferredTask().start_task(
208
send_data,
209
- endpoint_name="/webui",
210
- event_name="clear_cache",
211
- data={"areas": areas},
209
+ "clear_cache",
210
+ {"areas": areas},
211
+ endpoint_name="/ws",
212
)
213
214
helpers/state_monitor.py
+10
-24
@@ -14,26 +14,12 @@ from helpers.state_snapshot import (
14
advance_state_request_after_snapshot,
15
build_snapshot_from_request,
16
)
17
-from helpers.ws import ConnectionNotFoundError
17
+from helpers.ws import ConnectionIdentity, ConnectionNotFoundError, _ws_debug_enabled, ws_debug
18
19
if TYPE_CHECKING: # pragma: no cover - hints only
20
from helpers.ws_manager import WsManager
21
22
23
-ConnectionIdentity = tuple[str, str] # (namespace, sid)
24
-
25
-
26
-def _ws_debug_enabled() -> bool:
27
- value = os.getenv("A0_WS_DEBUG", "").strip().lower()
28
- return value in {"1", "true", "yes", "on"}
29
-
30
-
31
-def _debug_log(message: str) -> None:
32
- if not _ws_debug_enabled():
33
- return
34
- PrintStyle.debug(message)
35
-
36
-
23
@dataclass
24
class ConnectionProjection:
25
namespace: str
@@ -73,7 +59,7 @@ class StateMonitor:
59
# Use the manager's dispatcher loop for all scheduling so mark_dirty can be
60
# invoked safely from non-async contexts and other threads.
61
self._dispatcher_loop = getattr(manager, "_dispatcher_loop", None)
76
- _debug_log(
62
+ ws_debug(
63
f"[StateMonitor] bind_manager handler_id={handler_id or self._emit_handler_id}"
64
)
65
@@ -83,7 +69,7 @@ class StateMonitor:
69
self._projections.setdefault(
70
identity, ConnectionProjection(namespace=namespace, sid=sid)
71
)
86
- _debug_log(f"[StateMonitor] register_sid namespace={namespace} sid={sid}")
72
+ ws_debug(f"[StateMonitor] register_sid namespace={namespace} sid={sid}")
73
74
def unregister_sid(self, namespace: str, sid: str) -> None:
75
identity: ConnectionIdentity = (namespace, sid)
@@ -95,7 +81,7 @@ class StateMonitor:
81
if task is not None:
82
task.cancel()
83
self._projections.pop(identity, None)
98
- _debug_log(f"[StateMonitor] unregister_sid namespace={namespace} sid={sid}")
84
+ ws_debug(f"[StateMonitor] unregister_sid namespace={namespace} sid={sid}")
85
86
def mark_dirty_all(self, *, reason: str | None = None) -> None:
87
wave_id = None
@@ -142,7 +128,7 @@ class StateMonitor:
128
projection.request = request
129
projection.seq_base = seq_base
130
projection.seq = seq_base
145
- _debug_log(
131
+ ws_debug(
132
f"[StateMonitor] update_projection namespace={namespace} sid={sid} context={request.context!r} "
133
f"log_from={request.log_from} notifications_from={request.notifications_from} "
134
f"timezone={request.timezone!r} seq_base={seq_base}"
@@ -221,7 +207,7 @@ class StateMonitor:
207
self.debounce_seconds, self._on_debounce_fire, identity
208
)
209
self._debounce_handles[identity] = handle
224
- _debug_log(
210
+ ws_debug(
211
f"[StateMonitor] schedule_push namespace={projection.namespace} sid={projection.sid} "
212
f"delay_s={self.debounce_seconds} "
213
f"dirty={projection.dirty_version} pushed={projection.pushed_version} "
@@ -298,7 +284,7 @@ class StateMonitor:
284
if isinstance(snapshot.get("logs"), list)
285
else None
286
)
301
- _debug_log(
287
+ ws_debug(
288
f"[StateMonitor] emit state_push namespace={namespace} sid={sid} seq={seq} "
289
f"context={request.context!r} logs_len={logs_len} "
290
f"reason={dirty_reason!r} wave={dirty_wave_id!r}"
@@ -312,13 +298,13 @@ class StateMonitor:
298
)
299
except ConnectionNotFoundError:
300
# Sid was removed before the emit; treat as benign.
315
- _debug_log(
301
+ ws_debug(
302
f"[StateMonitor] emit skipped: sid not found namespace={namespace} sid={sid}"
303
)
304
return
305
except RuntimeError:
306
# Dispatcher loop may be closing (e.g., during shutdown or test teardown).
321
- _debug_log(
307
+ ws_debug(
308
f"[StateMonitor] emit skipped: dispatcher closing namespace={namespace} sid={sid}"
309
)
310
return
@@ -341,7 +327,7 @@ class StateMonitor:
327
if not follow_up:
328
return
329
344
- _debug_log(
330
+ ws_debug(
331
f"[StateMonitor] follow_up_push namespace={namespace} sid={sid} dirty={dirty_version} pushed={pushed_version}"
332
)
333
try:
helpers/ws.py
+16
-1
@@ -1,3 +1,4 @@
1
+import os
2
import threading
3
import uuid
4
from abc import abstractmethod
@@ -17,10 +18,24 @@ if TYPE_CHECKING:
18
from helpers.ws_manager import WsManager
19
20
20
-# Utilities
21
+# Shared types and utilities
22
23
from helpers.network import is_loopback_address
24
25
+ConnectionIdentity = tuple[str, str] # (namespace, sid)
26
+
27
+
28
+def _ws_debug_enabled() -> bool:
29
+ """Check A0_WS_DEBUG env var — lightweight, no heavy imports."""
30
+ value = os.getenv("A0_WS_DEBUG", "").strip().lower()
31
+ return value in {"1", "true", "yes", "on"}
32
+
33
+
34
+def ws_debug(message: str) -> None:
35
+ """Log *message* via :class:`PrintStyle` when ``A0_WS_DEBUG`` is active."""
36
+ if _ws_debug_enabled():
37
+ PrintStyle.debug(message)
38
+
39
40
class ConnectionNotFoundError(RuntimeError):
41
"""Raised when attempting to emit to a non-existent WebSocket connection."""
helpers/ws_manager.py
+9
-12
@@ -15,13 +15,7 @@ import uuid
15
from helpers.defer import DeferredTask
16
from helpers.print_style import PrintStyle
17
from helpers import runtime
18
-from helpers.ws import ConnectionNotFoundError, WsHandler
19
-
20
-
21
-def _ws_debug_enabled() -> bool:
22
- """Check A0_WS_DEBUG env var — no heavyweight imports needed."""
23
- value = os.getenv("A0_WS_DEBUG", "").strip().lower()
24
- return value in {"1", "true", "yes", "on"}
18
+from helpers.ws import ConnectionIdentity, ConnectionNotFoundError, WsHandler, _ws_debug_enabled, ws_debug
19
20
21
# Event validation
@@ -182,9 +176,16 @@ _shared_ws_manager: WsManager | None = None
176
async def send_data(
177
event_type: str,
178
data: dict[str, Any],
179
+ *,
180
endpoint_name: str = "/ws",
181
connection_id: str | None = None,
182
) -> None:
183
+ """Convenience wrapper around :pymeth:`WsManager.send_data`.
184
+
185
+ All optional parameters are keyword-only to match the instance method's
186
+ ``(endpoint_name, event_type, data, connection_id)`` order and avoid
187
+ positional confusion between the two signatures.
188
+ """
189
manager = get_shared_ws_manager()
190
await manager.send_data(endpoint_name, event_type, data, connection_id)
191
@@ -222,9 +223,6 @@ class ConnectionInfo:
223
last_activity: datetime = field(default_factory=_utcnow)
224
225
225
-ConnectionIdentity = tuple[str, str] # (namespace, sid)
226
-
227
-
226
@dataclass
227
class _HandlerExecution:
228
handler: WsHandler
@@ -262,8 +260,7 @@ class WsManager:
260
261
# Internal: development-only debug logging to avoid noise in production
262
def _debug(self, message: str) -> None:
265
- if _ws_debug_enabled():
266
- PrintStyle.debug(message)
263
+ ws_debug(message)
264
265
def _ensure_dispatcher_loop(self) -> None:
266
if self._dispatcher_loop is None: