Chat broadcast and event delivery

源码版本v2026.6.11

Responsibilities

server-chat.ts delivers events emitted by the agent main loop back to clients. It doesn't produce events; it only does dispatch by type: it takes AgentEventPayload (delta, final, tool, lifecycle, etc.) and decides — based on event type, session ownership, and client subscriptions — whether to broadcast to all visible connections, broadcastToConnIds to a targeted set of connIds, or nodeSendToSession to a process on another node that has subscribed to that session.

This layer maintains three categories of subscription state: toolEventRecipients (runId-level tool event subscribers, registered by clients before starting a run), sessionEventSubscribers (session-level, attached by operator UIs after a run is already in flight), and sessionMessageSubscribers (session-level chat message subscribers for targeted delivery when the control UI is hidden).

Design motivation

Why not just socket.send from inside the agent main loop? A single session may be subscribed to by multiple clients, clients may span multiple gateway nodes, agent events need to be split by stream type (agent / chat / session.tool / sessions.changed), some events need throttling, some need dropping if slow. Stuffing these policies into the main loop would turn the loop into a mess — the main loop should only care about the agent state machine, so broadcast policy is extracted into its own layer.

The three primitives are injected at the outer layer by createGatewayNodeSessionRuntime (createGatewayNodeSessionRuntime:971). createAgentEventHandler only consumes them and doesn't care about the implementation — the chat module can be tested without real WebSockets.

Key files

• AgentEventHandlerOptions:255-293 — three broadcast primitives + three subscriber registry type declarations.
• createAgentEventHandler signature:299-317 — injects broadcast primitives and returns the event dispatcher.
• resolveNodeSessionDeliveryKeys:495 — expands a global session into two delivery keys: agent:DEFAULT:global + global.
• sendChatPayload:890 — chat event delivery; control UI visible → broadcast, hidden → targeted.
• sendAgentPayload:967 — agent event delivery; same split.
• dispatcher closure:1136 — (evt: AgentEventPayload) => {...} event entry.
• createGatewayBroadcaster:100-209 — real implementation of broadcast / broadcastToConnIds, with scope filtering and backpressure.
• createGatewayNodeSessionRuntime:14-53 — nodeSendToSession routes via nodeSubscriptions.

Data flow

createAgentEventHandler (createAgentEventHandler:299) takes the three broadcast primitives and three subscriber registries and returns an event dispatcher (evt: AgentEventPayload) => void. Internally it maintains throttle state, lifecycle retry state, and a spawnedBy cache.

broadcast and broadcastToConnIds come from createGatewayBroadcaster (createGatewayBroadcaster:100). The implementation iterates clients: Set<GatewayWsClient> with the core loop:

for (const c of params.clients) {
  if (targetConnIds && !targetConnIds.has(c.connId)) continue;
  if (!hasEventScope(c, event)) continue;        // scope filter
  const slow = c.socket.bufferedAmount > MAX_BUFFERED_BYTES;
  if (slow && opts?.dropIfSlow) {
    if (!isTargeted) clientSeq.set(c, (clientSeq.get(c) ?? 0) + 1);  // targeted doesn't consume seq
    continue;                                                       // droppable, drop
  }
  if (slow) { c.socket.close(1008, "slow consumer"); continue; }    // non-droppable, disconnect
  c.socket.send(frame);  // {"type":"event","event":...,"payload":...,"seq":N}
}

Backpressure has two tiers: dropIfSlow: true (high-frequency chat delta, droppable streams) drops but updates seq to avoid seq misalignment; dropIfSlow: false (critical lifecycle, tool results) directly calls socket.close(1008, "slow consumer") to disconnect — better to disconnect than to lose a critical event. seq is incremented only for non-targeted broadcasts (isTargeted ? undefined : nextSeq); targeted broadcasts are filtered subsets and don't participate in client-side reordering.

nodeSendToSession comes from createGatewayNodeSessionRuntime (nodeSendToSession:31) and is a one-liner: nodeSubscriptions.sendToSession(sessionKey, event, payload, nodeSendEvent). The source comment "session fanout goes through the subscription manager so node reconnects and explicit unsubscribes keep both node->session indexes in sync" explains why cross-node delivery goes through the subscription manager rather than nodeRegistry.send directly — when nodes reconnect or explicitly unsubscribe, the bidirectional index must stay in sync, otherwise delivery hits a disconnected node or misses a newly-connected one.

The event entry is the returned closure (dispatcher:1136). It first resolves sessionKey, agentId, and clientRunId (many events don't carry these fields, so they're reverse-looked-up from chatRunState.registry), then splits by stream type. sendChatPayload (sendChatPayload:890) splits by control UI visibility: visible → broadcast("chat", payload) plus sendNodeSessionPayloadForAgent for cross-node delivery; hidden → look up sessionMessageSubscribers.get(deliverySessionKey), and if there are subscribers, broadcastToConnIds for targeted delivery — typically operator backend tools, where the main panel doesn't see this run. sendAgentPayload (sendAgentPayload:967) is symmetric, just with event name "agent".

Cross-node delivery has one more detail (resolveNodeSessionDeliveryKeys:495): when sessionKey is "global", it expands into two delivery keys: agent:DEFAULT:global and global. Global events need to reach both clients subscribed to "some agent's global" and clients subscribed to "global" — but only when agentId is the default agent does it push both; otherwise only agent-scoped, to avoid duplicate delivery.

Boundaries and failures

• slow consumers: socket.bufferedAmount > MAX_BUFFERED_BYTES triggers backpressure. dropIfSlow: true (streaming delta, chat.send_timing) skips send but updates seq; dropIfSlow: false (critical lifecycle, tool results) directly calls socket.close(1008, "slow consumer") to disconnect — not disconnecting would let buffering drag the process down (backpressure:157).
• lifecycle error retry: the pendingTerminalLifecycleErrors Map tracks retry state for end / error phase events within a grace period. lifecycleErrorRetryGraceMs controls the window; a new lifecycle generation cancels the old timer.
• spawnedBy cache: resolveSpawnedBy only goes to storage for subagent / ACP sessionKeys; non-lineage returns null directly to avoid polluting the cache (spawnedByCache:383). High-frequency delta / flush / final paths avoid hitting session storage every time.
• restart recovery suppression: resolveRestartRecoveryLifecycleState checks whether a lifecycle event is a restart-recovery artifact; if so, suppress: true skips broadcasting — avoids re-pumping old recovered run events to clients after a restart.
• sessionKey resolution fallback: prefer chatLink.sessionKey, then evt.sessionKey, then resolveSessionKeyForRun(evt.runId). Three-layer fallback ensures events always find their owning session.
• cross-node delivery doesn't consume seq: isTargeted ? undefined : nextSeq. Incrementing seq on targeted broadcasts would make clients see seq gaps and assume events were lost.
• node subscription sync: nodeSendToSession goes through the subscription manager rather than nodeRegistry.send directly, so node reconnect / disconnect / unsubscribe keep the bidirectional index in sync. Bypassing the manager would miss delivery after node reconnect.

Summary

The broadcast layer separates agent event delivery policy from the main loop, abstracting it into three primitives injected by createGatewayNodeSessionRuntime and consumed by createAgentEventHandler. The real broadcast implementation (createGatewayBroadcaster) handles scope filtering, backpressure, and seq numbering; cross-node delivery routes through the subscription manager to keep indexes in sync across reconnects. The event entry is a dispatcher closure that splits by stream type into chat / agent / session.tool / sessions.changed, and each stream further splits by control-UI visibility into broadcast-all or targeted.

How requests come in and get dispatched is in RPC method table and request dispatch; how broadcast primitives are injected into the whole gateway is in Gateway core; how session subscription state binds to the agent run lifecycle is in Session Manager.

Official references: Gateway docs · README.