WebSocket Session Management#

Overview#

LangBot's WebSocket layer enables real-time, bidirectional chat in the web dashboard (Debug Chat) and embedded chat widgets. Isolation across concurrent connections is enforced at four levels: pipeline, session type, browser session, and connection.

Four-Level Isolation Architecture#

1. Connection (WebSocketConnection)#

Each incoming WebSocket is wrapped in a WebSocketConnection Pydantic model carrying connection_id (UUID4), pipeline_uuid, session_type, session_id (optional client conversation identifier used by embed widgets; must be a valid UUID v4), a per-connection send_queue, and last_active timestamp. The connection_id is echoed to the client in the connected handshake 1.

2. Pipeline isolation#

WebSocketConnectionManager 2 is a process-wide singleton 3 that maintains three indexes:

• connections — flat map of connection_id → WebSocketConnection
• pipeline_connections — pipeline_uuid → {connection_id, …}
• session_connections — session_type → {connection_id, …}

All mutations go through an asyncio.Lock 4 to prevent race conditions under concurrent connections.

Response broadcast uses broadcast_to_pipeline(pipeline_uuid, message, session_type, session_id), which first narrows to connections for that pipeline, then optionally narrows further by session_type. When session_id is omitted (default), messages broadcast across all conversations; when set to None, targets only non-embed connections; when set to a UUID string, targets only that specific embed session. All sends are enqueued into per-connection send_queues and dispatched by a dedicated sender coroutine 5.

3. Session type isolation#

WebSocketAdapter holds two WebSocketSession objects — websocket_person_session and websocket_group_session 6. Each session maintains per-pipeline message lists and streaming indexes 7. Per-pipeline message lists are keyed by pipeline_uuid:session_id (if session_id exists) or just pipeline_uuid (for non-embed connections). This applies to both message_lists and stream_message_indexes dictionaries, so history from one pipeline or browser session cannot collide with another.

On the client side, WebSocketClient ignores any response or user_message frame whose session_type doesn't match the connection's own session type 8.

4. Browser session isolation#

For embed widgets, a cryptographically random UUID v4 (session_id) is generated per browser tab and persisted in sessionStorage across refreshes. This identifier is validated server-side using the is_valid_session_id() function, which accepts only canonical UUID v4 strings. Embed WebSocket events use launcher IDs formatted as websocket_{pipeline_uuid}:{session_id} for person sessions and websocketgroup_{pipeline_uuid}:{session_id} for group sessions. Non-embed connections continue using the websocket_{connection_id} format. Helper methods _conversation_key(pipeline_uuid, session_id), get_connection_by_session_id(session_id, pipeline_uuid), and _parse_embed_target(target_id) support session-scoped routing and history lookups.

Entry Points#

Dashboard Debug Chat#

Route: /api/v1/pipelines/<pipeline_uuid>/ws/connect?session_type=person|group
File: websocket_chat.py

Always runs under the websocket_proxy_bot singleton adapter. The owner_bot parameter is deliberately not passed so debug requests are never attributed to a coincidentally-bound embed bot 9.

Embed Widget Chat#

Route: /api/v1/embed/<bot_uuid>/ws/connect?session_type=person|group&session_id=<uuid>

Uses bot_uuid to resolve a web_page_bot RuntimeBot. The owner_bot is passed to the message handler so replies are attributed to the correct bot identity. The embed WebSocket route requires a session_id query parameter containing a valid UUID v4. The get_embed_messages and reset_embed_session HTTP endpoints also require this parameter.

Connection Lifecycle#

1. Client connects → add_connection() registers it in all three indexes under asyncio.Lock 10, optionally storing the session_id for embed connections.
2. Two coroutines are spawned per connection: _handle_receive (reads from WebSocket, dispatches to handle_websocket_message) and _handle_send (drains send_queue to wire) 11.
3. Incoming messages create FriendMessage or GroupMessage events. For embed connections, the sender ID is formatted as websocket_{pipeline_uuid}:{session_id} (person) or websocketgroup_{pipeline_uuid}:{session_id} (group); non-embed connections use websocket_{connection_id} 12. Events then fire listeners asynchronously via asyncio.create_task 13.
4. Replies broadcast back through broadcast_to_pipeline → send_queue → _handle_send.
5. On disconnect (or error), remove_connection() removes the connection from all indexes and marks it inactive 14.

Heartbeat: client sends ping every 30 s 15; server responds with pong 16. Auto-reconnect retries up to 5 times with exponential backoff 17.

Known Issue: Cross-Pipeline Leak (Issue #2286)#

The dashboard Debug Chat path had a concurrency bug: handle_websocket_message wrote the current pipeline_uuid onto the shared websocket_proxy_bot.bot_entity.use_pipeline_uuid singleton field 18, and reply_message/reply_message_chunk re-read it from the same singleton . Under concurrent requests to two different pipelines, pipeline B could overwrite this field while pipeline A was still streaming, causing A's responses to broadcast to B's connections 21.

The session-based routing architecture resolves this issue for embedded chat by using stable pipeline/session launcher identifiers. The singleton use_pipeline_uuid field is no longer the primary mechanism for routing in embed contexts. Dashboard connections continue to use the legacy singleton behavior.

Key Files#