fix(channels): add reasoning delivery modes (#135)

This commit is contained in:
Duy /zuey/ authored and GitHub committed 2026-06-03 11:01:14 +07:00
1 parent 3e7876a4b4
commit 53ef912441
35 files changed
+1137 -148

No files matched your search

+9 -6
View File
@@ -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
+14
View File
@@ -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
+14
View File
@@ -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.
+22
View File
@@ -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
+11
View File
@@ -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 != "" {
@@ -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)
}
}
+6
View File
@@ -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
@@ -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")
+73 -14
View File
@@ -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 <think> — update reasoning stream, wait for close
// Still inside <think> — 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()
}
}
+27 -24
View File
@@ -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 <think> 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 <think> tags (skip re-parsing)
reasoningBubbles *reasoningBubbleBuffer
reasoningBubbleTimer *time.Timer
}
// Manager manages all registered channels, handling their lifecycle
+183
View File
@@ -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)
}
}
}
+60
View File
@@ -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
}
@@ -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": "<think>hidden</think>visible"})
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": "<think>tagged reasoning</think>tagged 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 }
+25 -1
View File
@@ -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)
}
+6 -5
View File
@@ -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).
+41 -39
View File
@@ -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).
@@ -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
}
@@ -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)
}
}
+21 -20
View File
@@ -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.
+17 -1
View File
@@ -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 {
+17 -1
View File
@@ -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"))
@@ -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<string, unknown> {
const config = (instance.config ?? {}) as Record<string, unknown>
const config = normalizeReasoningDeliveryConfig((instance.config ?? {}) as Record<string, unknown>)
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
@@ -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<string, FieldDef[]> = {
{ 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<string, string[]> = {
// 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'])
@@ -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<string, unknown> | 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<T extends Record<string, unknown>>(config: T): T {
const next = { ...config } as Record<string, unknown>
if (next.reasoning_delivery !== undefined || next.reasoning_stream !== undefined) {
next.reasoning_delivery = resolveReasoningDeliveryValue(next)
delete next.reasoning_stream
}
return next as T
}
+9 -4
View File
@@ -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": {
+5 -2
View File
@@ -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",
+7 -2
View File
@@ -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",
+5 -2
View File
@@ -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",
+7 -2
View File
@@ -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": "群组策略",
+5 -2
View File
@@ -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": "关闭",
@@ -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<string, unknown> {
const config = (instance.config ?? {}) as Record<string, unknown>;
const config = normalizeReasoningDeliveryConfig((instance.config ?? {}) as Record<string, unknown>);
// 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
@@ -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"]!;
+2 -1
View File
@@ -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<string, FieldDef[]> = {
{ 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 },
@@ -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<string, unknown> | 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<T extends Record<string, unknown>>(config: T): T {
const next = { ...config } as Record<string, unknown>;
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);
}
@@ -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<string, any>;
@@ -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) {
/>
</div>
)}
{(!!chData.dm_stream || !!chData.group_stream) && (
<div className="rounded-md border px-3 py-2 flex items-center justify-between">
<div>
<p className="text-sm font-medium">{t("channels.reasoningStream")}</p>
<p className="text-xs text-muted-foreground">{t("channels.reasoningStreamHint")}</p>
</div>
<Switch
checked={chData.reasoning_stream !== false}
onCheckedChange={(v) => updateChannel(ch, { reasoning_stream: v })}
/>
</div>
)}
<div className="rounded-md border px-3 py-2 grid gap-1.5 sm:col-span-2">
<Label>{t("channels.reasoningDelivery")}</Label>
<Select
value={resolveReasoningDeliveryValue(chData)}
onValueChange={(v) => updateChannel(ch, { reasoning_delivery: v, reasoning_stream: undefined })}
>
<SelectTrigger><SelectValue /></SelectTrigger>
<SelectContent>
<SelectItem value="streaming_only">{t("channels.reasoningStreamingOnly")}</SelectItem>
<SelectItem value="always_bubbles">{t("channels.reasoningAlwaysBubbles")}</SelectItem>
<SelectItem value="off">{t("channels.reasoningOff")}</SelectItem>
</SelectContent>
</Select>
<p className="text-xs text-muted-foreground">{t("channels.reasoningDeliveryHint")}</p>
</div>
</div>
{chData.reaction_level !== undefined && (
<div className="grid gap-1.5">