WebSocketCallbackClient.stop() does not close a silent WebSocket and leaves its worker thread running

Summary

In openhands-sdk==1.43.1, WebSocketCallbackClient.stop() does not reliably stop its worker thread when the WebSocket connection is open but silent.

stop() sets _stop and waits for five seconds, but the worker can remain blocked inside:

async for message in ws:

Because setting a threading.Event does not wake the pending WebSocket receive, stop() times out. It then clears self._thread even though the underlying thread is still alive.

This is observable when repeatedly creating and closing RemoteConversation instances: WebSocket worker threads and connections can accumulate after the conversations have been closed.

Environment

• Python: 3.13.4
• openhands-sdk: 1.43.1
• openhands-agent-server: 1.43.1
• websockets: 16.1.1

Relevant implementation

The current stop() implementation is:

def stop(self) -> None:
    if not self._thread:
        return
    self._stop.set()
    self._thread.join(timeout=5)
    self._thread = None

The worker remains blocked in _client_loop() while no message or close frame is received:

async with websockets.connect(ws_url) as ws:
    async for message in ws:
        if self._stop.is_set():
            break
        ...

The _stop check can only run after async for yields a message.

Minimal reproduction

The following test starts a silent local WebSocket server, connects an official WebSocketCallbackClient, and calls stop():

import asyncio
import threading
import time

import websockets
from openhands.sdk.conversation.impl.remote_conversation import (
    WebSocketCallbackClient,
)

server_ready = threading.Event()
client_connected = threading.Event()
state = {}


async def handler(websocket):
    client_connected.set()
    await websocket.wait_closed()


async def serve():
    stop_event = asyncio.Event()
    state["loop"] = asyncio.get_running_loop()
    state["stop_event"] = stop_event

    async with websockets.serve(handler, "127.0.0.1", 0) as server:
        state["port"] = server.sockets[0].getsockname()[1]
        server_ready.set()
        await stop_event.wait()


server_thread = threading.Thread(
    target=lambda: asyncio.run(serve()),
    daemon=True,
)
server_thread.start()
assert server_ready.wait(5)

client = WebSocketCallbackClient(
    host=f"http://127.0.0.1:{state['port']}",
    conversation_id="shutdown-reproduction",
    callback=lambda event: None,
)
client.start()

worker_thread = client._thread
assert worker_thread is not None
assert client_connected.wait(5)

started = time.monotonic()
client.stop()
elapsed = time.monotonic() - started

print(f"stop elapsed: {elapsed:.3f}s")
print(f"worker alive: {worker_thread.is_alive()}")
print(f"client thread reference: {client._thread}")

state["loop"].call_soon_threadsafe(state["stop_event"].set)
server_thread.join(5)

Observed output:

stop elapsed: 5.005s
worker alive: True
client thread reference: None

If the server subsequently closes the peer connection, the original worker thread exits. This confirms that the stop flag was set, but the blocked receive was not awakened.

Expected behavior

After WebSocketCallbackClient.stop() returns:

1. The active WebSocket connection has been closed.
2. The worker thread has exited.
3. self._thread is cleared only after the thread is no longer alive.
4. Repeated start/stop cycles do not accumulate worker threads or sockets.

Suggested direction

The WebSocket receive task should be actively awakened during shutdown. One possible approach is to retain the worker event loop and its main asyncio task, then cancel that task from stop() using loop.call_soon_threadsafe(task.cancel).

Cancelling the task unwinds the websockets.connect() async context manager, which closes the WebSocket. stop() can then join the worker and clear self._thread only after confirming that the thread exited.

Alternatively, the implementation could retain the active WebSocket protocol object and close it through its owning event loop.

It would also be useful to add regression coverage for:

• stopping an established but silent WebSocket;
• calling stop() immediately after start();
• repeated connect/stop cycles with no worker-thread growth;
• preserving the thread reference when shutdown times out.

Start in openhands/sdk/conversation/impl/remote_conversation.py at WebSocketCallbackClient._client_loop() and stop(); reproduce the silent-server case described in the issue. Add regression coverage for stopping an established silent WebSocket and verify that stop() returns with the connection closed, worker thread exited, and no thread growth across repeated cycles.