fix(core): prevent allBounded listener leak via PubSub subscription

[Shalin-Shah-2002]

Issue for this PR

Closes#36677

Type of change

• Bug fix
• New feature
• Refactor / code improvement
• Documentation

What does this PR do?

The root cause of #36677 is that EventV2.allBounded relied on events.listen(), which registers the callback in a global listeners Set inside the EventV2 service. The cleanup of that registration depended on Effect.addFinalizer running inside the scope provided by Stream.unwrap → Channel.unwrap. When that scope didn't close reliably (location expiry/reboot cycles, or SSE client disconnect patterns that skip channel finalization), the listener stayed in the Set permanently.

What this means for a leaked listener: every call to EventV2.notify iterates the entire listeners Set and calls each listener function. A leaked allBounded listener calls Queue.offer on every published event, allocating regardless of whether the queue is already failed. Over hours of uptime with repeated location reboots, these allocations push the JSC heap into continuous GC cycles — the 80% CPU / constant madvise behavior reported in the issue.

The fix has three parts:

1. Move allBounded out of events.listen() and into the Interface directly.

The old module-level function:

Effect.gen(function*(){constqueue=Queue.dropping(capacity)constunsubscribe=yield*events.listen(callback)// registers in global Setyield*addFinalizer(()=>unsubscribe.then(shutdown))// depends on scope closereturnStream.fromQueue(queue)})

The new Interface method inside the layer:

Effect.gen(function*(){constqueue=Queue.dropping(capacity)constsubscription=yield*PubSub.subscribe(pubsub.all)// synchronous, no raceyield*forkScoped(Stream.fromSubscription(subscription).pipe(runForEach(callback)))yield*addFinalizer(()=>Queue.shutdown(queue))returnStream.fromQueue(queue)})

The key difference: PubSub.subscribe(pubsub.all) registers the subscriber directly on the PubSub, not in a separate global Set. The subscription is managed by Scope — when the scope closes (stream ends or is interrupted), the subscriber is removed atomically as part of PubSub's internal subscriber tracking. There is no separate cleanup step that can be skipped.

PubSub.subscribe is also synchronous — it returns the subscriber immediately after registering it. This means there is no race between subscribing and publishing events, so the existing test (which creates bounded streams then immediately publishes) continues to pass.

2. Added missing Queue.shutdown in the opencode SSE handler finalizer.

At packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts:32, the finalizer was calling unsubscribe but never Queue.shutdown(queue). This meant the unbounded queue survived the SSE disconnect, holding references to buffered events.

3. Collected unsubscription effects in share-next watch helper.

At packages/opencode/src/share/share-next.ts, the watch function called events.listen() but never stored or ran the returned Unsubscribe effect. The listeners accumulated for the process lifetime. The fix stores each unsubscribe and registers them as a scope finalizer.

How did you verify your code works?

• All 44 event tests pass, specifically including ends only an overflowing bounded subscriber without blocking other listeners which tests the exact bounded-stream pattern used in production.
• Full core test suite: 1080 tests pass, 0 fail across 142 files.
• TypeScript typecheck passes in packages/core and packages/server.
• The monorepo-wide bun turbo typecheck passes across all 36 packages.

Screenshots / recordings

Not applicable — logic change only.

Checklist

• I have tested my changes locally
• I have not included unrelated changes in this PR

[github-actions]

Thanks for updating your PR! It now meets our contributing guidelines. 👍

[Shalin-Shah-2002]

This PR is ready for review. All tests pass (1080/1080), typecheck is clean across all 36 packages, and the fix directly addresses the root cause of the allocation loop described in #36677.