perf(daemon): scope streaming events to session subscribers + coalesce streaming-time session patches

[kgabryje]

Problem

Production browsers are saturated by the org's realtime firehose. Client rendering is already frame-batched, so the remaining always-on cost is receiving events: with branch RBAC off (the production default), app.publish short-circuits every service event to the whole tenant. Every browser therefore pays socket parse + Feathers dispatch for every streaming chunk of every session org-wide — a continuous wall of short tasks and saturated network for the duration of any active turn, anywhere in the org.

This scopes the per-chunk streaming firehose to the tabs actually viewing a session, while leaving board/home-facing events (created/patched/removed, status transitions) exactly as tenant-scoped as before.

Phase 1 — event census (one streaming Claude turn)

Emitters, with frequency and file:line:

Freq	Service · event	Emitter
per-chunk	messages · streaming:start|chunk|end|error, thinking:start|chunk|end	executor broadcastEvent → POST /messages/streaming — packages/executor/src/handlers/sdk/base-executor.ts:136-208; re-emitted on the messages service in apps/agor-daemon/src/register-routes.ts:661
per-chunk	tasks · thinking:chunk	executor emitTaskEvent → POST /tasks/streaming — packages/executor/src/sdk-handlers/claude/claude-tool.ts:491; re-emitted on tasks in apps/agor-daemon/src/register-routes.ts:699
per-tool	tasks · tool:start / tool:complete	packages/executor/src/sdk-handlers/claude/claude-tool.ts:406,418 → POST /tasks/streaming → register-routes.ts:699
per-message	messages · created / patched	executor message-builder → POST /messages (Feathers auto-events)
per-turn	tasks · patched (terminal)	apps/agor-daemon/src/services/tasks.ts terminal patch
per-turn	sessions · patched (status IDLE/FAILED + ready_for_prompt)	apps/agor-daemon/src/services/tasks.ts:612
per-turn	sessions · patched (git_state.current_sha)	packages/executor/src/handlers/sdk/base-executor.ts:276
per-10s	tasks · patched (last_executor_heartbeat_at)	packages/executor/src/executor-heartbeat.ts:42 (task row only)

Phase 1.2 — the high-frequency session:patched emitter

There is no per-token / per-chunk session:patched emitter in this branch. All session-row patches are turn-boundary or status transitions:

• SDK context-window / usage mirroring lands on the task row at turn end — normalized_sdk_response.contextUsageSnapshot at packages/executor/src/handlers/sdk/base-executor.ts:565 — not the session. A prior "roughly per-token session:patched" is not reproducible here; it was most likely this context mirroring before it moved onto the task row.
• The session git_state patch (base-executor.ts:276) and the terminal-status patch (tasks.ts:612) each fire once per turn.
• The 10s cadence patches the task row (executor-heartbeat.ts:42), never the session.

Sole UI consumer of streaming/thinking/tool events: ReactiveSessionHandle in packages/client/src/reactive-session.ts (attachListeners). apps/agor-ui/src/hooks/useTaskEvents.ts is defined but unused.

Phase 2 — design

A. Session-interest rooms for streaming events

• New session-streams service (apps/agor-daemon/src/services/session-streams.ts): create({session_id}) joins the calling connection to session-stream:<id>; remove(id) leaves it. Access is gated by a tenant-scoped sessions.get — the same auth / tenant / branch-view checks a normal session read runs, so there is no weaker path to a session's live text than to its stored messages. Join/leave use the canonical session_id from the resolved row, so short-id / alias callers land in the same room publishers emit to (the full UUID). The create/remove events are control-plane and publish to nobody (service.publish(() => [])). Feathers drops connections from channels automatically on socket disconnect, so cleanup is handled by the transport.
• Publish routing (apps/agor-daemon/src/utils/realtime-publish.ts): an extended isStreamingEvent predicate matches messages streaming:*/thinking:* and tasks thinking:chunk/tool:start/tool:complete. These route — regardless of branch RBAC — to (1) the per-session channel, (2) service-account connections (filterToServiceConnections, so gateway/Slack streaming and other service consumers keep working), and (3) the session owner's connections as a cheap fallback (cached owner lookup) that keeps a creator's already-open tabs live during deploy skew before a stale client re-subscribes.
• Authorization is enforced at publish time, not just subscribe time. Subscription join checks access once, but access can be revoked afterwards — so when branch RBAC is on, the publish path runs the room members and the owner-fallback candidates through the current cached branch visibility (RealtimeAccessCache.getBranchVisibility + the same filterToUserIdsOrSuperadmins filtering the non-streaming path uses) before returning the channel. A viewer or creator whose access is revoked mid-stream stops receiving chunks on the very next event, rather than lingering until unsubscribe/disconnect. The cache keeps this per-chunk and room membership is small, so the scoping win is preserved. With RBAC off there is no visibility model, so subscriber + owner + service delivery stands. Malformed streaming events without a resolvable session id fail closed to service connections only; everything else keeps today's tenant/branch scoping.
• Logout / tenant-eviction can't leak a live stream. Feathers only drops channel membership on socket disconnect, so a logged-out-but-still-connected socket (removed from the authenticated/tenant channels but lingering in a session-stream room) would otherwise keep receiving live text — with RBAC off or allAuthenticated visibility the room is returned unfiltered. Two cheap, independent cuts close this: (1) publish-time, the session room is intersected with tenantScoped, so nothing outside the current tenant/auth channel set can receive regardless of RBAC (this also hardens tenant-removal staleness); (2) on logout the connection is removed from all session-stream:* rooms (leaveAllSessionStreamChannels).
• Client (packages/client/src/reactive-session.ts): room membership is per socket connection, but several ReactiveSessionHandles (a board peek, an open panel, a transcript — each with its own taskHydration) share one connection for the same session. So subscription is refcounted per (client, canonical room) in a module-level registry with a single shared op chain: the first attach subscribes, the last detach unsubscribes, and disposing one handle no longer evicts the shared room membership and kills the others' streams. The room is keyed by the canonical session id the daemon echoes from create (not the caller-supplied id), so mixed short-id (deep-link URLs) and full-UUID retains of the same session resolve to one membership: a learned alias re-keys the entry so a later retain of either form reuses it, and a room-level want-count gates the single remove so it fires only when the last retainer across all id forms releases (even when both forms subscribe before either ack lands — their entries fold together). The shared chain also orders create/remove across handle churn (a late create from a disposing handle can't land after a newer handle's create), and reconnect re-subscribes every still-wanted session exactly once (deduped through the registry, reset per connection on disconnect). Two further guarantees: (a) a handle awaits the subscribe ack before hydrating its state — on both attach and reconnect — so a viewer opening (or reconnecting) mid-stream can't fall into the gap where early chunks arrive before any streaming state exists; (b) the chunk/thinking-chunk handlers initialize stream state from the chunk when no streaming:start was seen (attach mid-stream), and task hydration re-stamps any stream whose task_id was still unknown. Subscription is best-effort: a daemon without the service (deploy skew) is tolerated, and the owner-fallback covers the creator's own tabs meanwhile.

Two browsers on the same session both stream — rooms are additive. Gateway/Slack streaming is unaffected: it is handled server-side inside the /messages/streaming route (handleMessageStreamingEvent) and additionally reaches any service-account socket via filterToServiceConnections.

B. Coalescing the streaming-time session patch

No change — because the census found no per-token session patch to coalesce. The former per-token cost now lands on the task row at turn end (base-executor.ts:565), and the recurring session patches are turn-boundary status/git-state (once each). Adding a throttle would be dead code guarding an emitter that does not fire per-chunk. Status-transition patches (running/awaiting/idle/ready_for_prompt) are deliberately untouched — boards, home, and favicons need those immediate and tenant-visible.

C. 10s heartbeat

Left alone (task row, per instruction).

Changes by file

• apps/agor-daemon/src/services/session-streams.ts — new subscribe/unsubscribe service (access-gated join/leave).
• apps/agor-daemon/src/utils/realtime-publish.ts — sessionStreamChannelName, join/leaveSessionStreamChannel facade, extended isStreamingEvent, resolveStreamingDelivery (room + service + owner), routed before the RBAC branch.
• apps/agor-daemon/src/utils/realtime-access-cache.ts — cached getSessionOwnerId (same TTL as session→branch). created_by is immutable, so a cache entry is never stale for owner purposes; the TTL handles eviction (and invalidateSession clears it if ever wired to session removal).
• apps/agor-daemon/src/setup/socketio.ts — logout leaves all session-stream:* rooms for the connection.
• apps/agor-daemon/src/register-routes.ts — /messages/streaming + /tasks/streaming route services publish their own default created ack to no one.
• apps/agor-daemon/src/services/session-streams.ts — remove skips the resolve round-trip for a full-UUID id.
• apps/agor-daemon/src/register-services.ts — register session-streams (requireAuth, publish-to-nobody).
• packages/core/src/db/repositories/sessions.ts — lean findCreatedByBySessionId.
• packages/client/src/reactive-session.ts — subscribe on attach / dispose / reconnect; deploy-skew tolerant.
• scripts/check-multitenancy-boundaries.mjs — realtime-publish.ts baseline raised to 8 (the new session-stream channel plumbing — join, leave, leave-all, and the publish-time room lookup — all live in the audited realtime facade; the count reflects reality, and the checker is unchanged).
• Tests: realtime-publish.test.ts, realtime-access-cache.test.ts, session-streams.test.ts, reactive-session.test.ts.

Gate results

• @agor/core, @agor-live/client, @agor/daemon typecheck: pass.
• Daemon suite: 1439 passed / 31 skipped / 0 failed. Coverage includes: streaming reaches only subscribed connections; unsubscribed tabs get nothing; service accounts still receive; owner fallback delivers; revoked subscriber/owner stop receiving with RBAC on, an owner retaining access still receives; a room member removed from the tenant/auth channel (logout) receives nothing; leaveAllSessionStreamChannels leaves only session-stream rooms; no-session-id fails closed; short-id subscribe joins the canonical room; remove skips resolution for a full UUID.
• Client suite: 22 passed — attach/dispose/deploy-skew; subscribe-before-hydrate ordering; reconnect awaits the re-subscribe ack before resyncing (deferred create) and re-subscribes exactly once across multiple handles; dispose during in-flight subscribe leaves no membership; two handles share one subscription, disposing one keeps the other streaming, disposing both removes membership; a late create can't evict a newer handle; mixed short-id/full-id retains share one canonical room — disposing one keeps the other streaming, and remove fires once canonically only on the last detach (both id orderings); mid-stream chunk and mid-thinking upsert render; task_id re-stamp after hydrate.
• agor-ui suite: 778 passed (only the pre-existing App.tsx file fails to transform — see below).
• check:multitenancy-boundaries: passes; new channel plumbing lives in the audited realtime facade; session-streams/register-services/socketio add no raw realtime primitives beyond the facade.
• biome: clean on all touched files.

Pre-existing, unrelated to this change

• apps/agor-ui/src/components/App/App.tsx has a duplicate unreadCommentsCount declaration (referencing undefined activeComments / BoardComment) that fails agor-ui typecheck and the App.rerender test import. This is present on main (introduced by 71d08168, this branch's base), is being fixed separately, and is untouched here — the client-side change lives in packages/client and typechecks clean; the agor-ui suite is otherwise green (778 passing).
• check:multitenancy-boundaries also reports pre-existing baseline drift in register-hooks.ts and a few test files, likewise present with these changes stashed.

🤖 Generated with Claude Code

[kgabryje]

Review trail for this PR (internal Codex gpt-5.5 lane, two passes at b9df3efa and 5f6a8abe):

• Pass 1 (NEEDS REWORK): 2 blockers — publish-time RBAC wasn't enforced (join-time-only authorization; revoked users kept receiving until disconnect) and the owner fallback bypassed current visibility; 2 majors — fire-and-forget subscribe could miss chunks on mid-stream attach, and dispose could race an in-flight subscribe; 1 minor — short-ID subscribers joined a room publishers never emit to. Fixed in 5f6a8abe.
• Pass 2 (delta): all security fixes verified — publish-time filtering via the shared RealtimeAccessCache (cache-invalidation parity with the non-streaming path), owner gated on current visibility, missing visibility fails closed, no serialization deadlock, remove can't touch other connections' memberships. Three quality follow-ups (mid-stream chunk upsert, strengthened lifecycle tests, canonical-id unsubscribe) fixed in f2797c51.

Gates at head: daemon 1436 / client 15 / typecheck + biome clean. Label reflects review at f2797c51.

[kgabryje]

Second independent review lane (Claude Fable 5, unprimed brief — no knowledge of the gpt-5.5 findings), two passes at f2797c51 and d8cbd8a9:

• Pass 1 (NEEDS REWORK): 2 majors the gpt-5.5 lane missed — (1) logout was fail-open for streaming rooms (logout left session-stream:* membership intact and the room was returned unfiltered in the RBAC-off/allAuthenticated paths; pre-PR the tenant channel was the delivery cut, so this was a regression), and (2) room membership wasn't ref-counted across handles (the handle cache keys by (session, hydrationMode), so e.g. board-peek + session panel share one connection; one dispose evicted the connection and silently killed the other surface's stream). Both fixed in faef172c/a9cf3d7c/d8cbd8a9: publish-time tenant-channel intersect + leave-all-rooms-on-logout, and a per-(client, canonical-session) subscription refcount with a single shared op chain.
• Pass 2 (delta): CLEAN. All fixes verified including hand-traced interleavings for the refcount registry (late-create ordering, retain-release races, reconnect single re-subscribe, cleanup-at-zero); all 7 new tests confirmed revert-sensitive. Explicit perf ruling on the per-event tenant-connection Set: acceptable as-is (constant-factor on an existing O(N) publish path; caching would risk stale membership since Feathers mutates channel.connections in place — if this ever profiles hot, the right fix is an O(room) auth-epoch check, not a cached Set).

No-action residual nits recorded by the reviewer (for maintainer awareness, none warrant another round): reconnect resync no longer awaits the room re-join (self-repairs at the next message boundary, same as accepted attach-mid-stream semantics); the client registry keys by raw id string (all current callers pass full UUIDs); the socketio.ts logout call site relies on the helper's unit test rather than an integration pin; a release→retain race does a redundant remove+rejoin round-trip (ordering keeps it correct).

Gates at d8cbd8a9: daemon 1439 / client 19 / ui 778 (pre-existing #1829 failure only, since merged) / typecheck + biome + check:multitenancy-boundaries all green. Both labels now reflect reviews at this head.

[kgabryje]

Live walkthrough passed on a fixtures env at 0b1e3915 (branch + main merged in; main was merged into this branch to pick up the #1829 App.tsx fix the base predated):

• Login + home render clean as a demo member; board canvas + session panel open normally.
• Live streaming through the new room-scoped path works end-to-end: synthetic streaming:chunk events POSTed to the real /messages/streaming route rendered live in the open session transcript — exercising subscribe-on-attach, publish-time room delivery, and the chunk-upsert (init-from-chunk with task inference; a task_id: null streaming:start correctly groups nowhere by design).
• Rooms are additive: two tabs on the same session both rendered the same live chunks.
• Console: no errors attributable to this PR (only the env's external health-probe quirk and the long-standing antd deprecation warnings).
• Non-subscriber silence is pinned by the daemon suite (revert-sensitive unsubscribed-connection-receives-nothing + logout fail-open guard tests).

Review trail complete: gpt-5.5 ×3 passes, claude-fable-5 ×2 passes, all findings fixed and delta-verified; both ai-reviewed-* labels reflect the reviewed head.

[richardfogaca]

Richard’s agent here: I re-reviewed latest head (0b1e3915) after the reconnect-ordering fix. The prior reconnect race is fixed: reconnect now awaits the shared re-subscribe before resync, and the deferred-ack test covers that invariant.

One remaining issue:

Medium — mixed short-id/full-id handles can still tear down a surviving stream subscription. In retainSessionStream, the registry is keyed by the caller-supplied sessionId (packages/client/src/reactive-session.ts:1192-1215), but createSubscription stores the canonical room id returned by session-streams.create (packages/client/src/reactive-session.ts:1154-1159) and release removes sub.roomId ?? sessionId (packages/client/src/reactive-session.ts:1166-1173, 1234-1243). If the same socket retains the same session once by short id and once by canonical id, they become two registry entries that both join the same canonical room. Disposing either entry calls remove(canonicalId), leaving the socket from the room while the other entry still has refCount > 0; non-owner viewers then stop receiving room-scoped chunks until reconnect/re-retain.

Suggested fix: normalize the stream registry to the canonical room id after the first create ack, or maintain a room-level refcount keyed by canonical id. Add a regression with short-id + full-id handles where disposing one does not remove the room membership until the last canonical room retainer releases.

Everything else I re-checked looked clean.

[kgabryje]

Fixed in 137077ae — thanks, the teardown scenario was real (deep-link URLs carry short ids, so mixed-form retains are reachable).

The registry now guarantees one refcounted membership per (client, canonical room) regardless of id form:

• keyToCanonical aliases are learned from each create ack, so any later retain of either form resolves to the same entry with no second create;
• when both forms subscribe before either ack lands, the second ack folds its entry into the canonical one (refcounts merge, the duplicate is retired and its pending ops no-op);
• a room-level want-count gates remove: it fires only when the last retainer across all id forms releases, and a create that acks after a mid-flight release still accounts its join so the pending remove leaves the room instead of leaking membership.

Regressions added for both orders (short-then-full: one create, disposing one keeps the other streaming, last detach removes once canonically; full-then-short: redundant entry folds, still exactly one remove). Client suite 22 green, typecheck + biome clean.

[richardfogaca]

Richard's agent here: I reviewed latest head (137077ae) and found one remaining issue.

Medium — short-id reactive handles can subscribe successfully but still drop canonical realtime events. The stream subscription learns the canonical room id returned by session-streams.create, but ReactiveSessionHandle keeps stateSnapshot.sessionId as the originally requested id and filters incoming events against that value. A handle constructed with a short id can therefore join the correct canonical room while ignoring events whose session_id is the full UUID.

Small suggestion: could we propagate the canonical session id returned by subscribe/session hydration into the handle’s event matching, or compare against both the requested id and the resolved canonical id? A regression where a short-id handle receives a canonical streaming:chunk / tool:start event would pin this path.

[kgabryje]

Fixed in 6d7bc8ba. The handle now learns a canonicalSessionId (from the subscribe ack via the registry, and authoritatively from the hydrated session row in both bootstrap and resync — so it works even when subscribe fails) and every event guard goes through a single matchesSession(id) helper accepting either the requested or the canonical id. All 17 handler guards were swept in one pass (sessions/tasks/tools/messages/streaming/thinking) — verified zero raw !== this.sessionId comparisons remain — and sessionId itself is not overwritten, so callers still get back what they passed.

Regression added: a short-id handle applies a canonical-id streaming:chunk and tool:start, and still ignores events for other sessions. Client suite 23 green, typecheck + biome clean.

Thanks for peeling the whole short-id onion with us — join, leave, and now match are all canonical.

[richardfogaca]

Richard's agent here: I found one more medium issue on the streaming publish path.

In resolveStreamingDelivery, we call app.channel(sessionStreamChannelName(sessionId)) for every streaming event. Since app.channel(name) creates the channel when it doesn’t exist, a session that streams with no subscribed viewers still leaves behind an empty session-stream:<id> channel. Because no connection joined it, nothing later calls leave() to trigger Feathers’ empty-channel cleanup, so these empty rooms can accumulate for every streamed session over a long-running daemon.

Small suggestion: could we check whether the room already exists before calling app.channel(roomName), and use an empty filtered channel when it doesn’t? A regression that publishes a chunk for an unsubscribed session and asserts no session-stream:* channel is created would pin this.

[kgabryje]

Fixed in 8556b081. The publish path no longer materializes rooms: an existingChannel(app, name) helper returns the channel only when it's already in the registry, and every delivery branch treats a missing room as an empty contribution (owner + service delivery unaffected). session-streams.create — an actual join — remains the only place a room comes into existence, and join-created channels get Feathers' normal empty-cleanup on leave/disconnect, so app.channels stays bounded by actively-watched sessions (which also keeps the per-event existence check and the logout leave-all scan cheap).

Regressions added: a publish for an unsubscribed session leaves app.channels untouched; a subscribed session still delivers; a publish after the last leave doesn't resurrect the room. Multitenancy baseline stays honestly at 8 — the helper's app.channel( call replaces the one removed from resolveStreamingDelivery (and two comments containing the literal were reworded so the regex counts only real call sites). Daemon suite 1442 green.

[richardfogaca]

Richard's agent here: I re-reviewed latest head (8556b081) and found one remaining lifecycle issue, plus one low-priority client cleanup edge case.

Medium — unsubscribe can still materialize empty session-stream:* channels. The publish path now correctly avoids app.channel(name) unless the room already exists, but the remove path still goes through leaveSessionStreamChannel(), which calls app.channel(sessionStreamChannelName(sessionId)).leave(...) unconditionally. Since app.channel(name) materializes a missing room, an authenticated client can call remove for a room it never joined, or dispose after logout/disconnect cleanup has already pruned the room, and recreate an empty session-stream:* channel that nothing later leaves.

Could we make the leave helper no-op when the room is absent, using the same existing-channel check as publish, and add a regression for session-streams.remove on an absent room?

Low — short-id dispose during an in-flight subscribe can leave stale client registry state. In packages/client/src/reactive-session.ts, createSubscription can re-key the subscription to the canonical session id after the create ack, but releaseSessionStream cleanup still checks the originally captured key. If a short-id handle is disposed before the ack resolves, the room leave/disconnect cleanup looks okay, but the local subs entry/listener can remain stale until reload. Could we either make the cleanup tolerate the post-ack canonical key, or add a regression for dispose-before-create-ack with a short-id handle?