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.

Design Motivation

The agent's run_conversation runs in a worker thread and tokens are synchronously yielded — that's dictated by the underlying SDK (LLM streaming interfaces are usually blocking iterators). Platform sending, meanwhile, is an asyncio coroutine; calling it across threads directly would step on the event loop. Without an intermediate layer, you either buffer the entire reply before sending (the user stares at "typing..." for tens of seconds with no content) or call asyncio.run_coroutine_threadsafe from the worker thread on every token (heavy, and error handling gets messy). GatewayStreamConsumer uses a queue.Queue as the thread boundary — the worker thread only does queue.put, and the consumer coroutine does get_nowait from the asyncio side. The two clock domains are decoupled, and throttling, merging, and backoff all live on the consumer side. Hooks are listed separately because cross-cutting concerns like safety and billing should not pollute the main message pipeline.

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.

on_delta is the consumer's entry point — it does one thing: push the worker thread's token into a queue.Queue:

def on_delta(self, text: str) -> None:
    """Thread-safe callback — called from the agent's worker thread.

    When *text* is ``None``, signals a tool boundary: the current message
    is finalized and subsequent text will be sent as a new message so it
    appears below any tool-progress messages the gateway sent in between.
    """
    if text:
        self._queue.put(text)
    elif text is None:
        self.on_segment_break()

def finish(self) -> None:
    """Signal that the stream is complete."""
    self._queue.put(_DONE)

None is the tool boundary signal — subsequent tokens go to a new message, inserted below the tool-progress bubbles. _DONE is the sentinel that tells the consumer coroutine to trigger the final edit. "Value = token, None = boundary, sentinel = done" lets one queue carry the entire control flow.

The consumer coroutine run() pulls increments in batches and dispatches per strategy:

async def run(self) -> None:
    """Async task that drains the queue and edits the platform message."""
    _len_fn = (
        self.adapter.message_len_fn
        if isinstance(self.adapter, _BasePlatformAdapter)
        else len
    )
    _raw_limit = self._raw_message_limit()
    _safe_limit = max(500, _raw_limit - _len_fn(self.cfg.cursor) - 100)
    self._use_draft_streaming = self._resolve_draft_streaming()
    try:
        while True:
            if not self._run_still_current():
                return
            got_done = False
            while True:
                try:
                    item = self._queue.get_nowait()
                    if item is _DONE:
                        got_done = True
                        break

A few details: message_len_fn comes from the adapter (Telegram uses utf-16 length), so overflow checks match the platform's real limit; _safe_limit leaves 100 characters of headroom for the cursor and rich-text formatting; _run_still_current() is checked every loop so that if the session is /new'd or /stop'd, the coroutine exits early instead of pushing stale tokens to the screen.

GatewayEventDispatcher (gateway/stream_dispatch.py) is a more structured wrapper — the agent emits typed events (MessageChunk, ToolCallChunk, Commentary...), and the dispatcher decides how to render them based on adapter capability. When a platform can't render tool chrome, it returns None and eats the event. This is the extension point for "platforms that need native UI (DingTalk AI Card, Slack Block Kit) rather than plain text editing":

class GatewayEventDispatcher:
    """Route typed stream events through an adapter onto a delivery sink."""
    # Message/commentary/segment events flow into the consumer (native draft
    # on Telegram DMs, edit-in-place elsewhere).  Tool events are formatted
    # by the adapter — which may return None to *eat* the event on platforms
    # that can't render tool chrome — and the rendered line is enqueued onto
    # the same tool progress queue the gateway already drains, so the two no
    # longer race through independent code paths.

Boundaries and Failure

• Stream interrupted: after a session is /new'd or /stop'd, the consumer coroutine's next loop check of _run_still_current() returns immediately. Increments already in queue.put are lost, so before the final edit _flush_think_buffer() flushes any leftover half think tag.
• Flood control breach: _MAX_FLOOD_STRIKES = 3 — three consecutive 429s permanently degrades to non-progressive mode; remaining tokens for this stream are batched and sent in one shot.
• Platform doesn't support edit: Signal's SUPPORTED_MESSAGE_EDITING = False makes the consumer take the non-edit fallback — only one full message is sent, and no typewriter effect during streaming. That's a capability-flag fallback, not a bug.
• Sentinel missed: if the worker thread crashes after on_delta without calling finish(), the consumer coroutine blocks on queue.get(). run_still_current is the backstop — when the session is cleaned up, it triggers exit too.

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.