Chat-Broadcast und Ereignisauslieferung

源码版本v2026.6.11

Verantwortung

server-chat.ts liefert die vom Agent-Hauptschleife ausgespuckten Ereignisse (agent event) an Clients. Es produziert keine Ereignisse, sondern macht nur verteilte Auslieferung: Es nimmt AgentEventPayload (delta, final, tool, lifecycle u. a.), entscheidet nach Ereignistyp, Sitzungszugehörigkeit (session) und Client-Abonnements, ob broadcastet (broadcast) an alle sichtbaren Verbindungen, zielgerichtet ausgeliefert (broadcastToConnIds) an eine Gruppe von connIds oder knotenübergreifend ausgeliefert (nodeSendToSession) an einen anderen Knoten, der diese Sitzung abonniert hat.

Diese Schicht pflegt drei Abonnements-Zustände: toolEventRecipients (runId-weite Tool-Ereignis-Abonnenten, vom Client vor dem Run registriert), sessionEventSubscribers (sessionweit, wird von der Operator-UI angehängt, nachdem ein Run bereits läuft), sessionMessageSubscribers (sessionweite Chat-Nachrichten-Abonnements, zielgerichtet, wenn das Control-UI versteckt ist).

Designmotivation

Warum nicht einfach socket.send in der Agent-Hauptschleife? Eine Sitzung kann von mehreren Clients abonniert sein, Clients können über mehrere Gateway-Knoten verteilt sein, Agent-Ereignisse müssen nach Stream-Typ (agent / chat / session.tool / sessions.changed) aufgeteilt werden, einige Ereignisse müssen gedrosselt (throttle), andere bei Langsamkeit verworfen werden (drop if slow). All diese Strategien in die Hauptschleife zu quetschen, würde diese überladen — die Hauptschleife sollte nur den Agent-Zustandsautomaten pflegen, die Broadcast-Strategie wird in eine eigene Schicht ausgelagert.

Die drei Primitive werden von createGatewayNodeSessionRuntime (createGatewayNodeSessionRuntime:971) außen injiziert; createAgentEventHandler konsumiert nur und kümmert sich nicht um die Implementierung — das Chat-Modul lässt sich ohne echte WebSockets testen.

Schlüsseldateien

• AgentEventHandlerOptions:255-293 — Drei Broadcast-Primitive + drei Registry-Typdeklarationen für Abonnements.
• createAgentEventHandler Signatur:299-317 — Injiziert Broadcast-Primitive, gibt Event-Dispatcher zurück.
• resolveNodeSessionDeliveryKeys:495 — Globale Sitzung entfaltet sich in agent:DEFAULT:global + global als zwei Delivery-Keys.
• sendChatPayload:890 — Chat-Ereignis-Auslieferung, sichtbares Control-UI geht an broadcast, unsichtbares geht zielgerichtet.
• sendAgentPayload:967 — Agent-Ereignis-Auslieferung, gleiche Aufteilung.
• Dispatcher-Closure:1136 — (evt: AgentEventPayload) => {...} Ereigniseintritt.
• createGatewayBroadcaster:100-209 — Echte broadcast / broadcastToConnIds-Implementierung inklusive Scope-Filter und Backpressure.
• createGatewayNodeSessionRuntime:14-53 — nodeSendToSession routet über nodeSubscriptions.

Datenfluss

createAgentEventHandler (createAgentEventHandler:299) nimmt drei Broadcast-Primitive und drei Abonnements-Registries entgegen und gibt einen Event-Dispatcher (evt: AgentEventPayload) => void zurück. Intern pflegt es Throttle-, Lifecycle-Retry-, spawnedBy-Cache und andere Zustände.

broadcast und broadcastToConnIds stammen von createGatewayBroadcaster (createGatewayBroadcaster:100) und implementieren eine Iteration über clients: Set<GatewayWsClient> mit der Kernschleife:

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);  // Zielgerichtet verbraucht seq nicht
    continue;                                                       // Wenn verworbar, verwerfen
  }
  if (slow) { c.socket.close(1008, "slow consumer"); continue; }    // Nicht verwerfbar, Verbindung schließen
  c.socket.send(frame);  // {"type":"event","event":...,"payload":...,"seq":N}
}

Backpressure in zwei Stufen: dropIfSlow: true (Chat-Delta, hochfrequent verwerfbar) verwirft, aktualisiert aber seq, um Seq-Verschiebungen zu vermeiden; dropIfSlow: false (wichtige Lifecycle-, Tool-Ergebnisse) ruft direkt socket.close(1008, "slow consumer") auf und trennt die Verbindung — lieber Verbindung abbrechen als wichtige Ereignisse verlieren. seq wird nur bei nicht zielgerichteten Broadcasts inkrementiert (isTargeted ? undefined : nextSeq), zielgerichtete Broadcasts sind bereits eine gefilterte Teilmenge und nehmen nicht an der Client-Reihenfolge teil.

nodeSendToSession stammt von createGatewayNodeSessionRuntime (nodeSendToSession:31) und ist ein Einzeiler nodeSubscriptions.sendToSession(sessionKey, event, payload, nodeSendEvent). Der Quellcode-Kommentar „session fanout goes through the subscription manager so node reconnects and explicit unsubscribes keep both node->session indexes in sync" erklärt, warum knotenübergreifende Auslieferung über den Subscription-Manager läuft statt direkt nodeRegistry.send — bei Knoten-Reconnect und explizitem Unsubscribe müssen die bidirektionalen Indizes synchron bleiben, sonst wird an einen bereits getrennten Knoten geliefert oder ein neu verbundener Knoten übersehen.

Ereigniseintritt ist die zurückgegebene Closure (dispatcher:1136) — sie löst zuerst sessionKey, agentId und clientRunId auf (viele Ereignisse bringen diese Felder nicht mit, sie müssen über chatRunState.registry rückgesucht werden), dann nach Stream-Typ aufteilen. sendChatPayload (sendChatPayload:890) teilt nach Sichtbarkeit des Control-UI auf: sichtbar → broadcast("chat", payload) plus sendNodeSessionPayloadForAgent knotenübergreifend; unsichtbar → sessionMessageSubscribers.get(deliverySessionKey) abfragen, bei Abonnenten broadcastToConnIds zielgerichtet ausliefern — üblicherweise ein Hintergrundwerkzeug des Operators, das diesen Run im Hauptpanel nicht sieht. sendAgentPayload (sendAgentPayload:967) hat eine symmetrische Aufteilungslogik, nur mit anderem Event-Namen "agent".

Ein weiteres Detail bei der knotenübergreifenden Auslieferung (resolveNodeSessionDeliveryKeys:495): Wenn der sessionKey "global" ist, wird er in agent:DEFAULT:global und global als zwei Delivery-Keys entfaltet. Globale Ereignisse sollen sowohl Clients erreichen, die „global eines bestimmten Agenten" abonniert haben, als auch solche, die „global" abonniert haben; aber nur wenn die agentId der Standardagent ist, werden beide gepusht, sonst nur der agent-scoped — um Doppelbelieferung zu vermeiden.

Grenzen und Fehler

• Slow-Consumer-Behandlung: socket.bufferedAmount > MAX_BUFFERED_BYTES löst Backpressure aus. dropIfSlow: true (Streaming-Delta, chat.send_timing) überspringt Senden, inkrementiert aber seq; dropIfSlow: false (wichtige Lifecycle-, Tool-Ergebnisse) ruft direkt socket.close(1008, "slow consumer") auf, um die Verbindung zu trennen — ohne Trennung würde der Puffer den Prozess überlasten (backpressure:157).
• Lifecycle-Fehler-Wiederholung: pendingTerminalLifecycleErrors Map speichert den Wiederholungszustand von End/Error-Phase-Ereignissen innerhalb der Grace-Periode. lifecycleErrorRetryGraceMs steuert das Fenster; beim Eintreffen einer neuen Lifecycle-Generation wird der alte Timer abgebrochen.
• spawnedBy-Cache: resolveSpawnedBy macht nur für Subagent/ACP-SessionKeys eine Storage-Abfrage, Nicht-Lineage liefert direkt null zurück und verschmutzt den Cache nicht (spawnedByCache:383). Delta/flush/final auf hochfrequenten Pfaden vermeiden so pro Ereignis eine Session-Storage-Abfrage.
• Restart-Recovery-Unterdrückung: resolveRestartRecoveryLifecycleState prüft, ob ein Lifecycle-Ereignis ein Restart-Recovery-Produkt ist; wenn ja, wird suppress: true nicht gebroadcastet — sonst würden nach einem Restart die von alten Runs wiederhergestellten Ereignisse erneut an Clients gepusht.
• sessionKey-Auflösungs-Fallback: Erstens chatLink.sessionKey, zweitens evt.sessionKey, letztens resolveSessionKeyForRun(evt.runId). Drei Fallback-Stufen stellen sicher, dass jedes Ereignis eine zugehörige Sitzung findet.
• Knotenübergreifende Auslieferung verbraucht seq nicht: isTargeted ? undefined : nextSeq, bei zielgerichteten Broadcasts würde eine Inkrementierung die Client-Sicht auf Seq-Sprünge bringen und versehentlich Ereignisverlust vermuten lassen.
• Knotenabonnement-Synchronisation: nodeSendToSession läuft über den Subscription-Manager statt direkt nodeRegistry.send, damit bei Knoten-Reconnect/Trennung/Unsubscribe die bidirektionalen Indizes synchron bleiben. Umgehung des Managers würde nach Knoten-Reconnect Belieferung vergessen.

Zusammenfassung

Die Broadcast-Schicht koppelt die Agent-Ereignis-Auslieferungsstrategie von der Hauptschleife ab und abstrahiert sie in drei Primitive, die von createGatewayNodeSessionRuntime außen injiziert werden; createAgentEventHandler konsumiert nur. Die echte Broadcast-Implementierung (createGatewayBroadcaster) behandelt Scope-Filter, Backpressure und seq-Nummerierung; die knotenübergreifende Auslieferung läuft über den Subscription-Manager, damit Indizes nach Knoten-Reconnect synchron bleiben. Ereigniseintritt ist die Dispatcher-Closure, die nach Stream-Typ in chat / agent / session.tool / sessions.changed aufteilt, und jeder Stream geht je nach Control-UI-Sichtbarkeit voll oder zielgerichtet.

Wie Anfragen hereinkommen und verteilt werden, siehe RPC-Methodentabelle und Anfrageverteilung; wie Broadcast-Primitive ins gesamte Gateway injiziert werden, siehe Gateway-Kern; wie Sitzungsabonnements an den Agent-Run-Lebenszyklus gebunden werden, siehe Session Manager.

Vergleich mit offiziellen Ressourcen: Gateway-Doku · README.