fix: follow pysher thread lifecycle in wait_forever - #4829

…n handler rejections (#4903)

This PR:
- supersedes #4814 and
#4813, combining both into
one coherent subscription lifecycle and callback error boundary
- closes #4812
- fixes the TypeScript half of
#4810 (the Python half
stays open; see Limitations)
- gives each subscription period (first `subscribe()` → next
`unsubscribe()`) its own pusher-js client: concurrent `subscribe()`
calls share one connection attempt, and `unsubscribe()` aborts setup
still in flight, unbinds the SDK's handlers and disconnects
- routes trigger handlers and `onSubscriptionError` through one
`invokeUserCallback()` boundary that contains synchronous throws and
rejected promises and logs a printable description (`describeThrown()`)
- keeps the containment of async `pusher:subscription_error` failures
from #4448, and also closes a
remaining gap there: a rejection whose value `String()` cannot convert
used to throw inside the `.catch` and escape again
- adds a patch changeset for `@composio/core` / `@composio/slim`

Designated PR reviewer: [@jkomyno](mailto:[email protected]).

## Root causes

These are two independent defects.

**Connection never closed
(#4812
`PusherService.unsubscribe()` unsubscribed the channel but nothing
called `disconnect()`, so the pusher-js websocket kept Node's event loop
alive. It also called `getPusherClient()`, so an `unsubscribe()` before
any `subscribe()` fetched credentials and opened a connection.
`getPusherClient()` checked `this.pusherClient` before two awaits
(credentials, `import('pusher-js')`), so concurrent first `subscribe()`
calls each created a client and the last assignment orphaned the others,
which `unsubscribe()` could never reach.

Reusing one client across periods (the alternative discussed in
#4812) is unsafe with
pusher-js 8.6.0:
- `disconnect()` keeps every channel in the client's table, with its
bound handlers, and the next `connect()` resubscribes all of them
(`subscribeAll()`).
- `unsubscribe()` on a channel whose auth is still pending only sets
`subscriptionCancelled`; the channel stays, and a later `subscribe()` of
the same name reinstates it with the old handlers. The harness below
shows this on `next`: a handler removed by `unsubscribe()` still
received the next event.
- A channel-auth reply requested for the old socket can arrive after a
reconnect and be sent on the new socket.

A fresh client per period isolates all three: late replies and cancelled
channels belong to a disconnected client whose `send_event` is a no-op,
and its handlers are unbound.

**Async handler rejections
(#4810
`Triggers.subscribe()` and `PusherService`'s `safeCallback` called the
handler inside a synchronous `try/catch`. An `async` handler returns a
promise, so its rejection skipped the `catch`, nothing handled it, and
Node's default `unhandledRejection` behavior ended the process.
Separately, the logger `JSON.stringify`s object arguments, so a thrown
`Error` was logged as `{}`.

## Ownership and cancellation model

- `PusherService` holds at most one `attempt` (an `AbortController` plus
the shared `ready` promise) and, once created, the `connection` (client,
channel, channel name). Both are cleared together by `unsubscribe()`.
- `openConnection()` checks `signal.aborted` after its last await; the
client creation and channel subscription after it are synchronous, so
`unsubscribe()` cannot interleave with them.
- A `subscribe()` whose attempt was aborted resolves without
subscribing, including when the aborted setup later fails (for example,
the credentials request rejects). A `subscribe()` that resumes after the
client was created but after `unsubscribe()` ran installs no handlers.
- A failed (not aborted) attempt is forgotten, so the next `subscribe()`
retries.
- `unsubscribe()` ends every subscription of the `Triggers` instance, as
documented ("Unsubscribe from all the triggers"), so it always closes
the connection; there is no other channel on that client to keep it
open.

## Verification

Focused regression tests
(`ts/packages/core/test/services/pusher.test.ts`,
`ts/packages/core/test/models/triggers.test.ts`), using a fake pusher-js
that keeps its binding semantics. 15 of them fail on `next` source and
pass with this change:
- disconnects and unbinds on unsubscribe; unsubscribe first creates no
client and fetches no credentials
- concurrent subscribes share one client and credentials request
- unsubscribe while credentials are pending, and when that pending
request then fails
- unsubscribe after the client exists but before the waiting
`subscribe()` resumes
- unsubscribe interleaved with pending setup keeps only the later
subscription
- resubscribe after cleanup uses a new client; late events and auth
errors on the old client reach nothing
- five cycles leave every client disconnected and unbound; retry after a
credentials failure
- sync throw and async rejection from trigger handlers, non-`Error` and
unprintable rejection values, and an unprintable rejection from
`onSubscriptionError`

Built SDK against unmodified `pusher-js` 8.6.0 and a local
Pusher-protocol (v7) server over TLS, which validates channel signatures
against the socket they arrive on. A preload only redirects the TLS
connection for the fake cluster host to localhost. Node 24.17.0, one
process per scenario; "exits" means the process ended on its own after
the scenario finished:

| Scenario | `next` | This PR |
| --- | --- | --- |
| subscribe → event → unsubscribe | still running, socket open (killed
after 5 s) | exits in ~6 ms, 0 open sockets |
| unsubscribe before subscribe | 1 credentials request, 1 client, still
running | 0 requests, 0 clients, exits |
| unsubscribe → resubscribe → event | both receive, still running | both
receive on separate connections, exits |
| two concurrent subscribes | 2 credentials requests, 2 sockets, both
left open | 1 request, 1 socket, each handler once, exits |
| unsubscribe during initialization | event delivered after unsubscribe,
2 sockets open | no socket opened, nothing delivered, exits |
| async handler rejects | process exits with code 1 (unhandled
rejection) | logs `❌ Error in trigger callback: <message>`, keeps
running, exits 0 |
| auth reply delayed past unsubscribe → resubscribe | unsubscribed
handler still receives the event | unsubscribed handler receives
nothing, 0 stale or invalid subscribes, exits |
| 100 subscribe/unsubscribe cycles | 1 client, never exits | 100 clients
created, 0 connected, 0 open sockets, exits |

Commands:
- `pnpm --filter @composio/core test`: 73 files, 1842 passed, 2 expected
failures
- `pnpm --filter @composio/core typecheck`
- `oxlint` on the changed files: no new warnings (one existing warning
in `Triggers.ts`)
- `pnpm validate:changesets`

## Limitations

- pusher-js 8.6.0 never releases a client: the constructor pushes it
onto the static `Pusher.instances` array, and its `ConnectionManager`
binds `online`/`offline` listeners on the module-level `Network` object
that `disconnect()` does not remove, so each disconnected client stays
reachable. Neither has a documented release API, and the listeners can't
be told apart from other clients' listeners. With pusher-js alone, each
connect/subscribe/disconnect cycle retained about 14 KB. Through the
SDK, the per-cycle growth was about 14.6 KB over 300 cycles, which is
mostly that client. The SDK unbinds its own handlers, so retained
clients don't keep user callbacks alive. A script that subscribes once
is unaffected; an app that cycles subscribe/unsubscribe many times keeps
one disconnected client per cycle.
- Each subscription period fetches realtime credentials again.
- Python is unchanged. `TriggerSubscription.stop()` already disconnects,
and #4829 covered the
connection-thread lifecycle. The Python half of
#4810 (an `async def`
handler is never awaited) is a separate defect that needs a decision
between rejecting async handlers and running them, so this PR doesn't
close that issue.
- `triggers.subscribe()` keeps its `(data) => void` handler type; it
already accepts async functions, and widening the return type would
reject handlers that return other values.

## Credits

Thanks to @Hugo-DDT for #4814
and to @TanishGudise for
#4813 and the investigation
in #4812 and
#4810. Both are co-authors
of the commit.
- From #4814: disconnecting
on unsubscribe, the no-client-on-bare-unsubscribe rule, shared
initialization for concurrent subscribers, and treating setup errors
after cancellation as cancellation. This PR replaces the generation
counter and `allChannels()` check with an aborted attempt per period and
a fresh client per period, which also covers a `subscribe()` that
resumes after `unsubscribe()` and the cancelled-channel and late-auth
cases.
- From #4813: the
`Promise.resolve(...).catch(...)` containment and the
failure-description fallbacks, now shared by both callback paths as
`describeThrown()`. Its regression tests are included unchanged.
- The issue analysis, especially the `Pusher.instances` and `Network`
retention finding and the late-auth concern, shaped the design and the
harness above. Thanks also to @VANDRANKI for the community reviews.

Co-authored-by: DDT <[email protected]>
Co-authored-by: Tanish Gudise <[email protected]>