Streams & Hooks

源码版本v2026.7.20

Responsibility

(1) Stream consumption — the agent's worker thread synchronously calls stream_delta_callback(text) while platform sending is asynchronous; GatewayStreamConsumer bridges sync increments to async, progressively editing the same platform message so the user sees a "typewriter" effect. (2) Hooks — insert custom logic at message-lifecycle nodes (on receive, before generation, after generation, on delivery), such as safety filtering, billing, Kanban sync. (3) Relay — message relay across instances/profiles.

Key Files

• class GatewayStreamConsumer:83-120 — the async consumer itself; on_delta is fed to the agent as a callback
• class StreamConsumerConfig:55-83 — per-consumption runtime config (native draft streaming strategy, final-edit delay, etc.)
• module docstring:1-40 — explains the sync→async bridge principle and sentinel completion signal
• stream_dispatch — stream event dispatch
• stream_events — stream event type definitions
• hooks — lifecycle hook registration framework
• builtin_hooks/ directory — built-in hook implementations
• relay/ directory — cross-instance relay
• slash_commands — / command routing
• authz_mixin — authorization (mixed into GatewayRunner)
• profile_routing — multi-profile routing

Data Flow

1. GatewayRunner starts _start_stream_consumer:21218.
2. A GatewayStreamConsumer is constructed and consumer.on_delta is passed as stream_delta_callback to AIAgent (run_agent.py:400).
3. The agent calls on_delta(text) synchronously on the worker thread → the consumer enqueues the increment.
4. The consumer coroutine pulls increments from the queue and follows the StreamConsumerConfig:55 strategy:
   • auto/draft: prefer native draft streaming (send_draft:2650), fall back to plain edit
   • the final edit may be delayed (slow-reasoning-model scenario) to avoid high-frequency jitter
5. When the stream finishes, a sentinel signal triggers the final edit, which is recorded by delivery_ledger:155.
6. Throughout, hooks fire at each node (on receive / before generation / after generation / on delivery); built-in hooks (gateway/builtin_hooks/) handle safety, billing, Kanban, etc.

Summary

The stream consumer is a buffer and throttle layer between "synchronous agent callback" and "asynchronous platform delivery", resolving the clock-domain mismatch. Hooks are the insertion point for cross-cutting concerns (safety / billing / sync). Together they let the gateway support real-time typing, safety filtering, and cross-instance relay without touching the agent's trunk.