[Bug]: Trigger subscription stop() from a callback deadlocks the process; subscribe() timeout leaksa forever-reconnecting websocket thread

[AdityaPainuli]

SDK Language

Python SDK (composio package)

SDK Version

composio==0.17.1 (reproduced on current next, pysher==1.0.8)

Runtime Environment

Python 3.12 on macOS (darwin 25.4)

Environment

Local Development

Describe the Bug

Two related lifecylce bugs in the realtime trigger code (python/composio/core/models/triggers.py). Same ~60 lines, so filing them together. Found by reading the pusher paths; both confirmed with offline, determinictic repros.

Bug 1 — stop() called from a trigger callback deadlocks the whole process.

stop() (triggers.py:781) calls self._connection.disconnect() before setting _alive = False. pysher's Connection.disconnect(timeout=None) does self.socket.close(); self.join(timeout) — an unbounded join of the websocket thread. But callbacks run on that thread's dispatch path: _handle_event (triggers.py:752) submits callbacks to a ThreadPoolExecutor and blocks on future.result() until they finish. So the natural "got my event, stop listening" pattern deadlocks:

• worker thread: stop() → disconnect() → join() — waits for the pysher thread
• pysher thread: blocked in _handle_event — waits for that worker

_alive = False is never reached, so a main thread parked in wait_forever() never exits, and the non-daemon executor worker prevents interpreter shutdown. This is effectively the only way to stop a subscription in the documented usage — the main thread sits in wait_forever() (README quickstart, python/examples/triggers.py), so stop() has to come from a callback.

Expected: stop() returns, wait_forever() unblocks, process exits.

Bug 2 — _SubcriptionBuilder.connect() timeout raises without disconnecting.

triggers.py:865-878: pusher.connect() starts pysher's connection thread, and the deadline path raises ComposioSDKTimeoutError without ever calling pusher.disconnect(). pysher's run loop (while self.needs_reconnect and not self.disconnect_called) then redials every reconnect_interval (10s) for the life of the process. Every timed-out subscribe() leaks one thread; retrying subscribe() (the natural response to a timeout) accumulates one per attempt. If a leaked connection later succeeds, it authenticates and subscribes the channel of an abandoned TriggerSubscription, silently consuming trigger events.

Expected: the timeout path tears down the connection it started before raising.

Steps to Reproduce

Bug 1:

1. Create a TriggerSubscription and register a callback that calls subscription.stop()
2. Deliver one trigger frame through _handle_event from the connection thread (what pysher does on a live event)
3. Observe: callback never returns, connection thread never exits, _alive stays True, process cannot exit

Bug 2:

1. Start a pysher connection to an unreachable endpoint (the state connect() creates)
2. Let the SDK's deadline elapse — ComposioSDKTimeoutError is raised, no disconnect() on that path
3. Observe: connection thread still alive, needs_reconnect=True, disconnect_called=False - redials forever

Minimal Reproducible Example

# Bug 1 — real TriggerSubscription, real _handle_event/stop(), pysher's real
# inherited disconnect(); only the socket is faked (thread delivers one frame).
import json, os, threading
import pysher.connection
from composio.client import HttpClient
from composio.core.models.triggers import TriggerSubscription

os.environ.setdefault("COMPOSIO_API_KEY", "ck_test_dummy")

V3_EVENT = json.dumps({
    "type": "composio.trigger.message", "id": "evt_1", "data": {},
    "metadata": {"trigger_id": "trg_1", "trigger_slug": "GMAIL_NEW_MESSAGE",
                 "user_id": "u", "connected_account_id": "ca", "auth_config_id": "ac"},
})

class FakeWireConnection(pysher.connection.Connection):
    def __init__(self, subscription):
        super().__init__(event_handler=lambda *a, **k: None,
                         url="ws://127.0.0.1:1/app/x", daemon=True)
        self._subscription = subscription
    def run(self):
        self._subscription._handle_event(V3_EVENT)

client = HttpClient(api_key="ck_test_dummy", provider="openai")
sub = TriggerSubscription(client=client)
stopped = threading.Event()

@sub.handle()
def on_event(event):
    sub.stop()
    stopped.set()

conn = FakeWireConnection(sub)
sub._connection = conn
sub._alive = True
conn.start()

print("callback returned:", stopped.wait(timeout=6))   # False
print("pysher thread alive:", conn.is_alive())          # True
print("subscription._alive:", sub._alive)               # True — never cleared


# Bug 2 — the exact state the SDK leaves behind on the timeout path.
import time

conn2 = pysher.connection.Connection(
    event_handler=lambda *a, **k: None,
    url="ws://127.0.0.1:1/app/x",  # unreachable
    reconnect_interval=1, daemon=True,
)
conn2.start()
time.sleep(3.5)  # SDK's connect() would have raised by now; no disconnect()

print(conn2.is_alive())                                       # True
print(conn2.needs_reconnect and not conn2.disconnect_called)  # True — redials forever

Error Output / Stack Trace

No exception — that is the failure mode. Observed output of the repro:

BUG 2 — connect() timeout path (no disconnect before raise):
  connection thread alive after failure: True
  still scheduled to reconnect forever:  True
[ERROR] [Errno 61] Connection refused - goodbye   (repeats every reconnect_interval, indefinitely)

BUG 1 — stop() from callback:
  callback returned within 6s: False
  pysher thread still alive:   True
  subscription._alive:         True (never set False)

The Bug 1 repro process never exits on its own (deadlocked non-daemon worker); it has to be killed.

Reproducibility

• Always reproducible
• Intermittent / Sometimes
• Happened once, can’t reproduce

Additional Context or Screenshots

No response

[AdityaPainuli]

Alberto Schiabel (@jkomyno) working on the fix locally, will raise a PR soon!