From c757c5466c47ac4ba91905a6538ad360832aa970 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Wed, 1 Jul 2026 16:55:24 +0800 Subject: [PATCH] docs: update channel plugin runtime event contract --- docs/channel-plugin-guide.md | 102 ++++++++++++++++++++--------------- 1 file changed, 60 insertions(+), 42 deletions(-) diff --git a/docs/channel-plugin-guide.md b/docs/channel-plugin-guide.md index 292ad00f..179f2c09 100644 --- a/docs/channel-plugin-guide.md +++ b/docs/channel-plugin-guide.md @@ -103,7 +103,8 @@ class WebhookChannel(BaseChannel): msg.content — markdown text (convert to platform format as needed) msg.media — list of local file paths to attach msg.chat_id — the recipient (same chat_id you passed to _handle_message) - msg.metadata — may contain "_progress": True for streaming chunks + msg.metadata — channel routing context such as message/thread ids + msg.event — typed runtime event for progress/status messages """ logger.info("[webhook] -> {}: {}", msg.chat_id, msg.content[:80]) # In a real plugin: POST to a callback URL, send via SDK, etc. @@ -238,15 +239,15 @@ nanobot channels login --force # re-authenticate | `supports_streaming` (property) | `True` when config has `"streaming": true` **and** subclass overrides `send_delta()`. | | `is_running` | Returns `self._running`. | | `login(force=False)` | Perform interactive login (e.g. QR code scan). Returns `True` if already authenticated or login succeeds. Override in subclasses that support interactive login. | -| `send_reasoning_delta(chat_id, delta, metadata?)` | Optional hook for streamed model reasoning/thinking content. Default is no-op. | -| `send_reasoning_end(chat_id, metadata?)` | Optional hook marking the end of a reasoning block. Default is no-op. | +| `send_reasoning_delta(chat_id, delta, metadata?, *, stream_id?)` | Optional hook for streamed model reasoning/thinking content. Default is no-op. | +| `send_reasoning_end(chat_id, metadata?, *, stream_id?)` | Optional hook marking the end of a reasoning block. Default is no-op. | | `send_reasoning(msg)` | Optional one-shot reasoning fallback. Default translates to `send_reasoning_delta()` + `send_reasoning_end()`. | ### Optional (streaming) | Method | Description | |--------|-------------| -| `async send_delta(chat_id, delta, metadata?)` | Override to receive streaming chunks. See [Streaming Support](#streaming-support) for details. | +| `async send_delta(chat_id, delta, metadata?, *, stream_id?, stream_end=False, resuming=False)` | Override to receive streaming chunks. See [Streaming Support](#streaming-support) for details. | ### Message Types @@ -257,10 +258,12 @@ class OutboundMessage: chat_id: str # recipient (same value you passed to _handle_message) content: str # markdown text — convert to platform format as needed media: list[str] # local file paths to attach (images, audio, docs) - metadata: dict # may contain: "_progress" (bool) for streaming chunks, - # "message_id" for reply threading + metadata: dict # channel routing context, e.g. "message_id" for threading + event: object | None # typed runtime/UI event; usually inspect with isinstance() ``` +Runtime/UI semantics live on `msg.event`. Plugin-authored outbound messages should not rely on legacy metadata flags such as `_progress`, `_stream_delta`, `_stream_end`, `_reasoning_delta`, `_turn_end`, or `_goal_status`; they are treated as ordinary metadata, not parsed as runtime events. + ## Streaming Support Channels can opt into real-time streaming — the agent sends content token-by-token instead of one final message. This is entirely optional; channels work fine without it. @@ -279,10 +282,18 @@ If either is missing, the agent falls back to the normal one-shot `send()` path. Override `send_delta` to handle two types of calls: ```python -async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: - meta = metadata or {} - - if meta.get("_stream_end"): +async def send_delta( + self, + chat_id: str, + delta: str, + metadata: dict[str, Any] | None = None, + *, + stream_id: str | None = None, + stream_end: bool = False, + resuming: bool = False, +) -> None: + buffer_key = stream_id or chat_id + if stream_end: # Streaming finished — do final formatting, cleanup, etc. return @@ -290,12 +301,7 @@ async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | # delta contains a small chunk of text (a few tokens) ``` -**Metadata flags:** - -| Flag | Meaning | -|------|---------| -| `_stream_delta: True` | A content chunk (delta contains the new text) | -| `_stream_end: True` | Streaming finished (delta is empty) | +Streaming state is passed through keyword-only arguments, not `_stream_delta` or `_stream_end` metadata flags. Use `stream_id` to key any per-stream buffers; fall back to `chat_id` when it is missing. ### Example: Webhook with Streaming @@ -310,18 +316,27 @@ class WebhookChannel(BaseChannel): super().__init__(config, bus) self._buffers: dict[str, str] = {} - async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: - meta = metadata or {} - if meta.get("_stream_end"): - text = self._buffers.pop(chat_id, "") + async def send_delta( + self, + chat_id: str, + delta: str, + metadata: dict[str, Any] | None = None, + *, + stream_id: str | None = None, + stream_end: bool = False, + resuming: bool = False, + ) -> None: + buffer_key = stream_id or chat_id + if stream_end: + text = self._buffers.pop(buffer_key, "") # Final delivery — format and send the complete message await self._deliver(chat_id, text, final=True) return - self._buffers.setdefault(chat_id, "") - self._buffers[chat_id] += delta + self._buffers.setdefault(buffer_key, "") + self._buffers[buffer_key] += delta # Incremental update — push partial text to the client - await self._deliver(chat_id, self._buffers[chat_id], final=False) + await self._deliver(chat_id, self._buffers[buffer_key], final=False) async def send(self, msg: OutboundMessage) -> None: # Non-streaming path — unchanged @@ -350,7 +365,7 @@ When `streaming` is `false` (default) or omitted, only `send()` is called — no | Method / Property | Description | |-------------------|-------------| -| `async send_delta(chat_id, delta, metadata?)` | Override to handle streaming chunks. No-op by default. | +| `async send_delta(chat_id, delta, metadata?, *, stream_id?, stream_end=False, resuming=False)` | Override to handle streaming chunks. No-op by default. | | `supports_streaming` (property) | Returns `True` when config has `streaming: true` **and** subclass overrides `send_delta`. | ## Progress, Tool Hints, and Reasoning @@ -359,18 +374,20 @@ Besides normal assistant text, nanobot can emit low-emphasis trace blocks. These ### Progress and Tool Hints -Progress and tool hints arrive through the normal `send(msg)` path. Check `msg.metadata` before rendering: +Progress and tool hints arrive through the normal `send(msg)` path. Check `msg.event` before rendering: ```python -async def send(self, msg: OutboundMessage) -> None: - meta = msg.metadata or {} +from nanobot.bus.outbound_events import ProgressEvent - if meta.get("_tool_hint"): +async def send(self, msg: OutboundMessage) -> None: + event = msg.event + + if isinstance(event, ProgressEvent) and event.tool_hint: # A short tool breadcrumb, e.g. read_file("config.json") await self._send_trace(msg.chat_id, msg.content, kind="tool") return - if meta.get("_progress"): + if isinstance(event, ProgressEvent): # Generic non-final status, e.g. "Thinking..." or "Running command..." await self._send_trace(msg.chat_id, msg.content, kind="progress") return @@ -412,32 +429,33 @@ class WebhookChannel(BaseChannel): chat_id: str, delta: str, metadata: dict[str, Any] | None = None, + *, + stream_id: str | None = None, ) -> None: - meta = metadata or {} - stream_id = str(meta.get("_stream_id") or chat_id) - self._reasoning_buffers[stream_id] = self._reasoning_buffers.get(stream_id, "") + delta - await self._update_reasoning_block(chat_id, self._reasoning_buffers[stream_id], final=False) + buffer_key = stream_id or chat_id + self._reasoning_buffers[buffer_key] = self._reasoning_buffers.get(buffer_key, "") + delta + await self._update_reasoning_block(chat_id, self._reasoning_buffers[buffer_key], final=False) async def send_reasoning_end( self, chat_id: str, metadata: dict[str, Any] | None = None, + *, + stream_id: str | None = None, ) -> None: - meta = metadata or {} - stream_id = str(meta.get("_stream_id") or chat_id) - text = self._reasoning_buffers.pop(stream_id, "") + buffer_key = stream_id or chat_id + text = self._reasoning_buffers.pop(buffer_key, "") if text: await self._update_reasoning_block(chat_id, text, final=True) ``` -**Reasoning metadata flags:** +**Reasoning arguments:** -| Flag | Meaning | +| Argument | Meaning | |------|---------| -| `_reasoning_delta: True` | A reasoning/thinking chunk; `delta` contains the new text. | -| `_reasoning_end: True` | The current reasoning block is complete; `delta` is empty. | -| `_reasoning: True` | Legacy one-shot reasoning. `BaseChannel.send_reasoning()` converts it to delta + end. | -| `_stream_id` | Stable id for this assistant turn/segment. Use it to key buffers instead of only `chat_id`. | +| `delta` | A reasoning/thinking chunk for `send_reasoning_delta()`. | +| `stream_id` | Stable id for this assistant turn/segment. Use it to key buffers instead of only `chat_id`. | +| `send_reasoning_end()` | The current reasoning block is complete. | Reasoning visibility is controlled by `showReasoning` globally or per channel: