From 53ef912441103ce464fb0e5362ed60e5d6a96cfc Mon Sep 17 00:00:00 2001 From: Duy /zuey/ Date: Wed, 3 Jun 2026 11:01:14 +0700 Subject: [PATCH] fix(channels): add reasoning delivery modes (#135) --- cmd/gateway_consumer_normal.go | 15 +- docs/05-channels-messaging.md | 14 + docs/12-extended-thinking.md | 14 + docs/project-changelog.md | 22 ++ internal/agent/loop_pipeline_callbacks.go | 11 + .../agent/loop_pipeline_callbacks_test.go | 56 ++++ internal/channels/channel.go | 6 + .../channels/chat_behavior_events_test.go | 4 +- internal/channels/events.go | 87 +++++- internal/channels/manager.go | 51 ++-- internal/channels/reasoning_bubbles.go | 183 ++++++++++++ internal/channels/reasoning_delivery.go | 60 ++++ internal/channels/reasoning_delivery_test.go | 269 ++++++++++++++++++ internal/channels/runs.go | 26 +- internal/channels/telegram/channel.go | 11 +- internal/channels/telegram/factory.go | 80 +++--- internal/config/channel_reasoning_delivery.go | 80 ++++++ .../config/channel_reasoning_delivery_test.go | 45 +++ internal/config/config_channels.go | 41 +-- internal/gateway/methods/channel_instances.go | 18 +- internal/http/channel_instances.go | 18 +- .../channels/ChannelAdvancedTab.tsx | 5 +- .../components/channels/channel-schemas.ts | 5 +- .../channels/reasoning-delivery-config.ts | 20 ++ ui/web/src/i18n/locales/en/channels.json | 13 +- ui/web/src/i18n/locales/en/config.json | 7 +- ui/web/src/i18n/locales/vi/channels.json | 9 +- ui/web/src/i18n/locales/vi/config.json | 7 +- ui/web/src/i18n/locales/zh/channels.json | 9 +- ui/web/src/i18n/locales/zh/config.json | 7 +- .../channel-advanced-dialog.tsx | 7 +- .../pages/channels/channel-schemas.test.ts | 23 ++ ui/web/src/pages/channels/channel-schemas.ts | 3 +- .../channels/reasoning-delivery-config.ts | 28 ++ .../config/sections/channels-section.tsx | 31 +- 35 files changed, 1137 insertions(+), 148 deletions(-) create mode 100644 internal/agent/loop_pipeline_callbacks_test.go create mode 100644 internal/channels/reasoning_bubbles.go create mode 100644 internal/channels/reasoning_delivery.go create mode 100644 internal/channels/reasoning_delivery_test.go create mode 100644 internal/config/channel_reasoning_delivery.go create mode 100644 internal/config/channel_reasoning_delivery_test.go create mode 100644 ui/desktop/frontend/src/components/channels/reasoning-delivery-config.ts create mode 100644 ui/web/src/pages/channels/reasoning-delivery-config.ts diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index fc76f340..27ffca24 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -201,10 +201,13 @@ func processNormalMessage( "user_id", userID, ) - // Enable streaming when the channel supports it (so agent emits chunk events). - // The channel decides per chat type via separate dm_stream / group_stream flags. isGroup := peerKind == string(sessions.PeerGroup) - enableStream := deps.ChannelMgr != nil && deps.ChannelMgr.IsStreamingChannel(msg.Channel, isGroup) + channelStream := deps.ChannelMgr != nil && deps.ChannelMgr.IsStreamingChannel(msg.Channel, isGroup) + reasoningDelivery := channels.ResolveReasoningDelivery("", nil) + if deps.ChannelMgr != nil { + reasoningDelivery = deps.ChannelMgr.ResolveReasoningDelivery(msg.Channel) + } + providerStream := channels.ShouldStreamProviderForDelivery(channelStream, reasoningDelivery) // Group chats allow concurrent runs (multiple users can chat simultaneously). maxConcurrent := 1 @@ -246,7 +249,7 @@ func processNormalMessage( } toolStatus := deps.Cfg.Gateway.ToolStatus == nil || *deps.Cfg.Gateway.ToolStatus // default true if deps.ChannelMgr != nil { - deps.ChannelMgr.RegisterRunWithBehavior(runID, msg.Channel, chatIDForRun, messageID, outMeta, msg.TenantID, enableStream, blockReply, toolStatus, chatBehavior) + deps.ChannelMgr.RegisterRunWithBehavior(runID, msg.Channel, chatIDForRun, messageID, outMeta, msg.TenantID, channelStream, blockReply, toolStatus, chatBehavior, reasoningDelivery) } // Group-aware system prompt: help the LLM adapt tone and behavior for group chats. @@ -454,7 +457,7 @@ func processNormalMessage( Role: effectiveRole, SenderName: resolveSenderName(msg), RunID: runID, - Stream: enableStream, + Stream: providerStream, HistoryLimit: msg.HistoryLimit, ToolAllow: msg.ToolAllow, ExtraSystemPrompt: extraPrompt, @@ -602,7 +605,7 @@ func processNormalMessage( if deps.TeamStore != nil && channel != tools.ChannelSystem && channel != tools.ChannelTeammate && channel != tools.ChannelDashboard { go autoSetFollowup(ctx, deps.TeamStore, deps.AgentStore, agentKey, channel, chatID, replyContent) } - }(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, msg.Content, outMeta, blockReply, chatBehavior, enableStream, ptd, msg.TenantID, agentLoop.UUID(), agentLoop.OtherConfig()) + }(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, msg.Content, outMeta, blockReply, chatBehavior, channelStream, ptd, msg.TenantID, agentLoop.UUID(), agentLoop.OtherConfig()) } // isSafeBitrixEntityToken validates a webhook-sourced Bitrix entity token before diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index e7ca9d0d..04dec281 100644 --- a/docs/05-channels-messaging.md +++ b/docs/05-channels-messaging.md @@ -87,6 +87,18 @@ Splitting is intentionally conservative. Replies containing fenced code, tables, Progress messages are not added to session history by this behavior. Existing run timeline handling for `block.reply` remains unchanged. +### Reasoning Delivery + +Telegram channel config supports explicit reasoning delivery: + +| `reasoning_delivery` | Behavior | +|----------------------|----------| +| `streaming_only` | Legacy behavior. Show model reasoning only in the live streaming lane. | +| `always_bubbles` | Force provider streaming internally and send reasoning as bounded channel bubbles, even when `dm_stream` / `group_stream` are off. | +| `off` | Suppress reasoning output in the channel. Traces and provider usage remain unaffected. | + +Backward compatibility: if `reasoning_delivery` is missing, legacy `reasoning_stream=false` resolves to `off`; otherwise it resolves to `streaming_only`. Explicit `reasoning_delivery` always wins over the legacy boolean. Reasoning bubbles are delivery-only messages and are not added to assistant history. + **Multi-attachment coalescing (#63).** Messages carrying attachments do NOT bypass the debouncer — that pre-fix shortcut was the source of N-replies for one user action. Instead, when media is present the effective window is `max(configured, mediaFloor)` so multi-file uploads land in the same buffer and flush together. Three surfaces apply the same invariant: | Surface | Buffer key | Trigger | @@ -122,6 +134,7 @@ Every channel must implement the base interface: | `ReactionChannel` | Status reactions on messages | Telegram, Slack, Feishu | | `BlockReplyChannel` | Override gateway block_reply setting | Discord, Feishu/Lark, Pancake, Slack, Zalo OA, Zalo Personal | | `ChatBehaviorChannel` | Override gateway chat_behavior setting | Bitrix24, Discord, Feishu/Lark, Pancake, Slack, Telegram, WhatsApp, Zalo OA, Zalo Personal | +| `ReasoningDeliveryChannel` | Override channel-visible reasoning delivery | Telegram | `BaseChannel` provides a shared implementation that all channels embed: allowlist matching, `HandleMessage()`, `CheckPolicy()`, and user ID extraction. @@ -216,6 +229,7 @@ The Telegram channel uses long polling via the `telego` library (Telegram Bot AP - **Cancel commands**: `/stop` and `/stopall` intercepted before the 800ms debouncer. See [08-scheduling-cron.md](./08-scheduling-cron.md) for details. - **Concurrent group support**: Group sessions support up to 3 concurrent agent runs. - **Bot reply as implicit mention**: Replying to a bot message in a group counts as mentioning the bot. +- **Show Reasoning delivery**: `reasoning_delivery=always_bubbles` decouples provider streaming from Telegram live streaming so reasoning can appear as bounded normal messages while the final answer remains non-streaming. ### Formatting Pipeline diff --git a/docs/12-extended-thinking.md b/docs/12-extended-thinking.md index f5e60ae2..bde6623d 100644 --- a/docs/12-extended-thinking.md +++ b/docs/12-extended-thinking.md @@ -176,6 +176,20 @@ flowchart TD | DashScope | Same as OpenAI (when tools absent) | Same as OpenAI | | Codex | `reasoning` items with text summaries | `content` items | +### Channel Delivery + +Provider streaming and channel streaming are separate decisions. A channel may request provider streaming only to receive reasoning events, while still delivering the final answer as a normal non-streamed message. + +Telegram exposes this through `reasoning_delivery`: + +| Mode | Channel output | +|------|----------------| +| `streaming_only` | Show reasoning only when `dm_stream` / `group_stream` is enabled. | +| `always_bubbles` | Force provider streaming and flush reasoning into bounded channel bubbles. | +| `off` | Do not show reasoning in channel messages. | + +`reasoning_stream=false` remains a legacy alias for `off` when no explicit mode is present. The bubble path is not persisted as assistant history; it is only a channel delivery surface. + ### Token Estimation Thinking tokens are estimated as `character_count / 4` for context window tracking. This rough estimate ensures the agent loop can account for thinking overhead when calculating context usage. diff --git a/docs/project-changelog.md b/docs/project-changelog.md index df96c9d7..df610919 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -4,6 +4,28 @@ Significant changes, features, and fixes in reverse chronological order. --- +## 2026-06-01 + +### Telegram Show Reasoning delivery modes (issues #132, #133) + +**Fixes** + +- Added `reasoning_delivery=streaming_only|always_bubbles|off` with backward compatibility for legacy `reasoning_stream`. +- HTTP and WebSocket channel-instance writes normalize explicit `reasoning_delivery` over legacy `reasoning_stream`. +- `always_bubbles` forces provider streaming internally so reasoning can be delivered as bounded channel bubbles even when Telegram live streaming is disabled. +- Stream-path final `resp.Thinking` is now emitted when no thinking chunk arrived during the stream. +- Terminal channel events preserve interim delivery state until the consumer reads the final dedup snapshot. + +**UI** + +- Web and desktop channel settings now expose Show Reasoning as a mode selector instead of a streaming-only boolean. + +**Tests** + +- Added channel delivery, config resolution, and agent stream final-thinking coverage. + +--- + ## 2026-05-31 ### CI/CD: zuey release asset race fix diff --git a/internal/agent/loop_pipeline_callbacks.go b/internal/agent/loop_pipeline_callbacks.go index 3ddb2277..8e2125ff 100644 --- a/internal/agent/loop_pipeline_callbacks.go +++ b/internal/agent/loop_pipeline_callbacks.go @@ -301,8 +301,10 @@ func (l *Loop) makeCallLLM(req *RunRequest, emitRun func(AgentEvent)) func(ctx c } } + streamThinkingEmitted := false emitChunk := func(chunk providers.StreamChunk) { if chunk.Thinking != "" { + streamThinkingEmitted = true emitRun(AgentEvent{ Type: protocol.ChatEventThinking, AgentID: l.id, @@ -410,6 +412,15 @@ func (l *Loop) makeCallLLM(req *RunRequest, emitRun func(AgentEvent)) func(ctx c }()) } + if req.Stream && err == nil && resp != nil && resp.Thinking != "" && !streamThinkingEmitted { + emitRun(AgentEvent{ + Type: protocol.ChatEventThinking, + AgentID: l.id, + RunID: req.RunID, + Payload: map[string]string{"content": resp.Thinking}, + }) + } + // Non-streaming: emit content events matching v2 behavior (channels need these). if !req.Stream && err == nil && resp != nil { if resp.Thinking != "" { diff --git a/internal/agent/loop_pipeline_callbacks_test.go b/internal/agent/loop_pipeline_callbacks_test.go new file mode 100644 index 00000000..2e7617df --- /dev/null +++ b/internal/agent/loop_pipeline_callbacks_test.go @@ -0,0 +1,56 @@ +package agent + +import ( + "context" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/pipeline" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +type finalThinkingStreamProvider struct{} + +func (p finalThinkingStreamProvider) Chat(context.Context, providers.ChatRequest) (*providers.ChatResponse, error) { + return &providers.ChatResponse{Content: "final", Thinking: "non-stream thinking"}, nil +} + +func (p finalThinkingStreamProvider) ChatStream(context.Context, providers.ChatRequest, func(providers.StreamChunk)) (*providers.ChatResponse, error) { + return &providers.ChatResponse{Content: "final", Thinking: "final streamed thinking"}, nil +} + +func (p finalThinkingStreamProvider) DefaultModel() string { return "test-model" } +func (p finalThinkingStreamProvider) Name() string { return "test-provider" } + +func TestMakeCallLLM_StreamsFinalThinkingWhenNoThinkingChunkArrives(t *testing.T) { + col := &eventCollector{} + loop := &Loop{id: "test-agent", onEvent: col.onEvent} + req := &RunRequest{ + RunID: "run-1", + SessionKey: "sess-1", + Channel: "telegram", + Stream: true, + } + state := &pipeline.RunState{ + Provider: finalThinkingStreamProvider{}, + Model: "test-model", + Iteration: 0, + } + + resp, err := loop.makeCallLLM(req, col.onEvent)(context.Background(), state, providers.ChatRequest{}) + if err != nil { + t.Fatalf("makeCallLLM returned error: %v", err) + } + if resp == nil || resp.Thinking != "final streamed thinking" { + t.Fatalf("stream response = %+v, want final thinking preserved", resp) + } + + thinking := col.filter(protocol.ChatEventThinking) + if len(thinking) != 1 { + t.Fatalf("thinking events = %+v, want exactly one final thinking event", thinking) + } + payload, ok := thinking[0].Payload.(map[string]string) + if !ok || payload["content"] != "final streamed thinking" { + t.Fatalf("thinking payload = %+v", thinking[0].Payload) + } +} diff --git a/internal/channels/channel.go b/internal/channels/channel.go index 6f06468a..2e6653e4 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -171,6 +171,12 @@ type ChatBehaviorChannel interface { ChatBehaviorConfig() *config.ChatBehaviorConfig } +// ReasoningDeliveryChannel is optionally implemented by channels that expose +// how model reasoning should be delivered to end users. +type ReasoningDeliveryChannel interface { + ReasoningDeliveryConfig() (mode string, legacyReasoningStream *bool) +} + // WebhookChannel extends Channel with an HTTP handler that can be mounted // on the main gateway mux instead of starting a separate HTTP server. // This allows webhook-based channels (e.g. Feishu/Lark) to share the main diff --git a/internal/channels/chat_behavior_events_test.go b/internal/channels/chat_behavior_events_test.go index 8ecf7c8c..a2c9670f 100644 --- a/internal/channels/chat_behavior_events_test.go +++ b/internal/channels/chat_behavior_events_test.go @@ -261,8 +261,8 @@ func TestHandleAgentEvent_QuickAckDisabledSuppressesInitialExplicitBlockReply(t mgr.HandleAgentEvent(protocol.AgentEventRunCompleted, "run-1", nil) delivered, last = mgr.InterimDeliverySnapshot("run-1") - if delivered != 0 || last != "" { - t.Fatalf("completed interim delivery snapshot = (%d, %q), want none", delivered, last) + if delivered != 1 || last != "Second iteration update." { + t.Fatalf("completed interim delivery snapshot = (%d, %q), want preserved second reply until unregister", delivered, last) } mgr.UnregisterRun("run-1") delivered, last = mgr.InterimDeliverySnapshot("run-1") diff --git a/internal/channels/events.go b/internal/channels/events.go index 7ecfd380..ded32f7e 100644 --- a/internal/channels/events.go +++ b/internal/channels/events.go @@ -50,7 +50,8 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { if sc, ok := ch.(StreamingChannel); ok && rc.Streaming { switch eventType { case protocol.AgentEventRunStarted: - stream, err := sc.CreateStream(ctx, rc.ChatID, true) + firstStreamIsReasoning := rc.ReasoningDelivery.ShowInChannel && !rc.ReasoningDelivery.BubbleDelivery && sc.ReasoningStreamEnabled() + stream, err := sc.CreateStream(ctx, rc.ChatID, firstStreamIsReasoning) if err != nil { slog.Debug("stream start failed", "channel", rc.ChannelName, "error", err) } else { @@ -66,11 +67,15 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { // When the first chunk arrives, this stream is stopped (reasoning message stays // visible) and a new stream is created for the answer lane. // Gated by ReasoningStreamEnabled() — channels can opt out (e.g. Slack). - if !sc.ReasoningStreamEnabled() { - break - } content := extractPayloadString(payload, "content") if content != "" { + if rc.ReasoningDelivery.BubbleDelivery { + m.appendReasoningBubble(runID, rc, content) + break + } + if !rc.ReasoningDelivery.ShowInChannel || !sc.ReasoningStreamEnabled() { + break + } rc.mu.Lock() rc.thinkingBuffer += content rc.hasThinking = true @@ -82,6 +87,9 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { } } case protocol.AgentEventToolCall: + if rc.ReasoningDelivery.BubbleDelivery { + m.flushReasoningBubbles(runID) + } // Agent is executing a tool — mark tool phase so the next chunk // (new LLM iteration) resets the stream buffer. // Stop the current stream (reasoning or answer) and finalize only @@ -126,6 +134,9 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { // Accumulate chunk deltas into full text. content := extractPayloadString(payload, "content") if content != "" { + if rc.ReasoningDelivery.BubbleDelivery { + m.flushReasoningBubbles(runID) + } rc.mu.Lock() needNewStream := rc.inToolPhase if needNewStream { @@ -140,28 +151,54 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { candidate := rc.streamBuffer + content split := SplitThinkTags(candidate) if split.Thinking != "" { - // Found think tags — commit to buffer and route to reasoning lane + // Found think tags — commit to buffer and route or suppress reasoning + // before any tagged content can leak into the answer lane. + displayReasoningInStream := rc.ReasoningDelivery.ShowInChannel && !rc.ReasoningDelivery.BubbleDelivery && sc.ReasoningStreamEnabled() + previousThinking := rc.thinkingBuffer rc.streamBuffer = candidate - rc.hasThinking = true rc.thinkingBuffer = split.Thinking thinkText := rc.thinkingBuffer currentStream := rc.stream if split.Partial { - // Still inside — update reasoning stream, wait for close + // Still inside — wait for the close tag before streaming + // answer content. Native thinking uses hasThinking; tag parsing + // keeps it false until close so later chunks continue parsing. rc.mu.Unlock() - if currentStream != nil { + if rc.ReasoningDelivery.BubbleDelivery { + if delta := reasoningDelta(previousThinking, thinkText); delta != "" { + m.appendReasoningBubble(runID, rc, delta) + } + } else if displayReasoningInStream && currentStream != nil { currentStream.Update(ctx, formatReasoningPreview(thinkText)) } break } - // Tag closed — transition to answer + // Tag closed — transition to answer, or strip reasoning entirely + // when Show Reasoning is off. + answerText := split.Answer rc.thinkingDone = true - rc.streamBuffer = split.Answer + rc.hasThinking = displayReasoningInStream + rc.streamBuffer = answerText reasoningStream := currentStream rc.mu.Unlock() - // Stop reasoning stream + if rc.ReasoningDelivery.BubbleDelivery { + if delta := reasoningDelta(previousThinking, thinkText); delta != "" { + m.appendReasoningBubble(runID, rc, delta) + } + m.flushReasoningBubbles(runID) + } + + if !displayReasoningInStream { + if reasoningStream != nil && answerText != "" { + reasoningStream.Update(ctx, answerText) + } + break + } + + // Stop reasoning stream after showing the final extracted thinking. if reasoningStream != nil { + reasoningStream.Update(ctx, formatReasoningPreview(thinkText)) _ = reasoningStream.Stop(ctx) } // Create answer stream @@ -174,12 +211,12 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { rc.mu.Unlock() } // Update answer stream with extracted answer content - if split.Answer != "" { + if answerText != "" { rc.mu.Lock() currentStream = rc.stream rc.mu.Unlock() if currentStream != nil { - currentStream.Update(ctx, split.Answer) + currentStream.Update(ctx, answerText) } } break @@ -270,6 +307,26 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { } } + if !rc.Streaming && rc.ReasoningDelivery.BubbleDelivery { + switch eventType { + case protocol.ChatEventThinking: + content := extractPayloadString(payload, "content") + if content != "" { + m.appendReasoningBubble(runID, rc, content) + } + return + case protocol.ChatEventChunk: + content := extractPayloadString(payload, "content") + if content != "" { + m.appendReasoningBubbleFromThinkTagChunk(runID, rc, content) + } + m.flushReasoningBubbles(runID) + return + case protocol.AgentEventToolCall, protocol.AgentEventRunCompleted, protocol.AgentEventRunFailed, protocol.AgentEventRunCancelled: + m.flushReasoningBubbles(runID) + } + } + // Handle block.reply: deliver intermediate assistant text to non-streaming channels. // Gated by explicit block_reply or generated-progress chat behavior. // Streaming channels already deliver via chunks, so skip to avoid double-delivery. @@ -377,7 +434,9 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { // Clean up on terminal events if eventType == protocol.AgentEventRunCompleted || eventType == protocol.AgentEventRunFailed || eventType == protocol.AgentEventRunCancelled { m.cancelQuickAck(rc) - m.runs.Delete(runID) + rc.mu.Lock() + stopReasoningBubbleTimerLocked(rc) + rc.mu.Unlock() } } diff --git a/internal/channels/manager.go b/internal/channels/manager.go index 946b05bd..b732eb87 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -29,30 +29,33 @@ type ChannelStream interface { // RunContext tracks an active agent run for streaming/reaction event forwarding. type RunContext struct { - ChannelName string - ChatID string - MessageID string // platform message ID (string to support Feishu "om_xxx", Telegram "12345", etc.) - Metadata map[string]string // outbound routing metadata (thread_id, local_key, group_id) - TenantID uuid.UUID // tenant scope for per-tenant TTS - Streaming bool // whether run uses streaming (to avoid double-delivery of block replies) - BlockReplyEnabled bool // whether block.reply delivery is enabled for this run (resolved at RegisterRun time) - ToolStatusEnabled bool // whether tool name shows in streaming preview during tool execution - ChatBehavior ResolvedChatBehavior - mu sync.Mutex - ackTimer *time.Timer - ackSent bool - ackCancelled bool - blockReplySent bool - blockReplySeen int - interimDelivered int - lastInterimReply string - streamBuffer string // accumulated streaming text (chunks are deltas) - inToolPhase bool // true after tool.call, reset on next chunk (new LLM iteration) - stream ChannelStream // per-run stream handle (replaces per-chat sync.Map in channel impls) - thinkingBuffer string // accumulated thinking/reasoning text - hasThinking bool // true if any thinking events received this iteration - thinkingDone bool // true after first chunk arrives (reasoning→answer transition complete) - tagParseSkipped bool // true after first chunk with no tags (skip re-parsing) + ChannelName string + ChatID string + MessageID string // platform message ID (string to support Feishu "om_xxx", Telegram "12345", etc.) + Metadata map[string]string // outbound routing metadata (thread_id, local_key, group_id) + TenantID uuid.UUID // tenant scope for per-tenant TTS + Streaming bool // whether run uses streaming (to avoid double-delivery of block replies) + BlockReplyEnabled bool // whether block.reply delivery is enabled for this run (resolved at RegisterRun time) + ToolStatusEnabled bool // whether tool name shows in streaming preview during tool execution + ChatBehavior ResolvedChatBehavior + ReasoningDelivery ResolvedReasoningDelivery + mu sync.Mutex + ackTimer *time.Timer + ackSent bool + ackCancelled bool + blockReplySent bool + blockReplySeen int + interimDelivered int + lastInterimReply string + streamBuffer string // accumulated streaming text (chunks are deltas) + inToolPhase bool // true after tool.call, reset on next chunk (new LLM iteration) + stream ChannelStream // per-run stream handle (replaces per-chat sync.Map in channel impls) + thinkingBuffer string // accumulated thinking/reasoning text + hasThinking bool // true if any thinking events received this iteration + thinkingDone bool // true after first chunk arrives (reasoning→answer transition complete) + tagParseSkipped bool // true after first chunk with no tags (skip re-parsing) + reasoningBubbles *reasoningBubbleBuffer + reasoningBubbleTimer *time.Timer } // Manager manages all registered channels, handling their lifecycle diff --git a/internal/channels/reasoning_bubbles.go b/internal/channels/reasoning_bubbles.go new file mode 100644 index 00000000..e82f5f10 --- /dev/null +++ b/internal/channels/reasoning_bubbles.go @@ -0,0 +1,183 @@ +package channels + +import ( + "log/slog" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/bus" +) + +const ( + reasoningBubbleMaxRunes = 1200 + reasoningBubbleMaxMessages = 6 + reasoningBubbleMaxRetainedRunes = reasoningBubbleMaxRunes * reasoningBubbleMaxMessages + reasoningBubbleFlushInterval = 900 * time.Millisecond +) + +type reasoningBubbleBuffer struct { + pending []rune + retainedRunes int + emitted int + truncated bool +} + +func (b *reasoningBubbleBuffer) append(content string) []string { + if content == "" || b.emitted >= reasoningBubbleMaxMessages { + return nil + } + for _, r := range content { + if b.retainedRunes >= reasoningBubbleMaxRetainedRunes { + b.truncated = true + break + } + b.pending = append(b.pending, r) + b.retainedRunes++ + } + if len(b.pending) >= reasoningBubbleMaxRunes || b.truncated { + return b.flush() + } + return nil +} + +func (b *reasoningBubbleBuffer) flush() []string { + if len(b.pending) == 0 || b.emitted >= reasoningBubbleMaxMessages { + return nil + } + var messages []string + for len(b.pending) > 0 && b.emitted < reasoningBubbleMaxMessages { + n := min(len(b.pending), reasoningBubbleMaxRunes) + chunk := string(b.pending[:n]) + b.pending = b.pending[n:] + if len(b.pending) == 0 && b.truncated { + chunk += "\n\n_Reasoning truncated._" + } + b.emitted++ + messages = append(messages, formatReasoningBubble(chunk, b.emitted)) + } + if b.emitted >= reasoningBubbleMaxMessages { + b.pending = nil + } + return messages +} + +func formatReasoningBubble(content string, index int) string { + if index <= 1 { + return "_Reasoning:_\n" + content + } + return "_Reasoning continued:_\n" + content +} + +func reasoningDelta(previous, next string) string { + if next == "" || next == previous { + return "" + } + if previous != "" && strings.HasPrefix(next, previous) { + return next[len(previous):] + } + return next +} + +func (m *Manager) appendReasoningBubble(runID string, rc *RunContext, content string) { + var messages []string + rc.mu.Lock() + if !rc.ReasoningDelivery.BubbleDelivery { + rc.mu.Unlock() + return + } + if rc.reasoningBubbles == nil { + rc.reasoningBubbles = &reasoningBubbleBuffer{} + } + messages = rc.reasoningBubbles.append(content) + if len(messages) > 0 { + stopReasoningBubbleTimerLocked(rc) + } else if len(rc.reasoningBubbles.pending) > 0 && rc.reasoningBubbleTimer == nil { + rc.reasoningBubbleTimer = time.AfterFunc(reasoningBubbleFlushInterval, func() { + m.flushReasoningBubbles(runID) + }) + } + rc.mu.Unlock() + m.publishReasoningBubbles(rc, messages) +} + +func (m *Manager) appendReasoningBubbleFromThinkTagChunk(runID string, rc *RunContext, content string) { + var delta string + var done bool + + rc.mu.Lock() + if rc.thinkingDone || rc.tagParseSkipped { + rc.mu.Unlock() + return + } + + candidate := rc.streamBuffer + content + split := SplitThinkTags(candidate) + if split.Thinking == "" { + rc.tagParseSkipped = true + rc.mu.Unlock() + return + } + + previousThinking := rc.thinkingBuffer + rc.streamBuffer = candidate + rc.thinkingBuffer = split.Thinking + if !split.Partial { + rc.thinkingDone = true + rc.streamBuffer = split.Answer + done = true + } + delta = reasoningDelta(previousThinking, split.Thinking) + rc.mu.Unlock() + + if delta != "" { + m.appendReasoningBubble(runID, rc, delta) + } + if done { + m.flushReasoningBubbles(runID) + } +} + +func (m *Manager) flushReasoningBubbles(runID string) { + val, ok := m.runs.Load(runID) + if !ok { + return + } + rc, ok := val.(*RunContext) + if !ok { + return + } + m.flushReasoningBubblesForContext(rc) +} + +func (m *Manager) flushReasoningBubblesForContext(rc *RunContext) { + var messages []string + rc.mu.Lock() + stopReasoningBubbleTimerLocked(rc) + if rc.reasoningBubbles != nil { + messages = rc.reasoningBubbles.flush() + } + rc.mu.Unlock() + m.publishReasoningBubbles(rc, messages) +} + +func stopReasoningBubbleTimerLocked(rc *RunContext) { + if rc.reasoningBubbleTimer != nil { + rc.reasoningBubbleTimer.Stop() + rc.reasoningBubbleTimer = nil + } +} + +func (m *Manager) publishReasoningBubbles(rc *RunContext, messages []string) { + for _, content := range messages { + msg := bus.OutboundMessage{ + Channel: rc.ChannelName, + ChatID: rc.ChatID, + Content: content, + Metadata: copyRoutingMeta(rc.Metadata), + TenantID: rc.TenantID, + } + if !m.bus.TryPublishOutbound(msg) { + slog.Warn("reasoning bubble dropped because outbound queue is full", "channel", rc.ChannelName) + } + } +} diff --git a/internal/channels/reasoning_delivery.go b/internal/channels/reasoning_delivery.go new file mode 100644 index 00000000..e9544434 --- /dev/null +++ b/internal/channels/reasoning_delivery.go @@ -0,0 +1,60 @@ +package channels + +import "strings" + +const ( + ReasoningDeliveryOff = "off" + ReasoningDeliveryStreamingOnly = "streaming_only" + ReasoningDeliveryAlwaysBubbles = "always_bubbles" +) + +type ResolvedReasoningDelivery struct { + Mode string + ShowInChannel bool + ForceProviderStream bool + BubbleDelivery bool +} + +func ResolveReasoningDelivery(mode string, legacyReasoningStream *bool) ResolvedReasoningDelivery { + switch NormalizeReasoningDeliveryMode(mode) { + case ReasoningDeliveryOff: + return ResolvedReasoningDelivery{Mode: ReasoningDeliveryOff} + case ReasoningDeliveryAlwaysBubbles: + return ResolvedReasoningDelivery{ + Mode: ReasoningDeliveryAlwaysBubbles, + ShowInChannel: true, + ForceProviderStream: true, + BubbleDelivery: true, + } + case ReasoningDeliveryStreamingOnly: + return ResolvedReasoningDelivery{ + Mode: ReasoningDeliveryStreamingOnly, + ShowInChannel: true, + } + } + + if legacyReasoningStream != nil && !*legacyReasoningStream { + return ResolvedReasoningDelivery{Mode: ReasoningDeliveryOff} + } + return ResolvedReasoningDelivery{ + Mode: ReasoningDeliveryStreamingOnly, + ShowInChannel: true, + } +} + +func NormalizeReasoningDeliveryMode(mode string) string { + switch strings.ToLower(strings.TrimSpace(mode)) { + case ReasoningDeliveryOff: + return ReasoningDeliveryOff + case ReasoningDeliveryStreamingOnly: + return ReasoningDeliveryStreamingOnly + case ReasoningDeliveryAlwaysBubbles: + return ReasoningDeliveryAlwaysBubbles + default: + return "" + } +} + +func ShouldStreamProviderForDelivery(channelStreaming bool, delivery ResolvedReasoningDelivery) bool { + return channelStreaming || delivery.ForceProviderStream +} diff --git a/internal/channels/reasoning_delivery_test.go b/internal/channels/reasoning_delivery_test.go new file mode 100644 index 00000000..eb187fe2 --- /dev/null +++ b/internal/channels/reasoning_delivery_test.go @@ -0,0 +1,269 @@ +package channels + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +func TestResolveReasoningDelivery_PrecedenceAndLegacyFallback(t *testing.T) { + falseValue := false + trueValue := true + + tests := []struct { + name string + mode string + legacy *bool + wantMode string + wantShow bool + wantForce bool + wantBubbles bool + wantStreaming bool + }{ + { + name: "explicit always bubbles wins over legacy false", + mode: ReasoningDeliveryAlwaysBubbles, + legacy: &falseValue, + wantMode: ReasoningDeliveryAlwaysBubbles, + wantShow: true, + wantForce: true, + wantBubbles: true, + wantStreaming: true, + }, + { + name: "explicit streaming only wins over legacy false", + mode: ReasoningDeliveryStreamingOnly, + legacy: &falseValue, + wantMode: ReasoningDeliveryStreamingOnly, + wantShow: true, + wantForce: false, + wantBubbles: false, + wantStreaming: false, + }, + { + name: "explicit off wins over legacy true", + mode: ReasoningDeliveryOff, + legacy: &trueValue, + wantMode: ReasoningDeliveryOff, + wantShow: false, + wantForce: false, + wantBubbles: false, + wantStreaming: true, + }, + { + name: "legacy false maps to off", + mode: "", + legacy: &falseValue, + wantMode: ReasoningDeliveryOff, + wantShow: false, + wantForce: false, + wantBubbles: false, + wantStreaming: false, + }, + { + name: "missing legacy preserves streaming only default", + mode: "", + legacy: nil, + wantMode: ReasoningDeliveryStreamingOnly, + wantShow: true, + wantForce: false, + wantBubbles: false, + wantStreaming: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := ResolveReasoningDelivery(tc.mode, tc.legacy) + if got.Mode != tc.wantMode || got.ShowInChannel != tc.wantShow || got.ForceProviderStream != tc.wantForce || got.BubbleDelivery != tc.wantBubbles { + t.Fatalf("ResolveReasoningDelivery() = %+v", got) + } + if ShouldStreamProviderForDelivery(tc.wantStreaming, got) != (tc.wantStreaming || tc.wantForce) { + t.Fatalf("ShouldStreamProviderForDelivery(%v, %+v) returned unexpected result", tc.wantStreaming, got) + } + }) + } +} + +func TestHandleAgentEvent_AlwaysBubblesReasoningForNonStreamingRun(t *testing.T) { + mb := bus.New() + mgr := NewManager(mb) + mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"}) + delivery := ResolveReasoningDelivery(ReasoningDeliveryAlwaysBubbles, nil) + mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", map[string]string{"local_key": "chat-1/topic"}, uuid.Nil, false, false, true, ResolvedChatBehavior{}, delivery) + + mgr.HandleAgentEvent(protocol.ChatEventThinking, "run-1", map[string]string{"content": "first "}) + mgr.HandleAgentEvent(protocol.ChatEventThinking, "run-1", map[string]string{"content": "second"}) + mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "final chunk should stay internal"}) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + got, ok := mb.SubscribeOutbound(ctx) + if !ok { + t.Fatal("expected batched reasoning outbound message") + } + if got.ChatID != "chat-1" || got.Metadata["local_key"] != "chat-1/topic" { + t.Fatalf("reasoning outbound routing = %+v", got) + } + if !strings.Contains(got.Content, "first second") { + t.Fatalf("reasoning outbound content = %q, want batched thinking", got.Content) + } + if strings.Contains(got.Content, "final chunk") { + t.Fatalf("reasoning outbound leaked final content chunk: %q", got.Content) + } + + ctx, cancel = context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + if got, ok := mb.SubscribeOutbound(ctx); ok { + t.Fatalf("unexpected duplicate outbound after reasoning bubble: %+v", got) + } +} + +func TestHandleAgentEvent_ReasoningOffSuppressesThinking(t *testing.T) { + mb := bus.New() + mgr := NewManager(mb) + mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"}) + delivery := ResolveReasoningDelivery(ReasoningDeliveryOff, nil) + mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, false, false, true, ResolvedChatBehavior{}, delivery) + + mgr.HandleAgentEvent(protocol.ChatEventThinking, "run-1", map[string]string{"content": "hidden"}) + mgr.HandleAgentEvent(protocol.AgentEventRunCompleted, "run-1", nil) + + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + if got, ok := mb.SubscribeOutbound(ctx); ok { + t.Fatalf("reasoning off emitted outbound message: %+v", got) + } +} + +func TestHandleAgentEvent_AlwaysBubblesUsesAnswerStreamWhenChannelStreaming(t *testing.T) { + mb := bus.New() + mgr := NewManager(mb) + ch := &reasoningStreamingTestChannel{name: "test", reasoningEnabled: true} + mgr.RegisterChannel("test", ch) + delivery := ResolveReasoningDelivery(ReasoningDeliveryAlwaysBubbles, nil) + mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, ResolvedChatBehavior{}, delivery) + + mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil) + mgr.HandleAgentEvent(protocol.ChatEventThinking, "run-1", map[string]string{"content": "native reasoning"}) + mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "answer"}) + + if len(ch.streams) != 1 { + t.Fatalf("streams = %d, want answer stream only", len(ch.streams)) + } + if ch.streams[0].firstStream { + t.Fatal("always_bubbles created a reasoning stream instead of answer stream") + } + if got := strings.Join(ch.streams[0].updates, " "); got != "answer" { + t.Fatalf("stream updates = %q, want answer only", got) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + got, ok := mb.SubscribeOutbound(ctx) + if !ok { + t.Fatal("expected reasoning bubble outbound message") + } + if !strings.Contains(got.Content, "native reasoning") { + t.Fatalf("reasoning bubble content = %q", got.Content) + } +} + +func TestHandleAgentEvent_ReasoningOffStripsThinkTagsFromStreamingAnswer(t *testing.T) { + mb := bus.New() + mgr := NewManager(mb) + ch := &reasoningStreamingTestChannel{name: "test", reasoningEnabled: true} + mgr.RegisterChannel("test", ch) + delivery := ResolveReasoningDelivery(ReasoningDeliveryOff, nil) + mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, ResolvedChatBehavior{}, delivery) + + mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil) + mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "hiddenvisible"}) + + if len(ch.streams) != 1 { + t.Fatalf("streams = %d, want answer stream only", len(ch.streams)) + } + if ch.streams[0].firstStream { + t.Fatal("reasoning off created a reasoning stream") + } + if got := strings.Join(ch.streams[0].updates, " "); got != "visible" { + t.Fatalf("stream updates = %q, want stripped answer", got) + } + + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + if got, ok := mb.SubscribeOutbound(ctx); ok { + t.Fatalf("reasoning off emitted outbound message: %+v", got) + } +} + +func TestHandleAgentEvent_AlwaysBubblesExtractsThinkTagsFromStreamingChunk(t *testing.T) { + mb := bus.New() + mgr := NewManager(mb) + ch := &reasoningStreamingTestChannel{name: "test", reasoningEnabled: true} + mgr.RegisterChannel("test", ch) + delivery := ResolveReasoningDelivery(ReasoningDeliveryAlwaysBubbles, nil) + mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, ResolvedChatBehavior{}, delivery) + + mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil) + mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "tagged reasoningtagged answer"}) + + if len(ch.streams) != 1 { + t.Fatalf("streams = %d, want answer stream only", len(ch.streams)) + } + if got := strings.Join(ch.streams[0].updates, " "); got != "tagged answer" { + t.Fatalf("stream updates = %q, want stripped answer", got) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + got, ok := mb.SubscribeOutbound(ctx) + if !ok { + t.Fatal("expected think-tag reasoning bubble") + } + if !strings.Contains(got.Content, "tagged reasoning") { + t.Fatalf("reasoning bubble content = %q", got.Content) + } +} + +type reasoningStreamingTestChannel struct { + chatBehaviorTestChannel + name string + reasoningEnabled bool + streams []*reasoningRecordingStream +} + +func (c *reasoningStreamingTestChannel) Name() string { return c.name } +func (c *reasoningStreamingTestChannel) Type() string { return c.name } +func (c *reasoningStreamingTestChannel) StreamEnabled(bool) bool { return true } +func (c *reasoningStreamingTestChannel) ReasoningStreamEnabled() bool { return c.reasoningEnabled } +func (c *reasoningStreamingTestChannel) FinalizeStream(context.Context, string, ChannelStream) { +} +func (c *reasoningStreamingTestChannel) CreateStream(_ context.Context, _ string, firstStream bool) (ChannelStream, error) { + stream := &reasoningRecordingStream{firstStream: firstStream} + c.streams = append(c.streams, stream) + return stream, nil +} + +type reasoningRecordingStream struct { + firstStream bool + updates []string + stopped bool +} + +func (s *reasoningRecordingStream) Update(_ context.Context, text string) { + s.updates = append(s.updates, text) +} + +func (s *reasoningRecordingStream) Stop(context.Context) error { + s.stopped = true + return nil +} + +func (s *reasoningRecordingStream) MessageID() int { return 1 } diff --git a/internal/channels/runs.go b/internal/channels/runs.go index 1825d991..9a9a43d9 100644 --- a/internal/channels/runs.go +++ b/internal/channels/runs.go @@ -16,7 +16,11 @@ func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, meta // RegisterRunWithBehavior associates a run ID with channel context and // resolved delivery behavior so event handlers do not read mutable config mid-run. -func (m *Manager) RegisterRunWithBehavior(runID, channelName, chatID, messageID string, metadata map[string]string, tenantID uuid.UUID, streaming, blockReply, toolStatus bool, chatBehavior ResolvedChatBehavior) { +func (m *Manager) RegisterRunWithBehavior(runID, channelName, chatID, messageID string, metadata map[string]string, tenantID uuid.UUID, streaming, blockReply, toolStatus bool, chatBehavior ResolvedChatBehavior, reasoningDelivery ...ResolvedReasoningDelivery) { + delivery := ResolveReasoningDelivery("", nil) + if len(reasoningDelivery) > 0 { + delivery = reasoningDelivery[0] + } m.runs.Store(runID, &RunContext{ ChannelName: channelName, ChatID: chatID, @@ -27,6 +31,7 @@ func (m *Manager) RegisterRunWithBehavior(runID, channelName, chatID, messageID BlockReplyEnabled: blockReply, ToolStatusEnabled: toolStatus, ChatBehavior: chatBehavior, + ReasoningDelivery: delivery, }) } @@ -35,6 +40,7 @@ func (m *Manager) UnregisterRun(runID string) { if val, ok := m.runs.LoadAndDelete(runID); ok { if rc, ok := val.(*RunContext); ok { m.cancelQuickAck(rc) + m.flushReasoningBubblesForContext(rc) } } } @@ -99,3 +105,21 @@ func (m *Manager) ResolveChatBehavior(channelName string, globalDefault *config. } return ResolveChatBehavior(globalDefault, override) } + +func (m *Manager) ResolveReasoningDelivery(channelName string) ResolvedReasoningDelivery { + m.mu.RLock() + ch, exists := m.channels[channelName] + m.mu.RUnlock() + if !exists { + return ResolveReasoningDelivery("", nil) + } + if rc, ok := ch.(ReasoningDeliveryChannel); ok { + mode, legacy := rc.ReasoningDeliveryConfig() + return ResolveReasoningDelivery(mode, legacy) + } + if sc, ok := ch.(StreamingChannel); ok { + legacy := sc.ReasoningStreamEnabled() + return ResolveReasoningDelivery("", &legacy) + } + return ResolveReasoningDelivery("", nil) +} diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index b8005e79..9c05cd2c 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -367,12 +367,13 @@ func (c *Channel) draftTransportEnabled() bool { } // ReasoningStreamEnabled returns whether reasoning should be shown as a separate message. -// Default: true. Set "reasoning_stream": false to hide reasoning (only show answer). +// Default: true. Set "reasoning_delivery": "off" or legacy "reasoning_stream": false to hide reasoning. func (c *Channel) ReasoningStreamEnabled() bool { - if c.config.ReasoningStream == nil { - return true - } - return *c.config.ReasoningStream + return channels.ResolveReasoningDelivery(c.config.ReasoningDelivery, c.config.ReasoningStream).ShowInChannel +} + +func (c *Channel) ReasoningDeliveryConfig() (string, *bool) { + return c.config.ReasoningDelivery, c.config.ReasoningStream } // BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default). diff --git a/internal/channels/telegram/factory.go b/internal/channels/telegram/factory.go index 8d0078b7..7bf16f46 100644 --- a/internal/channels/telegram/factory.go +++ b/internal/channels/telegram/factory.go @@ -20,25 +20,26 @@ type telegramCreds struct { // telegramInstanceConfig maps the non-secret config JSONB from the channel_instances table. type telegramInstanceConfig struct { - APIServer string `json:"api_server,omitempty"` - Proxy string `json:"proxy,omitempty"` - DMPolicy string `json:"dm_policy,omitempty"` - GroupPolicy string `json:"group_policy,omitempty"` - RequireMention *bool `json:"require_mention,omitempty"` - MentionMode string `json:"mention_mode,omitempty"` - HistoryLimit int `json:"history_limit,omitempty"` - DMStream *bool `json:"dm_stream,omitempty"` - GroupStream *bool `json:"group_stream,omitempty"` - DraftTransport *bool `json:"draft_transport,omitempty"` // sendMessageDraft for DM streaming (default true) - ReasoningStream *bool `json:"reasoning_stream,omitempty"` // show reasoning as separate message (default true) - ReactionLevel string `json:"reaction_level,omitempty"` - MediaMaxMB int64 `json:"media_max_mb,omitempty"` - MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // deprecated: use media_max_mb - LinkPreview *bool `json:"link_preview,omitempty"` - BlockReply *bool `json:"block_reply,omitempty"` - ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"` - ForceIPv4 bool `json:"force_ipv4,omitempty"` - AllowFrom []string `json:"allow_from,omitempty"` + APIServer string `json:"api_server,omitempty"` + Proxy string `json:"proxy,omitempty"` + DMPolicy string `json:"dm_policy,omitempty"` + GroupPolicy string `json:"group_policy,omitempty"` + RequireMention *bool `json:"require_mention,omitempty"` + MentionMode string `json:"mention_mode,omitempty"` + HistoryLimit int `json:"history_limit,omitempty"` + DMStream *bool `json:"dm_stream,omitempty"` + GroupStream *bool `json:"group_stream,omitempty"` + DraftTransport *bool `json:"draft_transport,omitempty"` // sendMessageDraft for DM streaming (default true) + ReasoningDelivery string `json:"reasoning_delivery,omitempty"` + ReasoningStream *bool `json:"reasoning_stream,omitempty"` // show reasoning as separate message (default true) + ReactionLevel string `json:"reaction_level,omitempty"` + MediaMaxMB int64 `json:"media_max_mb,omitempty"` + MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // deprecated: use media_max_mb + LinkPreview *bool `json:"link_preview,omitempty"` + BlockReply *bool `json:"block_reply,omitempty"` + ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"` + ForceIPv4 bool `json:"force_ipv4,omitempty"` + AllowFrom []string `json:"allow_from,omitempty"` } // Factory creates a Telegram channel from DB instance data (no extra stores). @@ -97,26 +98,27 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage, } tgCfg := config.TelegramConfig{ - Enabled: true, - Token: c.Token, - Proxy: proxy, - APIServer: apiServer, - AllowFrom: ic.AllowFrom, - DMPolicy: ic.DMPolicy, - GroupPolicy: ic.GroupPolicy, - RequireMention: ic.RequireMention, - MentionMode: ic.MentionMode, - HistoryLimit: ic.HistoryLimit, - DMStream: ic.DMStream, - GroupStream: ic.GroupStream, - DraftTransport: ic.DraftTransport, - ReasoningStream: ic.ReasoningStream, - ReactionLevel: ic.ReactionLevel, - MediaMaxBytes: resolveMediaMaxBytes(ic), - LinkPreview: ic.LinkPreview, - BlockReply: ic.BlockReply, - ChatBehavior: ic.ChatBehavior, - ForceIPv4: ic.ForceIPv4, + Enabled: true, + Token: c.Token, + Proxy: proxy, + APIServer: apiServer, + AllowFrom: ic.AllowFrom, + DMPolicy: ic.DMPolicy, + GroupPolicy: ic.GroupPolicy, + RequireMention: ic.RequireMention, + MentionMode: ic.MentionMode, + HistoryLimit: ic.HistoryLimit, + DMStream: ic.DMStream, + GroupStream: ic.GroupStream, + DraftTransport: ic.DraftTransport, + ReasoningDelivery: ic.ReasoningDelivery, + ReasoningStream: ic.ReasoningStream, + ReactionLevel: ic.ReactionLevel, + MediaMaxBytes: resolveMediaMaxBytes(ic), + LinkPreview: ic.LinkPreview, + BlockReply: ic.BlockReply, + ChatBehavior: ic.ChatBehavior, + ForceIPv4: ic.ForceIPv4, } // DB instances default to "pairing" for groups (secure by default). diff --git a/internal/config/channel_reasoning_delivery.go b/internal/config/channel_reasoning_delivery.go new file mode 100644 index 00000000..9ae6dd19 --- /dev/null +++ b/internal/config/channel_reasoning_delivery.go @@ -0,0 +1,80 @@ +package config + +import "encoding/json" + +const ( + channelTypeTelegram = "telegram" + reasoningDeliveryKey = "reasoning_delivery" + legacyReasoningStreamKey = "reasoning_stream" + reasoningDeliveryOff = "off" + reasoningDeliveryStreamingOnly = "streaming_only" + reasoningDeliveryAlwaysBubbles = "always_bubbles" +) + +// NormalizeChannelInstanceConfigValue applies compatibility rewrites for +// channel-instance config payloads before they are persisted. +func NormalizeChannelInstanceConfigValue(channelType string, value any) any { + if channelType != channelTypeTelegram || value == nil { + return value + } + switch cfg := value.(type) { + case json.RawMessage: + return normalizeChannelConfigRaw(cfg) + case []byte: + return []byte(normalizeChannelConfigRaw(cfg)) + case map[string]any: + return normalizeTelegramReasoningDeliveryMap(cfg) + default: + return value + } +} + +func NormalizeChannelInstanceConfigRaw(channelType string, raw json.RawMessage) json.RawMessage { + if channelType != channelTypeTelegram { + return raw + } + return normalizeChannelConfigRaw(raw) +} + +func normalizeChannelConfigRaw(raw json.RawMessage) json.RawMessage { + if len(raw) == 0 { + return raw + } + var cfg map[string]any + if err := json.Unmarshal(raw, &cfg); err != nil { + return raw + } + cfg = normalizeTelegramReasoningDeliveryMap(cfg) + encoded, err := json.Marshal(cfg) + if err != nil { + return raw + } + return encoded +} + +func normalizeTelegramReasoningDeliveryMap(cfg map[string]any) map[string]any { + if cfg == nil { + return nil + } + _, hasMode := cfg[reasoningDeliveryKey] + _, hasLegacy := cfg[legacyReasoningStreamKey] + if !hasMode && !hasLegacy { + return cfg + } + cfg[reasoningDeliveryKey] = resolveReasoningDeliveryMode(cfg) + delete(cfg, legacyReasoningStreamKey) + return cfg +} + +func resolveReasoningDeliveryMode(cfg map[string]any) string { + if mode, ok := cfg[reasoningDeliveryKey].(string); ok { + switch mode { + case reasoningDeliveryOff, reasoningDeliveryStreamingOnly, reasoningDeliveryAlwaysBubbles: + return mode + } + } + if legacy, ok := cfg[legacyReasoningStreamKey].(bool); ok && !legacy { + return reasoningDeliveryOff + } + return reasoningDeliveryStreamingOnly +} diff --git a/internal/config/channel_reasoning_delivery_test.go b/internal/config/channel_reasoning_delivery_test.go new file mode 100644 index 00000000..077f725b --- /dev/null +++ b/internal/config/channel_reasoning_delivery_test.go @@ -0,0 +1,45 @@ +package config + +import ( + "encoding/json" + "testing" +) + +func TestNormalizeChannelInstanceConfigRaw_NormalizesTelegramReasoningDelivery(t *testing.T) { + raw := json.RawMessage(`{"reasoning_delivery":"always_bubbles","reasoning_stream":false,"dm_stream":false}`) + + got := NormalizeChannelInstanceConfigRaw("telegram", raw) + + var cfg map[string]any + if err := json.Unmarshal(got, &cfg); err != nil { + t.Fatalf("normalized config is invalid JSON: %v", err) + } + if cfg["reasoning_delivery"] != reasoningDeliveryAlwaysBubbles { + t.Fatalf("reasoning_delivery = %v", cfg["reasoning_delivery"]) + } + if _, ok := cfg["reasoning_stream"]; ok { + t.Fatalf("legacy reasoning_stream survived normalization: %s", got) + } +} + +func TestNormalizeChannelInstanceConfigValue_LegacyFalseMapsToOff(t *testing.T) { + got := NormalizeChannelInstanceConfigValue("telegram", map[string]any{ + "reasoning_stream": false, + }) + + cfg := got.(map[string]any) + if cfg["reasoning_delivery"] != reasoningDeliveryOff { + t.Fatalf("reasoning_delivery = %v, want off", cfg["reasoning_delivery"]) + } + if _, ok := cfg["reasoning_stream"]; ok { + t.Fatal("legacy reasoning_stream survived normalization") + } +} + +func TestNormalizeChannelInstanceConfigValue_IgnoresOtherChannels(t *testing.T) { + raw := json.RawMessage(`{"reasoning_stream":false}`) + got := NormalizeChannelInstanceConfigRaw("slack", raw) + if string(got) != string(raw) { + t.Fatalf("non-telegram config changed: %s", got) + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 35d60d7f..dfcdc445 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -48,26 +48,27 @@ type ChannelsConfig struct { } type TelegramConfig struct { - Enabled bool `json:"enabled"` - Token string `json:"token"` - Proxy string `json:"proxy,omitempty"` - APIServer string `json:"api_server,omitempty"` // custom Telegram Bot API server URL (e.g. "http://localhost:8081") - AllowFrom FlexibleStringSlice `json:"allow_from"` - DMPolicy string `json:"dm_policy,omitempty"` // "pairing" (default), "allowlist", "open", "disabled" - GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" - RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in groups (default true) - MentionMode string `json:"mention_mode,omitempty"` // "strict" (default) = only respond when mentioned; "yield" = respond unless another bot is mentioned - HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled) - DMStream *bool `json:"dm_stream,omitempty"` // enable streaming for DMs (default false) — edits placeholder progressively - GroupStream *bool `json:"group_stream,omitempty"` // enable streaming for groups (default false) — sends new message, edits progressively - DraftTransport *bool `json:"draft_transport,omitempty"` // use sendMessageDraft for DM streaming (default true) — stealth preview, no notifications per edit - ReasoningStream *bool `json:"reasoning_stream,omitempty"` // show reasoning as separate message when provider emits thinking events (default true) - ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full" — status emoji reactions - MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // max media download size in bytes (default 20MB) - LinkPreview *bool `json:"link_preview,omitempty"` // enable URL previews in messages (default true) - BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) - ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit) - ForceIPv4 bool `json:"force_ipv4,omitempty"` // force IPv4 for all Telegram API requests (use when IPv6 routing is broken) + Enabled bool `json:"enabled"` + Token string `json:"token"` + Proxy string `json:"proxy,omitempty"` + APIServer string `json:"api_server,omitempty"` // custom Telegram Bot API server URL (e.g. "http://localhost:8081") + AllowFrom FlexibleStringSlice `json:"allow_from"` + DMPolicy string `json:"dm_policy,omitempty"` // "pairing" (default), "allowlist", "open", "disabled" + GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" + RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in groups (default true) + MentionMode string `json:"mention_mode,omitempty"` // "strict" (default) = only respond when mentioned; "yield" = respond unless another bot is mentioned + HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled) + DMStream *bool `json:"dm_stream,omitempty"` // enable streaming for DMs (default false) — edits placeholder progressively + GroupStream *bool `json:"group_stream,omitempty"` // enable streaming for groups (default false) — sends new message, edits progressively + DraftTransport *bool `json:"draft_transport,omitempty"` // use sendMessageDraft for DM streaming (default true) — stealth preview, no notifications per edit + ReasoningDelivery string `json:"reasoning_delivery,omitempty"` // off, streaming_only, always_bubbles + ReasoningStream *bool `json:"reasoning_stream,omitempty"` // show reasoning as separate message when provider emits thinking events (default true) + ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full" — status emoji reactions + MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // max media download size in bytes (default 20MB) + LinkPreview *bool `json:"link_preview,omitempty"` // enable URL previews in messages (default true) + BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) + ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit) + ForceIPv4 bool `json:"force_ipv4,omitempty"` // force IPv4 for all Telegram API requests (use when IPv6 routing is broken) // Optional STT (Speech-to-Text) pipeline for voice/audio inbound messages. // When stt_proxy_url is set, audio/voice messages are transcribed before being forwarded to the agent. diff --git a/internal/gateway/methods/channel_instances.go b/internal/gateway/methods/channel_instances.go index 5058f0d7..df618d8b 100644 --- a/internal/gateway/methods/channel_instances.go +++ b/internal/gateway/methods/channel_instances.go @@ -9,6 +9,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -169,7 +170,7 @@ func (m *ChannelInstancesMethods) handleCreate(ctx context.Context, client *gate ChannelType: params.ChannelType, AgentID: agentID, Credentials: params.Credentials, - Config: params.Config, + Config: config.NormalizeChannelInstanceConfigRaw(params.ChannelType, params.Config), Enabled: enabled, } @@ -215,6 +216,7 @@ func (m *ChannelInstancesMethods) handleUpdate(ctx context.Context, client *gate slog.Warn("security.filtered_unknown_field", "field", k, "handler", "channels.instances.update") } } + m.normalizeChannelInstanceConfigUpdate(ctx, id, updates) if err := m.store.Update(ctx, id, updates); err != nil { slog.Error("channels.instances.update", "error", err) @@ -227,6 +229,20 @@ func (m *ChannelInstancesMethods) handleUpdate(ctx context.Context, client *gate client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"status": "updated"})) } +func (m *ChannelInstancesMethods) normalizeChannelInstanceConfigUpdate(ctx context.Context, id uuid.UUID, updates map[string]any) { + value, ok := updates["config"] + if !ok { + return + } + channelType, _ := updates["channel_type"].(string) + if channelType == "" { + if inst, err := m.store.Get(ctx, id); err == nil { + channelType = inst.ChannelType + } + } + updates["config"] = config.NormalizeChannelInstanceConfigValue(channelType, value) +} + func (m *ChannelInstancesMethods) handleDelete(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index afe8916f..3c978cb0 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -13,6 +13,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channelmemory" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/permissions" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -268,7 +269,7 @@ func (h *ChannelInstancesHandler) handleCreate(w http.ResponseWriter, r *http.Re ChannelType: body.ChannelType, AgentID: agentID, Credentials: body.Credentials, - Config: body.Config, + Config: config.NormalizeChannelInstanceConfigRaw(body.ChannelType, body.Config), Enabled: enabled, CreatedBy: userID, } @@ -317,6 +318,7 @@ func (h *ChannelInstancesHandler) handleUpdate(w http.ResponseWriter, r *http.Re // Allowlist: only permit known channel instance columns. updates = filterAllowedKeys(updates, channelInstanceAllowedFields) + h.normalizeChannelInstanceConfigUpdate(r.Context(), id, updates) if err := h.store.Update(r.Context(), id, updates); err != nil { slog.Error("channel_instances.update", "error", err) @@ -329,6 +331,20 @@ func (h *ChannelInstancesHandler) handleUpdate(w http.ResponseWriter, r *http.Re writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } +func (h *ChannelInstancesHandler) normalizeChannelInstanceConfigUpdate(ctx context.Context, id uuid.UUID, updates map[string]any) { + value, ok := updates["config"] + if !ok { + return + } + channelType, _ := updates["channel_type"].(string) + if channelType == "" { + if inst, err := h.store.Get(ctx, id); err == nil { + channelType = inst.ChannelType + } + } + updates["config"] = config.NormalizeChannelInstanceConfigValue(channelType, value) +} + func (h *ChannelInstancesHandler) handleDelete(w http.ResponseWriter, r *http.Request) { locale := store.LocaleFromContext(r.Context()) id, err := uuid.Parse(r.PathValue("id")) diff --git a/ui/desktop/frontend/src/components/channels/ChannelAdvancedTab.tsx b/ui/desktop/frontend/src/components/channels/ChannelAdvancedTab.tsx index cda9388e..d3f5489d 100644 --- a/ui/desktop/frontend/src/components/channels/ChannelAdvancedTab.tsx +++ b/ui/desktop/frontend/src/components/channels/ChannelAdvancedTab.tsx @@ -2,6 +2,7 @@ import { useState, useEffect, useCallback } from 'react' import { useTranslation } from 'react-i18next' import { ChannelFields } from './channel-field-renderer' import { configSchema, NETWORK_KEYS, LIMITS_KEYS, STREAMING_KEYS, BEHAVIOR_KEYS, ACCESS_KEYS } from './channel-schemas' +import { normalizeReasoningDeliveryConfig } from './reasoning-delivery-config' import type { ChannelInstanceData } from '../../types/channel' interface ChannelAdvancedTabProps { @@ -24,7 +25,7 @@ function getAdvancedFields(channelType: string) { } function deriveInitialValues(instance: ChannelInstanceData): Record { - const config = (instance.config ?? {}) as Record + const config = normalizeReasoningDeliveryConfig((instance.config ?? {}) as Record) const { groups: _groups, ...rest } = config return Object.fromEntries( Object.entries(rest).filter(([k]) => !ESSENTIAL_KEYS.has(k)), @@ -53,7 +54,7 @@ export function ChannelAdvancedTab({ instance, onUpdate }: ChannelAdvancedTabPro const cleanAdvanced = Object.fromEntries( Object.entries(values).filter(([, v]) => v !== undefined && v !== '' && v !== null), ) - const merged = { ...existingConfig, ...cleanAdvanced } + const merged = normalizeReasoningDeliveryConfig({ ...existingConfig, ...cleanAdvanced }) await onUpdate({ config: merged }) } catch { // toast shown by hook diff --git a/ui/desktop/frontend/src/components/channels/channel-schemas.ts b/ui/desktop/frontend/src/components/channels/channel-schemas.ts index 11c35633..cf96d5d1 100644 --- a/ui/desktop/frontend/src/components/channels/channel-schemas.ts +++ b/ui/desktop/frontend/src/components/channels/channel-schemas.ts @@ -1,5 +1,6 @@ // Per-channel-type field definitions for credentials and config. // Simplified from web UI — only telegram and discord supported in Lite edition. +import { reasoningDeliveryOptions } from './reasoning-delivery-config' export interface FieldDef { key: string @@ -72,7 +73,7 @@ export const configSchema: Record = { { key: 'dm_stream', label: 'DM Streaming', type: 'boolean', defaultValue: true, help: 'Stream response progressively in DMs' }, { key: 'group_stream', label: 'Group Streaming', type: 'boolean', defaultValue: false, help: 'Stream response progressively in groups' }, { key: 'draft_transport', label: 'Draft Preview', type: 'boolean', defaultValue: true, help: 'Stealth draft preview for streaming in DMs (requires DM Streaming)' }, - { key: 'reasoning_stream', label: 'Show Reasoning', type: 'boolean', defaultValue: true, help: 'Display AI thinking before the answer (requires streaming)' }, + { key: 'reasoning_delivery', label: 'Show Reasoning', type: 'select', options: reasoningDeliveryOptions, defaultValue: 'streaming_only', help: 'Choose how model reasoning is shown in channel messages.' }, { key: 'reaction_level', label: 'Reaction Level', type: 'select', options: reactionLevelOptions, defaultValue: 'full' }, { key: 'media_max_mb', label: 'Max Media Size (MB)', type: 'number', defaultValue: 20, help: 'Default: 20 MB. Increase when using local Bot API server.' }, { key: 'link_preview', label: 'Link Preview', type: 'boolean', defaultValue: true }, @@ -98,6 +99,6 @@ export const ESSENTIAL_CONFIG_KEYS: Record = { // Advanced config grouping keys (for ChannelAdvancedDialog) export const NETWORK_KEYS = new Set(['api_server', 'proxy']) export const LIMITS_KEYS = new Set(['history_limit', 'media_max_mb']) -export const STREAMING_KEYS = new Set(['dm_stream', 'group_stream', 'draft_transport', 'reasoning_stream']) +export const STREAMING_KEYS = new Set(['dm_stream', 'group_stream', 'draft_transport', 'reasoning_delivery']) export const BEHAVIOR_KEYS = new Set(['reaction_level', 'link_preview', 'block_reply']) export const ACCESS_KEYS = new Set(['allow_from']) diff --git a/ui/desktop/frontend/src/components/channels/reasoning-delivery-config.ts b/ui/desktop/frontend/src/components/channels/reasoning-delivery-config.ts new file mode 100644 index 00000000..34bbbaed --- /dev/null +++ b/ui/desktop/frontend/src/components/channels/reasoning-delivery-config.ts @@ -0,0 +1,20 @@ +export const reasoningDeliveryOptions = [ + { value: 'streaming_only', label: 'Streaming only' }, + { value: 'always_bubbles', label: 'Always as bubbles' }, + { value: 'off', label: 'Off' }, +] + +export function resolveReasoningDeliveryValue(config: Record | undefined): string { + const mode = typeof config?.reasoning_delivery === 'string' ? config.reasoning_delivery : '' + if (mode === 'streaming_only' || mode === 'always_bubbles' || mode === 'off') return mode + return config?.reasoning_stream === false ? 'off' : 'streaming_only' +} + +export function normalizeReasoningDeliveryConfig>(config: T): T { + const next = { ...config } as Record + if (next.reasoning_delivery !== undefined || next.reasoning_stream !== undefined) { + next.reasoning_delivery = resolveReasoningDeliveryValue(next) + delete next.reasoning_stream + } + return next as T +} diff --git a/ui/web/src/i18n/locales/en/channels.json b/ui/web/src/i18n/locales/en/channels.json index cd8cceb3..69590ad6 100644 --- a/ui/web/src/i18n/locales/en/channels.json +++ b/ui/web/src/i18n/locales/en/channels.json @@ -338,9 +338,9 @@ "label": "Draft Preview", "help": "Use stealth draft preview for answer stream in DMs — no notification per edit (requires DM Streaming)" }, - "reasoning_stream": { + "reasoning_delivery": { "label": "Show Reasoning", - "help": "Display AI thinking as a separate message before the answer (requires streaming)" + "help": "Choose how model reasoning is shown in channel messages." }, "reaction_level": { "label": "Reaction Level" }, "media_max_mb": { "label": "Max Media Size (MB)" }, @@ -436,6 +436,11 @@ "minimal": "Minimal", "full": "Full" }, + "reasoning_delivery": { + "streaming_only": "Streaming only", + "always_bubbles": "Always as bubbles", + "off": "Off" + }, "domain": { "lark": "Lark (Global)", "feishu": "Feishu (China)" @@ -492,9 +497,9 @@ "label": "Draft Preview", "help": "Use stealth draft preview for answer stream in DMs — no notification per edit (requires DM Streaming)" }, - "reasoning_stream": { + "reasoning_delivery": { "label": "Show Reasoning", - "help": "Display AI thinking as a separate message before the answer (requires streaming)" + "help": "Choose how model reasoning is shown in channel messages." } }, "fields": { diff --git a/ui/web/src/i18n/locales/en/config.json b/ui/web/src/i18n/locales/en/config.json index dcf569de..d81125ad 100644 --- a/ui/web/src/i18n/locales/en/config.json +++ b/ui/web/src/i18n/locales/en/config.json @@ -325,8 +325,11 @@ "channels.streaming": "Streaming", "channels.draftTransport": "Draft Preview", "channels.draftTransportHint": "Stealth streaming preview in DMs (no notifications per edit)", - "channels.reasoningStream": "Show Reasoning", - "channels.reasoningStreamHint": "Display AI thinking process as a separate message", + "channels.reasoningDelivery": "Show Reasoning", + "channels.reasoningDeliveryHint": "Choose how model reasoning appears in channel messages. Always as bubbles works even when live streaming is off.", + "channels.reasoningStreamingOnly": "Streaming only", + "channels.reasoningAlwaysBubbles": "Always as bubbles", + "channels.reasoningOff": "Off", "channels.reactionLevel": "Reaction Level", "channels.connectionMode": "Connection Mode", "channels.reactionOff": "Off", diff --git a/ui/web/src/i18n/locales/vi/channels.json b/ui/web/src/i18n/locales/vi/channels.json index bd0db37c..5438e85e 100644 --- a/ui/web/src/i18n/locales/vi/channels.json +++ b/ui/web/src/i18n/locales/vi/channels.json @@ -304,7 +304,7 @@ "dm_stream": { "label": "Stream DM", "help": "Truyền phản hồi liên tục trong DM" }, "group_stream": { "label": "Stream nhóm", "help": "Truyền phản hồi liên tục trong nhóm" }, "draft_transport": { "label": "Xem trước nháp", "help": "Dùng bản nháp ẩn cho stream trả lời trong DM — không thông báo mỗi lần chỉnh sửa (cần bật Stream DM)" }, - "reasoning_stream": { "label": "Hiện suy luận", "help": "Hiển thị quá trình suy nghĩ của AI thành tin nhắn riêng trước câu trả lời (cần bật streaming)" }, + "reasoning_delivery": { "label": "Hiện suy luận", "help": "Chọn cách hiển thị suy luận của model trong tin nhắn channel." }, "reaction_level": { "label": "Mức độ reaction" }, "media_max_mb": { "label": "Kích thước media tối đa (MB)" }, "link_preview": { "label": "Xem trước liên kết" }, @@ -363,6 +363,11 @@ "minimal": "Tối thiểu", "full": "Đầy đủ" }, + "reasoning_delivery": { + "streaming_only": "Chỉ khi streaming", + "always_bubbles": "Luôn thành bubble", + "off": "Tắt" + }, "domain": { "lark": "Lark (Quốc tế)", "feishu": "Feishu (Trung Quốc)" @@ -410,7 +415,7 @@ "dm_stream": { "label": "Stream DM", "help": "Truyền phản hồi liên tục trong DM" }, "group_stream": { "label": "Stream Nhóm", "help": "Truyền phản hồi liên tục trong nhóm" }, "draft_transport": { "label": "Xem trước nháp", "help": "Dùng bản nháp ẩn cho stream trả lời trong DM — không thông báo mỗi lần chỉnh sửa (cần bật Stream DM)" }, - "reasoning_stream": { "label": "Hiện suy luận", "help": "Hiển thị quá trình suy nghĩ của AI thành tin nhắn riêng trước câu trả lời (cần bật streaming)" } + "reasoning_delivery": { "label": "Hiện suy luận", "help": "Chọn cách hiển thị suy luận của model trong tin nhắn channel." } }, "fields": { "groupPolicy": "Chính sách nhóm", diff --git a/ui/web/src/i18n/locales/vi/config.json b/ui/web/src/i18n/locales/vi/config.json index 09e36d4b..0a0e88f3 100644 --- a/ui/web/src/i18n/locales/vi/config.json +++ b/ui/web/src/i18n/locales/vi/config.json @@ -324,8 +324,11 @@ "channels.streaming": "Streaming", "channels.draftTransport": "Xem trước nháp", "channels.draftTransportHint": "Xem trước streaming trong DM (không thông báo mỗi lần chỉnh sửa)", - "channels.reasoningStream": "Hiện suy luận", - "channels.reasoningStreamHint": "Hiển thị quá trình suy nghĩ của AI thành tin nhắn riêng", + "channels.reasoningDelivery": "Hiện suy luận", + "channels.reasoningDeliveryHint": "Chọn cách suy luận của model xuất hiện trong tin nhắn channel. Luôn thành bubble hoạt động kể cả khi tắt live streaming.", + "channels.reasoningStreamingOnly": "Chỉ khi streaming", + "channels.reasoningAlwaysBubbles": "Luôn thành bubble", + "channels.reasoningOff": "Tắt", "channels.reactionLevel": "Mức phản ứng", "channels.connectionMode": "Chế độ kết nối", "channels.reactionOff": "Tắt", diff --git a/ui/web/src/i18n/locales/zh/channels.json b/ui/web/src/i18n/locales/zh/channels.json index 484b7db6..3550dcad 100644 --- a/ui/web/src/i18n/locales/zh/channels.json +++ b/ui/web/src/i18n/locales/zh/channels.json @@ -304,7 +304,7 @@ "dm_stream": { "label": "私聊流式传输", "help": "在私聊中逐步传输回复" }, "group_stream": { "label": "群聊流式传输", "help": "在群聊中逐步传输回复" }, "draft_transport": { "label": "草稿预览", "help": "在私聊中使用隐形草稿预览回答流 — 每次编辑无通知（需启用私聊流式传输）" }, - "reasoning_stream": { "label": "显示推理", "help": "将AI思考过程显示为单独的消息（需启用流式传输）" }, + "reasoning_delivery": { "label": "显示推理", "help": "选择如何在频道消息中显示模型推理。" }, "reaction_level": { "label": "表情回应级别" }, "media_max_mb": { "label": "最大媒体大小 (MB)" }, "link_preview": { "label": "链接预览" }, @@ -363,6 +363,11 @@ "minimal": "最小", "full": "全部" }, + "reasoning_delivery": { + "streaming_only": "仅流式传输时", + "always_bubbles": "始终作为气泡", + "off": "关闭" + }, "domain": { "lark": "Lark（国际版）", "feishu": "飞书（中国版）" @@ -410,7 +415,7 @@ "dm_stream": { "label": "私聊流式传输", "help": "在私聊中逐步传输回复" }, "group_stream": { "label": "群聊流式传输", "help": "在群聊中逐步传输回复" }, "draft_transport": { "label": "草稿预览", "help": "在私聊中使用隐形草稿预览回答流 — 每次编辑无通知（需启用私聊流式传输）" }, - "reasoning_stream": { "label": "显示推理", "help": "将AI思考过程显示为单独的消息（需启用流式传输）" } + "reasoning_delivery": { "label": "显示推理", "help": "选择如何在频道消息中显示模型推理。" } }, "fields": { "groupPolicy": "群组策略", diff --git a/ui/web/src/i18n/locales/zh/config.json b/ui/web/src/i18n/locales/zh/config.json index 4789909c..14895c05 100644 --- a/ui/web/src/i18n/locales/zh/config.json +++ b/ui/web/src/i18n/locales/zh/config.json @@ -324,8 +324,11 @@ "channels.streaming": "流式传输", "channels.draftTransport": "草稿预览", "channels.draftTransportHint": "私聊中静默流式预览（每次编辑无通知）", - "channels.reasoningStream": "显示推理", - "channels.reasoningStreamHint": "将AI思考过程显示为单独的消息", + "channels.reasoningDelivery": "显示推理", + "channels.reasoningDeliveryHint": "选择模型推理在频道消息中的显示方式。始终作为气泡在关闭实时流式传输时也可用。", + "channels.reasoningStreamingOnly": "仅流式传输时", + "channels.reasoningAlwaysBubbles": "始终作为气泡", + "channels.reasoningOff": "关闭", "channels.reactionLevel": "反应级别", "channels.connectionMode": "连接模式", "channels.reactionOff": "关闭", diff --git a/ui/web/src/pages/channels/channel-detail/channel-advanced-dialog.tsx b/ui/web/src/pages/channels/channel-detail/channel-advanced-dialog.tsx index d6c6d169..df858fc8 100644 --- a/ui/web/src/pages/channels/channel-detail/channel-advanced-dialog.tsx +++ b/ui/web/src/pages/channels/channel-detail/channel-advanced-dialog.tsx @@ -11,6 +11,7 @@ import { import { ConfigGroupHeader } from "@/components/shared/config-group-header"; import { ChannelFields } from "../channel-fields"; import { configSchema } from "../channel-schemas"; +import { normalizeReasoningDeliveryConfig } from "../reasoning-delivery-config"; import type { ChannelInstanceData } from "@/types/channel"; interface ChannelAdvancedDialogProps { @@ -24,7 +25,7 @@ const ESSENTIAL_CONFIG_KEYS = new Set(["dm_policy", "group_policy", "require_men const NETWORK_KEYS = new Set(["api_server", "proxy", "domain", "connection_mode", "webhook_port", "webhook_path", "webhook_url"]); const LIMITS_KEYS = new Set(["history_limit", "media_max_mb", "text_chunk_limit"]); -const STREAMING_KEYS = new Set(["dm_stream", "group_stream", "draft_transport", "reasoning_stream", "native_stream", "debounce_delay", "thread_ttl"]); +const STREAMING_KEYS = new Set(["dm_stream", "group_stream", "draft_transport", "reasoning_delivery", "native_stream", "debounce_delay", "thread_ttl"]); const BEHAVIOR_KEYS = new Set(["reaction_level", "link_preview", "block_reply", "render_mode", "topic_session_mode"]); const ACCESS_KEYS = new Set(["allow_from", "group_allow_from"]); @@ -41,7 +42,7 @@ function getAdvancedFields(channelType: string) { } function deriveInitialValues(instance: ChannelInstanceData): Record { - const config = (instance.config ?? {}) as Record; + const config = normalizeReasoningDeliveryConfig((instance.config ?? {}) as Record); // Only keep advanced keys (exclude essential + groups) return Object.fromEntries( Object.entries(config).filter(([k]) => !ESSENTIAL_CONFIG_KEYS.has(k) && k !== "groups"), @@ -79,7 +80,7 @@ export function ChannelAdvancedDialog({ Object.entries(values).filter(([, v]) => v !== undefined && v !== "" && v !== null), ); // Merge: preserve essential keys and groups from existing, overwrite advanced keys - const merged = { ...existingConfig, ...cleanAdvanced }; + const merged = normalizeReasoningDeliveryConfig({ ...existingConfig, ...cleanAdvanced }); await onUpdate({ config: merged }); onOpenChange(false); } catch { // toast shown by hook diff --git a/ui/web/src/pages/channels/channel-schemas.test.ts b/ui/web/src/pages/channels/channel-schemas.test.ts index 6ef16a5a..d1c90181 100644 --- a/ui/web/src/pages/channels/channel-schemas.test.ts +++ b/ui/web/src/pages/channels/channel-schemas.test.ts @@ -1,5 +1,28 @@ import { describe, it, expect } from "vitest"; import { configSchema } from "./channel-schemas"; +import { normalizeReasoningDeliveryConfig, resolveReasoningDeliveryValue } from "./reasoning-delivery-config"; + +describe("telegram configSchema", () => { + const telegramConfig = configSchema["telegram"]!; + + it("uses reasoning_delivery mode instead of legacy reasoning_stream", () => { + const reasoningDelivery = telegramConfig.find((f) => f.key === "reasoning_delivery"); + expect(reasoningDelivery).toBeDefined(); + expect(reasoningDelivery!.type).toBe("select"); + expect(reasoningDelivery!.defaultValue).toBe("streaming_only"); + expect(reasoningDelivery!.help).not.toMatch(/requires streaming/i); + expect(reasoningDelivery!.options!.map((o) => o.value)).toEqual(["streaming_only", "always_bubbles", "off"]); + expect(telegramConfig.find((f) => f.key === "reasoning_stream")).toBeUndefined(); + }); + + it("normalizes legacy reasoning_stream into explicit reasoning_delivery", () => { + expect(resolveReasoningDeliveryValue({ reasoning_stream: false })).toBe("off"); + expect(resolveReasoningDeliveryValue({ reasoning_stream: true })).toBe("streaming_only"); + expect(normalizeReasoningDeliveryConfig({ reasoning_delivery: "always_bubbles", reasoning_stream: false })).toEqual({ + reasoning_delivery: "always_bubbles", + }); + }); +}); describe("pancake configSchema", () => { const pancakeConfig = configSchema["pancake"]!; diff --git a/ui/web/src/pages/channels/channel-schemas.ts b/ui/web/src/pages/channels/channel-schemas.ts index f8b0bf72..41b30f44 100644 --- a/ui/web/src/pages/channels/channel-schemas.ts +++ b/ui/web/src/pages/channels/channel-schemas.ts @@ -1,5 +1,6 @@ // Per-channel-type field definitions for credentials and config. // Used by the form dialog to render proper UI fields instead of raw JSON. +import { reasoningDeliveryOptions } from "./reasoning-delivery-config"; export interface FieldDef { key: string; @@ -137,7 +138,7 @@ export const configSchema: Record = { { key: "dm_stream", label: "DM Streaming", type: "boolean", defaultValue: true, help: "Stream response progressively in DMs" }, { key: "group_stream", label: "Group Streaming", type: "boolean", defaultValue: false, help: "Stream response progressively in groups" }, { key: "draft_transport", label: "Draft Preview", type: "boolean", defaultValue: true, help: "Use stealth draft preview for answer stream in DMs — no notification per edit (requires DM Streaming)" }, - { key: "reasoning_stream", label: "Show Reasoning", type: "boolean", defaultValue: true, help: "Display AI thinking as a separate message before the answer (requires streaming)" }, + { key: "reasoning_delivery", label: "Show Reasoning", type: "select", options: reasoningDeliveryOptions, defaultValue: "streaming_only", help: "Choose how model reasoning is shown in channel messages." }, { key: "reaction_level", label: "Reaction Level", type: "select", options: [{ value: "off", label: "Off" }, { value: "minimal", label: "Minimal" }, { value: "full", label: "Full" }], defaultValue: "full" }, { key: "media_max_mb", label: "Max Media Size (MB)", type: "number", defaultValue: 20, help: "Default: 20 MB (cloud API). Increase when using local Bot API server." }, { key: "link_preview", label: "Link Preview", type: "boolean", defaultValue: true }, diff --git a/ui/web/src/pages/channels/reasoning-delivery-config.ts b/ui/web/src/pages/channels/reasoning-delivery-config.ts new file mode 100644 index 00000000..9b4d2974 --- /dev/null +++ b/ui/web/src/pages/channels/reasoning-delivery-config.ts @@ -0,0 +1,28 @@ +export const REASONING_DELIVERY_VALUES = ["streaming_only", "always_bubbles", "off"] as const; + +export type ReasoningDeliveryMode = (typeof REASONING_DELIVERY_VALUES)[number]; + +export const reasoningDeliveryOptions: { value: ReasoningDeliveryMode; label: string }[] = [ + { value: "streaming_only", label: "Streaming only" }, + { value: "always_bubbles", label: "Always as bubbles" }, + { value: "off", label: "Off" }, +]; + +export function resolveReasoningDeliveryValue(config: Record | undefined): ReasoningDeliveryMode { + const mode = typeof config?.reasoning_delivery === "string" ? config.reasoning_delivery : ""; + if (isReasoningDeliveryMode(mode)) return mode; + return config?.reasoning_stream === false ? "off" : "streaming_only"; +} + +export function normalizeReasoningDeliveryConfig>(config: T): T { + const next = { ...config } as Record; + if (next.reasoning_delivery !== undefined || next.reasoning_stream !== undefined) { + next.reasoning_delivery = resolveReasoningDeliveryValue(next); + delete next.reasoning_stream; + } + return next as T; +} + +function isReasoningDeliveryMode(value: string): value is ReasoningDeliveryMode { + return (REASONING_DELIVERY_VALUES as readonly string[]).includes(value); +} diff --git a/ui/web/src/pages/config/sections/channels-section.tsx b/ui/web/src/pages/config/sections/channels-section.tsx index 99b8a5e8..b27bd3b7 100644 --- a/ui/web/src/pages/config/sections/channels-section.tsx +++ b/ui/web/src/pages/config/sections/channels-section.tsx @@ -9,6 +9,7 @@ import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/com import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Badge } from "@/components/ui/badge"; import { MultiUserPicker } from "@/components/shared/multi-user-picker"; +import { normalizeReasoningDeliveryConfig, resolveReasoningDeliveryValue } from "@/pages/channels/reasoning-delivery-config"; type ChannelsData = Record; @@ -67,13 +68,14 @@ export function ChannelsSection({ data, onSave, saving }: Props) { // Strip masked secret fields const toSave: ChannelsData = {}; for (const [ch, val] of Object.entries(draft)) { - const copy = { ...val }; + const copy = normalizeReasoningDeliveryConfig({ ...val }); const meta = CHANNEL_META[ch]; if (meta?.secretField && isSecret(copy[meta.secretField])) { delete copy[meta.secretField]; } // Also strip other known secret fields for (const key of Object.keys(copy)) { + if (copy[key] === undefined) delete copy[key]; if (isSecret(copy[key])) delete copy[key]; } toSave[ch] = copy; @@ -227,18 +229,21 @@ export function ChannelsSection({ data, onSave, saving }: Props) { /> )} - {(!!chData.dm_stream || !!chData.group_stream) && ( -
-
-

{t("channels.reasoningStream")}

-

{t("channels.reasoningStreamHint")}

-
- updateChannel(ch, { reasoning_stream: v })} - /> -
- )} +
+ + +

{t("channels.reasoningDeliveryHint")}

+
{chData.reaction_level !== undefined && (