diff --git a/internal/channels/channel.go b/internal/channels/channel.go index def6a188..a867ba7f 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -100,9 +100,21 @@ type StreamingChannel interface { // Channels may choose to always stream for DMs while gating group streaming // behind config (e.g. Telegram uses sendMessageDraft for DMs). StreamEnabled(isGroup bool) bool - OnStreamStart(ctx context.Context, chatID string) error - OnChunkEvent(ctx context.Context, chatID string, fullText string) error - OnStreamEnd(ctx context.Context, chatID string, finalText string) error + // CreateStream creates a new per-run streaming handle for the given chatID. + // The returned ChannelStream is stored on RunContext so each concurrent run + // gets its own stream — eliminates the chatID-keyed sync.Map collision bug. + // firstStream: true for the first stream in a run (may become reasoning lane — + // must use message transport so it persists as a real message). false for + // subsequent streams (answer lane — may use draft transport for stealth preview). + CreateStream(ctx context.Context, chatID string, firstStream bool) (ChannelStream, error) + // FinalizeStream is called after the stream has been stopped to hand off + // the stream's messageID (if any) back to the channel's placeholder map + // so that Send() can edit it with the final formatted response. + FinalizeStream(ctx context.Context, chatID string, stream ChannelStream) + // ReasoningStreamEnabled returns whether reasoning should be shown as a + // separate message. Default: true. Channels that don't support lanes can + // return false to skip reasoning routing. + ReasoningStreamEnabled() bool } // BlockReplyChannel is optionally implemented by channels that override diff --git a/internal/channels/events.go b/internal/channels/events.go index 6148e4c8..4daf7095 100644 --- a/internal/channels/events.go +++ b/internal/channels/events.go @@ -33,19 +33,63 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { if sc, ok := ch.(StreamingChannel); ok { switch eventType { case protocol.AgentEventRunStarted: - if err := sc.OnStreamStart(ctx, rc.ChatID); err != nil { + stream, err := sc.CreateStream(ctx, rc.ChatID, true) + if err != nil { slog.Debug("stream start failed", "channel", rc.ChannelName, "error", err) + } else { + rc.mu.Lock() + rc.stream = stream + rc.mu.Unlock() + } + case protocol.ChatEventThinking: + // Accumulate thinking/reasoning content and route to the current stream. + // The stream created on run.started becomes the "reasoning lane": + // - DMs: edits the "Thinking..." placeholder with reasoning text + // - Groups: edits a fresh message with reasoning text + // 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 != "" { + rc.mu.Lock() + rc.thinkingBuffer += content + rc.hasThinking = true + thinkText := rc.thinkingBuffer + currentStream := rc.stream + rc.mu.Unlock() + if currentStream != nil { + currentStream.Update(ctx, formatReasoningPreview(thinkText)) + } } case protocol.AgentEventToolCall: // Agent is executing a tool — mark tool phase so the next chunk // (new LLM iteration) resets the stream buffer. - // Also clear the current DraftStream so the next iteration starts - // a fresh streaming message (matching TS onAssistantMessageStart pattern). + // Stop the current stream (reasoning or answer) and finalize only + // the answer stream (reasoning messages stay visible). rc.mu.Lock() + // Capture current state before resetting for next iteration. + wasReasoningStream := rc.hasThinking && !rc.thinkingDone + currentStream := rc.stream + rc.stream = nil rc.inToolPhase = true + rc.thinkingDone = false // allow new thinking in next iteration + rc.thinkingBuffer = "" // reset thinking buffer for new iteration + rc.hasThinking = false // new iteration starts fresh + rc.tagParseSkipped = false // re-enable tag parsing for next iteration rc.mu.Unlock() - if err := sc.OnStreamEnd(ctx, rc.ChatID, ""); err != nil { - slog.Debug("stream tool-phase end failed", "channel", rc.ChannelName, "error", err) + if currentStream != nil { + if err := currentStream.Stop(ctx); err != nil { + slog.Debug("stream tool-phase stop failed", "channel", rc.ChannelName, "error", err) + } + // Only finalize answer streams (hand off messageID to Send()). + // Reasoning streams stay as visible messages — don't put their + // messageID into placeholders or it would confuse Send(). + if !wasReasoningStream { + sc.FinalizeStream(ctx, rc.ChatID, currentStream) + } } // Show tool status in streaming preview (edit placeholder with tool name). @@ -63,39 +107,130 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { } case protocol.ChatEventChunk: // Accumulate chunk deltas into full text. - // When entering a new LLM iteration (first chunk after tool.call), - // reset the buffer so we don't concatenate text from previous iterations. content := extractPayloadString(payload, "content") if content != "" { rc.mu.Lock() - if rc.inToolPhase { - // New LLM iteration — reset buffer and start fresh stream + needNewStream := rc.inToolPhase + if needNewStream { rc.streamBuffer = "" rc.inToolPhase = false - rc.mu.Unlock() - // Create new DraftStream for this iteration - if err := sc.OnStreamStart(ctx, rc.ChatID); err != nil { - slog.Debug("stream restart failed", "channel", rc.ChannelName, "error", err) - } - rc.mu.Lock() } + + // Fallback tag parsing: for providers that embed thinking + // in the content stream (DeepSeek-via-OpenRouter, Qwen, some Ollama models). + // Only activates when no native ChatEventThinking was received. + if !rc.hasThinking && !rc.thinkingDone && !rc.tagParseSkipped { + candidate := rc.streamBuffer + content + split := SplitThinkTags(candidate) + if split.Thinking != "" { + // Found think tags — commit to buffer and route to reasoning lane + 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 + rc.mu.Unlock() + if currentStream != nil { + currentStream.Update(ctx, formatReasoningPreview(thinkText)) + } + break + } + // Tag closed — transition to answer + rc.thinkingDone = true + rc.streamBuffer = split.Answer + reasoningStream := currentStream + rc.mu.Unlock() + + // Stop reasoning stream + if reasoningStream != nil { + _ = reasoningStream.Stop(ctx) + } + // Create answer stream + stream, err := sc.CreateStream(ctx, rc.ChatID, false) + if err != nil { + slog.Debug("stream restart after think-tag failed", "channel", rc.ChannelName, "error", err) + } else { + rc.mu.Lock() + rc.stream = stream + rc.mu.Unlock() + } + // Update answer stream with extracted answer content + if split.Answer != "" { + rc.mu.Lock() + currentStream = rc.stream + rc.mu.Unlock() + if currentStream != nil { + currentStream.Update(ctx, split.Answer) + } + } + break + } + // No think tags found — mark as skipped so we don't re-parse. + // Don't commit to streamBuffer here — the normal flow below appends content. + rc.tagParseSkipped = true + } + + // Reasoning→answer transition: first chunk after native thinking events. + // Stop the reasoning stream (keep message visible) and create a + // new stream for the answer lane. + needTransition := rc.hasThinking && !rc.thinkingDone + if needTransition { + rc.thinkingDone = true + rc.streamBuffer = "" // fresh answer buffer + } + reasoningStream := rc.stream + rc.mu.Unlock() + + // Finalize reasoning stream (stop editing, keep message) + if needTransition && reasoningStream != nil { + _ = reasoningStream.Stop(ctx) + // Don't call FinalizeStream — reasoning messageID should NOT + // go into placeholders. Send() must edit the answer message. + } + + // Create fresh stream for answer (or new tool iteration) + if needNewStream || needTransition { + stream, err := sc.CreateStream(ctx, rc.ChatID, false) + if err != nil { + slog.Debug("stream restart failed", "channel", rc.ChannelName, "error", err) + } else { + rc.mu.Lock() + rc.stream = stream + rc.mu.Unlock() + } + } + + rc.mu.Lock() rc.streamBuffer += content fullText := rc.streamBuffer + currentStream := rc.stream rc.mu.Unlock() - if err := sc.OnChunkEvent(ctx, rc.ChatID, fullText); err != nil { - slog.Debug("stream chunk failed", "channel", rc.ChannelName, "error", err) + if currentStream != nil { + currentStream.Update(ctx, fullText) } } case protocol.AgentEventRunCompleted: rc.mu.Lock() - finalText := rc.streamBuffer + currentStream := rc.stream + rc.stream = nil rc.mu.Unlock() - if err := sc.OnStreamEnd(ctx, rc.ChatID, finalText); err != nil { - slog.Debug("stream end failed", "channel", rc.ChannelName, "error", err) + if currentStream != nil { + if err := currentStream.Stop(ctx); err != nil { + slog.Debug("stream end failed", "channel", rc.ChannelName, "error", err) + } + sc.FinalizeStream(ctx, rc.ChatID, currentStream) } case protocol.AgentEventRunFailed: - // Clean up streaming state - _ = sc.OnStreamEnd(ctx, rc.ChatID, "") + // Clean up streaming state on failure + rc.mu.Lock() + currentStream := rc.stream + rc.stream = nil + rc.mu.Unlock() + if currentStream != nil { + _ = currentStream.Stop(ctx) + } } } @@ -279,6 +414,23 @@ func formatToolStatus(toolName string) string { return "🔧 Running " + toolName + "..." } +// formatReasoningPreview formats accumulated thinking text for display as a +// streaming reasoning message. Uses markdown italic prefix so channels that +// convert markdown (Telegram, Slack) show "Reasoning:" in italics. +// Truncated to 4096 runes (Telegram limit, rune-safe for CJK/emoji). +func formatReasoningPreview(thinking string) string { + if thinking == "" { + return "" + } + const maxRunes = 4096 + text := "_Reasoning:_\n" + thinking + runes := []rune(text) + if len(runes) > maxRunes { + text = string(runes[:maxRunes-3]) + "..." + } + return text +} + // resolveToolReactionStatus maps a tool name to a reaction status string. // Returns tool-specific statuses ("web", "coding") that activate existing // but previously unused reaction variants in channel implementations. diff --git a/internal/channels/manager.go b/internal/channels/manager.go index 2ed3a1ff..82f1a486 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -9,6 +9,20 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" ) +// ChannelStream is the per-run streaming handle stored on RunContext. +// Each channel implementation returns a ChannelStream from CreateStream(). +// RunContext owns the stream so concurrent runs in the same group chat +// each get their own stream — no sync.Map collision on chatID. +type ChannelStream interface { + // Update sends or edits the streaming message with the latest accumulated text. + Update(ctx context.Context, text string) + // Stop finalizes the stream (final edit/flush). Called on run.completed. + Stop(ctx context.Context) error + // MessageID returns the platform message ID of the streaming message (0 if none). + // Used to hand the message back to Send() via the channel's placeholder map. + MessageID() int +} + // RunContext tracks an active agent run for streaming/reaction event forwarding. type RunContext struct { ChannelName string @@ -19,8 +33,13 @@ type RunContext struct { 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 mu sync.Mutex - streamBuffer string // accumulated streaming text (chunks are deltas) - inToolPhase bool // true after tool.call, reset on next chunk (new LLM iteration) + 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) } // Manager manages all registered channels, handling their lifecycle diff --git a/internal/channels/slack/channel.go b/internal/channels/slack/channel.go index add9e1c2..8e965847 100644 --- a/internal/channels/slack/channel.go +++ b/internal/channels/slack/channel.go @@ -38,7 +38,6 @@ type Channel struct { placeholders sync.Map // localKey -> placeholderTS dedup sync.Map // channel+ts -> time.Time threadParticip sync.Map // channelID+threadTS -> time.Time (auto-reply without @mention) - streams sync.Map // localKey -> *streamState reactions sync.Map // chatID:messageID -> *reactionState pairingDebounce sync.Map // senderID -> time.Time approvedGroups sync.Map // channelID -> true @@ -243,17 +242,6 @@ func (c *Channel) sweepMaps() { } return true }) - - c.streams.Range(func(k, v any) bool { - st := v.(*streamState) - st.mu.Lock() - stale := now.Sub(st.lastUpdate) > 10*time.Minute - st.mu.Unlock() - if stale { - c.streams.Delete(k) - } - return true - }) } // eventLoop processes Socket Mode events. diff --git a/internal/channels/slack/stream.go b/internal/channels/slack/stream.go index f8b8836a..ca085650 100644 --- a/internal/channels/slack/stream.go +++ b/internal/channels/slack/stream.go @@ -8,12 +8,16 @@ import ( "time" slackapi "github.com/slack-go/slack" + + "github.com/nextlevelbuilder/goclaw/internal/channels" ) const streamThrottleInterval = 1000 * time.Millisecond -// streamState tracks per-chat streaming state. -type streamState struct { +// slackStream implements channels.ChannelStream for Slack. +// It edits the placeholder "Thinking..." message as chunks arrive. +type slackStream struct { + api *slackapi.Client channelID string threadTS string msgTS string // placeholder message timestamp @@ -21,6 +25,49 @@ type streamState struct { mu sync.Mutex } +// Update edits the placeholder with accumulated text, throttled to avoid rate limits. +func (s *slackStream) Update(_ context.Context, fullText string) { + s.mu.Lock() + defer s.mu.Unlock() + + if time.Since(s.lastUpdate) < streamThrottleInterval { + return + } + + formatted := markdownToSlackMrkdwn(fullText) + if len(formatted) > maxMessageLen { + formatted = formatted[:maxMessageLen] + "..." + } + + opts := []slackapi.MsgOption{slackapi.MsgOptionText(formatted, false)} + _, _, _, err := s.api.UpdateMessage(s.channelID, s.msgTS, opts...) + if err != nil { + slog.Debug("slack stream chunk update failed", "error", err) + return + } + + s.lastUpdate = time.Now() +} + +// Stop finalizes the stream. For Slack, Send() handles the final edit via the placeholder map, +// so Stop() is a no-op here — FinalizeStream stores the msgTS into c.placeholders. +func (s *slackStream) Stop(_ context.Context) error { + return nil +} + +// MessageID returns 0 — Slack uses string timestamps, not int message IDs. +// FinalizeStream handles the Slack-specific placeholder handoff via type assertion. +func (s *slackStream) MessageID() int { + return 0 +} + +// MsgTS returns the Slack message timestamp (placeholder TS) for FinalizeStream handoff. +func (s *slackStream) MsgTS() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.msgTS +} + // StreamEnabled reports whether streaming is active for DMs or groups. func (c *Channel) StreamEnabled(isGroup bool) bool { if isGroup { @@ -29,90 +76,41 @@ func (c *Channel) StreamEnabled(isGroup bool) bool { return c.config.DMStream != nil && *c.config.DMStream } -// OnStreamStart is called when a new agent run starts. +// CreateStream creates a per-run streaming handle for the given chatID. +// Implements channels.StreamingChannel. // The placeholder "Thinking..." was already sent in handleMessage. -func (c *Channel) OnStreamStart(_ context.Context, _ string) error { - return nil -} - -// OnChunkEvent edits the placeholder with accumulated text, throttled to avoid rate limits. -func (c *Channel) OnChunkEvent(_ context.Context, chatID string, fullText string) error { - stateVal, ok := c.streams.Load(chatID) - if !ok { - pTS, pOK := c.placeholders.Load(chatID) - if !pOK { - return nil - } - - st := &streamState{ +func (c *Channel) CreateStream(_ context.Context, chatID string, _ bool) (channels.ChannelStream, error) { + pTS, pOK := c.placeholders.Load(chatID) + if !pOK { + // No placeholder — stream will be a no-op + return &slackStream{ + api: c.api, channelID: extractChannelID(chatID), - msgTS: pTS.(string), - } - if threadTS := extractThreadTS(chatID); threadTS != "" { - st.threadTS = threadTS - } - c.streams.Store(chatID, st) - stateVal = st + threadTS: extractThreadTS(chatID), + }, nil } - st := stateVal.(*streamState) - st.mu.Lock() - defer st.mu.Unlock() - - if time.Since(st.lastUpdate) < streamThrottleInterval { - return nil - } - - formatted := markdownToSlackMrkdwn(fullText) - if len(formatted) > maxMessageLen { - formatted = formatted[:maxMessageLen] + "..." - } - - opts := []slackapi.MsgOption{slackapi.MsgOptionText(formatted, false)} - _, _, _, err := c.api.UpdateMessage(st.channelID, st.msgTS, opts...) - if err != nil { - slog.Debug("slack stream chunk update failed", "error", err) - return nil - } - - st.lastUpdate = time.Now() - return nil + return &slackStream{ + api: c.api, + channelID: extractChannelID(chatID), + threadTS: extractThreadTS(chatID), + msgTS: pTS.(string), + }, nil } -// OnStreamEnd sends final formatted text. -func (c *Channel) OnStreamEnd(_ context.Context, chatID string, finalText string) error { - if finalText == "" { - c.streams.Delete(chatID) - return nil +// FinalizeStream stores the stream's placeholder TS back into c.placeholders so that +// Send() can edit it with the properly formatted final response. +// Implements channels.StreamingChannel. +// ReasoningStreamEnabled returns false — Slack lane support is deferred to a separate PR. +// Slack streaming uses thread replies which have different UX from Telegram in-place edit. +func (c *Channel) ReasoningStreamEnabled() bool { return false } + +func (c *Channel) FinalizeStream(_ context.Context, chatID string, stream channels.ChannelStream) { + ss, ok := stream.(*slackStream) + if !ok || ss.msgTS == "" { + return } - - channelID := extractChannelID(chatID) - var msgTS, threadTS string - - if stateVal, ok := c.streams.Load(chatID); ok { - st := stateVal.(*streamState) - msgTS = st.msgTS - threadTS = st.threadTS - } else if pTS, ok := c.placeholders.Load(chatID); ok { - msgTS = pTS.(string) - } - - defer c.streams.Delete(chatID) - - if msgTS == "" { - return nil - } - - formatted := markdownToSlackMrkdwn(finalText) - if len(formatted) <= maxMessageLen { - _, _, _, err := c.api.UpdateMessage(channelID, msgTS, slackapi.MsgOptionText(formatted, false)) - return err - } - - // Final text exceeds limit -- update placeholder with first chunk, send rest as follow-ups - first, rest := splitAtLimit(formatted, maxMessageLen) - _, _, _, _ = c.api.UpdateMessage(channelID, msgTS, slackapi.MsgOptionText(first, false)) - return c.sendChunked(channelID, rest, threadTS) + c.placeholders.Store(chatID, ss.msgTS) } // extractChannelID gets the channel ID from a local_key. diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index bba37e41..3d3afaee 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -29,7 +29,6 @@ type Channel struct { placeholders sync.Map // localKey string → messageID int stopThinking sync.Map // localKey string → *thinkingCancel typingCtrls sync.Map // localKey string → *typing.Controller - streams sync.Map // localKey string → *DraftStream (streaming preview) reactions sync.Map // localKey string → *StatusReactionController pairingReplySent sync.Map // userID string → time.Time (debounce pairing replies) threadIDs sync.Map // localKey string → messageThreadID int (for forum topic routing) @@ -193,13 +192,9 @@ func (c *Channel) Start(ctx context.Context) error { // StreamEnabled reports whether streaming is active for the given chat type. // Controlled by separate dm_stream / group_stream config flags (both default false). // -// DM streaming: edits "Thinking..." placeholder progressively via editMessageText. +// DM streaming: uses sendMessageDraft (stealth preview) by default, falls back to +// sendMessage+editMessageText if draft API is unavailable. Controlled by draft_transport config. // Group streaming: sends a new message, edits progressively, hands off to Send(). -// -// Both are disabled by default. sendMessageDraft (draft transport) code exists in the -// codebase but is not used pending Telegram client-side fixes: -// - tdesktop#10315, bugs.telegram.org/c/561 — "reply to deleted message" artifacts -// - openclaw/openclaw#7803, openclaw/openclaw#32180 — community tracking func (c *Channel) StreamEnabled(isGroup bool) bool { if isGroup { return c.config.GroupStream != nil && *c.config.GroupStream @@ -207,6 +202,25 @@ func (c *Channel) StreamEnabled(isGroup bool) bool { return c.config.DMStream != nil && *c.config.DMStream } +// draftTransportEnabled returns whether sendMessageDraft should be used for DM streaming. +// Default: true (enabled). Uses stealth preview with no per-edit notifications. +// Note: may cause "reply to deleted message" artifacts on some Telegram clients (tdesktop#10315). +func (c *Channel) draftTransportEnabled() bool { + if c.config.DraftTransport == nil { + return true + } + return *c.config.DraftTransport +} + +// ReasoningStreamEnabled returns whether reasoning should be shown as a separate message. +// Default: true. Set "reasoning_stream": false to hide reasoning (only show answer). +func (c *Channel) ReasoningStreamEnabled() bool { + if c.config.ReasoningStream == nil { + return true + } + return *c.config.ReasoningStream +} + // BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default). func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply } diff --git a/internal/channels/telegram/factory.go b/internal/channels/telegram/factory.go index 0bfcb4ae..82cfa62e 100644 --- a/internal/channels/telegram/factory.go +++ b/internal/channels/telegram/factory.go @@ -19,20 +19,22 @@ 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"` - HistoryLimit int `json:"history_limit,omitempty"` - DMStream *bool `json:"dm_stream,omitempty"` - GroupStream *bool `json:"group_stream,omitempty"` - 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"` - 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"` + 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"` + AllowFrom []string `json:"allow_from,omitempty"` } // Factory creates a Telegram channel from DB instance data (no extra stores). @@ -89,9 +91,11 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage, GroupPolicy: ic.GroupPolicy, RequireMention: ic.RequireMention, HistoryLimit: ic.HistoryLimit, - DMStream: ic.DMStream, - GroupStream: ic.GroupStream, - ReactionLevel: ic.ReactionLevel, + DMStream: ic.DMStream, + GroupStream: ic.GroupStream, + DraftTransport: ic.DraftTransport, + ReasoningStream: ic.ReasoningStream, + ReactionLevel: ic.ReactionLevel, MediaMaxBytes: resolveMediaMaxBytes(ic), LinkPreview: ic.LinkPreview, BlockReply: ic.BlockReply, diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 1dd159c7..282c4b2b 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -410,20 +410,10 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { _, thinkCancel := context.WithCancel(ctx) c.stopThinking.Store(localKey, &thinkingCancel{fn: thinkCancel}) - // Send "Thinking..." placeholder for DMs. - // The streaming system will edit this message progressively (editMessageText), - // giving a smooth transition: "Thinking..." → streaming chunks → final formatted response. - // Groups: no placeholder; response replies to the sender's message. - if !isGroup { - thinkMsg := tu.Message(chatIDObj, "Thinking...") - if dmThreadID > 0 { - thinkMsg.MessageThreadID = dmThreadID - } - pMsg, err := c.bot.SendMessage(ctx, thinkMsg) - if err == nil { - c.placeholders.Store(localKey, pMsg.MessageID) - } - } + // No "Thinking..." placeholder — the DraftStream creates its own message + // on the first streaming chunk (sendMessage on first flush). + // This avoids "reply to deleted message" artifacts and is cleaner UX: + // user sees typing indicator → first content appears directly. metadata := map[string]string{ "message_id": fmt.Sprintf("%d", message.MessageID), diff --git a/internal/channels/telegram/send.go b/internal/channels/telegram/send.go index 2a7be67e..e1d19e63 100644 --- a/internal/channels/telegram/send.go +++ b/internal/channels/telegram/send.go @@ -164,30 +164,29 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { // Text-only message htmlContent := markdownToTelegramHTML(msg.Content) + chunks := chunkHTML(htmlContent, telegramMaxMessageLen) - // Try to edit the placeholder message (either "Thinking..." or a DraftStream message). - // If edit succeeds, we're done. If content is too long or edit fails, delete the - // placeholder and fall through to send new chunked messages. + // If a stream message exists (stored by FinalizeStream), edit the first chunk + // into it instead of deleting. This prevents the message from vanishing + // when HTML conversion makes content exceed the size limit. + startChunk := 0 if pID, ok := c.placeholders.Load(localKey); ok { c.placeholders.Delete(localKey) - if len(htmlContent) <= telegramMaxMessageLen { - if err := c.editMessage(ctx, chatID, pID.(int), htmlContent); err == nil { - return nil - } + if err := c.editMessage(ctx, chatID, pID.(int), chunks[0]); err == nil { + startChunk = 1 // first chunk edited into stream message + } else { + // Edit failed (message deleted externally, etc.) — delete and send all fresh + _ = c.deleteMessage(ctx, chatID, pID.(int)) } - // Delete the placeholder since we'll send new message(s) instead - _ = c.deleteMessage(ctx, chatID, pID.(int)) } - // Chunk long messages to respect Telegram's limit. - // TS ref: only reply to the first chunk (src/channels/plugins/outbound/telegram.ts). - chunks := chunkHTML(htmlContent, telegramMaxMessageLen) - for i, chunk := range chunks { + // Send remaining chunks (or all chunks if no stream message was edited). + for i := startChunk; i < len(chunks); i++ { replyTo := 0 if i == 0 { replyTo = replyToMsgID // only first chunk replies to user's message } - if err := c.sendHTML(ctx, chatID, chunk, replyTo, threadID); err != nil { + if err := c.sendHTML(ctx, chatID, chunks[i], replyTo, threadID); err != nil { return err } } @@ -279,7 +278,10 @@ func (c *Channel) sendHTML(ctx context.Context, chatID int64, html string, reply tgMsg.MessageThreadID = sendThreadID } if replyTo > 0 { - tgMsg.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo} + tgMsg.ReplyParameters = &telego.ReplyParameters{ + MessageID: replyTo, + AllowSendingWithoutReply: true, // don't fail if replied-to message was deleted + } } err := retrySend(ctx, "sendMessage", nil, func() error { @@ -321,7 +323,7 @@ func (c *Channel) sendPhoto(ctx context.Context, chatID telego.ChatID, filePath, params.MessageThreadID = sendThreadID } if replyTo > 0 { - params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo} + params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo, AllowSendingWithoutReply: true} } err = retrySend(ctx, "sendPhoto", func() { file.Seek(0, 0) }, func() error { @@ -364,7 +366,7 @@ func (c *Channel) sendVideo(ctx context.Context, chatID telego.ChatID, filePath, params.MessageThreadID = sendThreadID } if replyTo > 0 { - params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo} + params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo, AllowSendingWithoutReply: true} } err = retrySend(ctx, "sendVideo", func() { file.Seek(0, 0) }, func() error { @@ -407,7 +409,7 @@ func (c *Channel) sendAudio(ctx context.Context, chatID telego.ChatID, filePath, params.MessageThreadID = sendThreadID } if replyTo > 0 { - params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo} + params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo, AllowSendingWithoutReply: true} } err = retrySend(ctx, "sendAudio", func() { file.Seek(0, 0) }, func() error { @@ -450,7 +452,7 @@ func (c *Channel) sendDocument(ctx context.Context, chatID telego.ChatID, filePa params.MessageThreadID = sendThreadID } if replyTo > 0 { - params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo} + params.ReplyParameters = &telego.ReplyParameters{MessageID: replyTo, AllowSendingWithoutReply: true} } err = retrySend(ctx, "sendDocument", func() { file.Seek(0, 0) }, func() error { diff --git a/internal/channels/telegram/stream.go b/internal/channels/telegram/stream.go index f284d956..84cfd70d 100644 --- a/internal/channels/telegram/stream.go +++ b/internal/channels/telegram/stream.go @@ -12,6 +12,8 @@ import ( "github.com/mymmrac/telego" tu "github.com/mymmrac/telego/telegoutil" + + "github.com/nextlevelbuilder/goclaw/internal/channels" ) const ( @@ -89,10 +91,10 @@ type DraftStream struct { draftFailed bool // true = draft API rejected permanently, using message transport } -// newDraftStream creates a new streaming preview manager. +// NewDraftStream creates a new streaming preview manager. // When useDraft is true, the stream will attempt to use sendMessageDraft (Bot API 9.3+) // and automatically fall back to sendMessage+editMessageText if the API rejects it. -func newDraftStream(bot *telego.Bot, chatID int64, throttleMs int, messageThreadID int, useDraft bool) *DraftStream { +func NewDraftStream(bot *telego.Bot, chatID int64, throttleMs int, messageThreadID int, useDraft bool) *DraftStream { throttle := defaultStreamThrottle if throttleMs > 0 { throttle = time.Duration(throttleMs) * time.Millisecond @@ -269,8 +271,8 @@ func (ds *DraftStream) UsedDraftTransport() bool { // --- StreamingChannel implementation --- -// OnStreamStart prepares for streaming. -// chatID here is the localKey (composite key with :topic:N suffix for forum topics). +// CreateStream prepares a per-run streaming handle for the given chatID (localKey). +// Implements channels.StreamingChannel. // // For DMs: seeds the stream with the "Thinking..." placeholder messageID so that // flush() uses editMessageText to update it progressively. This gives a smooth @@ -278,10 +280,10 @@ func (ds *DraftStream) UsedDraftTransport() bool { // // For groups: deletes the placeholder and lets the stream create its own message, // since group placeholders drift away as other messages arrive. -func (c *Channel) OnStreamStart(ctx context.Context, chatID string) error { +func (c *Channel) CreateStream(ctx context.Context, chatID string, firstStream bool) (channels.ChannelStream, error) { id, err := parseRawChatID(chatID) if err != nil { - return err + return nil, err } // Look up thread ID stored during handleMessage @@ -292,68 +294,25 @@ func (c *Channel) OnStreamStart(ctx context.Context, chatID string) error { isDM := id > 0 - // Both DMs and groups use message transport (editMessageText). - // sendMessageDraft (draft transport) is available in the codebase but disabled - // because it causes "reply to deleted message" artifacts in Telegram clients. - ds := newDraftStream(c.bot, id, 0, threadID, false) + // Draft transport only for non-first streams (answer lane) in DMs. + // First stream must use message transport because it may become the + // reasoning lane — draft messages are ephemeral and would disappear + // when the answer stream starts. + useDraft := isDM && !firstStream && c.draftTransportEnabled() + ds := NewDraftStream(c.bot, id, 0, threadID, useDraft) - if isDM { - // DMs: seed the stream with the "Thinking..." placeholder messageID. - // flush() will use editMessageText to update it progressively. - if pID, ok := c.placeholders.Load(chatID); ok { - c.placeholders.Delete(chatID) - ds.messageID = pID.(int) - slog.Info("stream: DM using placeholder for progressive edit", "chat_id", id, "message_id", pID.(int)) - } else { - slog.Info("stream: DM starting stream (no placeholder found)", "chat_id", id) - } - } else { - // Groups: delete placeholder — the stream creates its own message. - if pID, ok := c.placeholders.Load(chatID); ok { - c.placeholders.Delete(chatID) - _ = c.deleteMessage(ctx, id, pID.(int)) - } - slog.Info("stream: group using message transport", "chat_id", id) - } + // No placeholder seeding — DraftStream creates its own message on first flush(). + // This avoids "reply to deleted/non-existent message" artifacts. - c.streams.Store(chatID, ds) - - return nil + return ds, nil } -// OnChunkEvent updates the streaming message with accumulated content. -func (c *Channel) OnChunkEvent(ctx context.Context, chatID string, fullText string) error { - - val, ok := c.streams.Load(chatID) - if !ok { - return nil - } - - ds := val.(*DraftStream) - ds.Update(ctx, fullText) - return nil -} - -// OnStreamEnd finalizes the streaming preview. -// Hands the stream's messageID back to the placeholders map so that Send() +// FinalizeStream hands the stream's messageID back to the placeholders map so that Send() // can edit it with the properly formatted final response. -func (c *Channel) OnStreamEnd(ctx context.Context, chatID string, _ string) error { - val, ok := c.streams.Load(chatID) - if !ok { - return nil - } - - ds := val.(*DraftStream) - - // Mark stream as stopped (no more edits) - ds.mu.Lock() - ds.stopped = true - msgID := ds.messageID - ds.mu.Unlock() - - c.streams.Delete(chatID) - - if msgID != 0 { +// Also stops any thinking animation for the chat. +// Implements channels.StreamingChannel. +func (c *Channel) FinalizeStream(ctx context.Context, chatID string, stream channels.ChannelStream) { + if msgID := stream.MessageID(); msgID != 0 { // Hand off the stream message to Send() for final formatted edit. c.placeholders.Store(chatID, msgID) slog.Info("stream: ended, handing off to Send()", "chat_id", chatID, "message_id", msgID) @@ -366,6 +325,4 @@ func (c *Channel) OnStreamEnd(ctx context.Context, chatID string, _ string) erro } c.stopThinking.Delete(chatID) } - - return nil } diff --git a/internal/channels/think_tag_parser.go b/internal/channels/think_tag_parser.go new file mode 100644 index 00000000..f0a5a6d0 --- /dev/null +++ b/internal/channels/think_tag_parser.go @@ -0,0 +1,73 @@ +package channels + +import ( + "regexp" + "strings" +) + +// thinkOpenRe matches opening think tags emitted by various models. +// Covers: , , , , plus optional attributes. +var thinkOpenRe = regexp.MustCompile(`(?i)<\s*(?:think(?:ing)?|thought|antthinking)\b[^>]*>`) + +// thinkCloseRe matches closing think tags. +var thinkCloseRe = regexp.MustCompile(`(?i)`) + +// ThinkTagSplit holds the result of splitting think-tagged content. +type ThinkTagSplit struct { + Thinking string // content inside tags (empty if no tags found) + Answer string // content outside tags + Partial bool // true if an unclosed tag is present (buffer for more) +} + +// SplitThinkTags extracts thinking content from ... tags. +// Returns empty Thinking if no tags are found. Handles multiple tag pairs +// and accumulates all thinking/answer segments. +func SplitThinkTags(text string) ThinkTagSplit { + lower := strings.ToLower(text) + // Fast path: no think-like tags at all + if !strings.Contains(lower, " 0 { + answer.WriteString(remaining[:openLoc[0]]) + } + + // Find closing tag after the opening tag + afterOpen := remaining[openLoc[1]:] + closeLoc := thinkCloseRe.FindStringIndex(afterOpen) + if closeLoc == nil { + // Unclosed tag — content is still arriving (streaming) + thinking.WriteString(afterOpen) + return ThinkTagSplit{ + Thinking: thinking.String(), + Answer: answer.String(), + Partial: true, + } + } + + // Content between open and close tags is thinking + thinking.WriteString(afterOpen[:closeLoc[0]]) + remaining = afterOpen[closeLoc[1]:] + } + + return ThinkTagSplit{ + Thinking: thinking.String(), + Answer: answer.String(), + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 0d759a42..18236c60 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -33,9 +33,11 @@ type TelegramConfig struct { GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in groups (default true) 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; disabled pending Telegram draft API fixes (tdesktop#10315) - GroupStream *bool `json:"group_stream,omitempty"` // enable streaming for groups (default false) — sends new message, edits progressively - ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full" — status emoji reactions + 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) diff --git a/ui/web/src/i18n/locales/en/channels.json b/ui/web/src/i18n/locales/en/channels.json index b91e7d6b..c6d2c4a3 100644 --- a/ui/web/src/i18n/locales/en/channels.json +++ b/ui/web/src/i18n/locales/en/channels.json @@ -153,6 +153,86 @@ } } }, + "fieldConfig": { + "token": { "label": "Bot Token" }, + "bot_token": { "label": "Bot Token" }, + "app_token": { "label": "App-Level Token", "help": "App-Level Token with connections:write scope (required for Socket Mode)" }, + "user_token": { "label": "User Token (Optional)", "help": "Optional: User OAuth Token for custom bot identity. Leave empty to use default bot identity." }, + "app_id": { "label": "App ID" }, + "app_secret": { "label": "App Secret" }, + "encrypt_key": { "label": "Encrypt Key", "help": "For webhook mode" }, + "verification_token": { "label": "Verification Token", "help": "For webhook mode" }, + "webhook_secret": { "label": "Webhook Secret" }, + "bridge_url": { "label": "Bridge URL" }, + "api_server": { "label": "API Server URL", "help": "Custom Telegram Bot API server for large file uploads (up to 2GB). Leave empty for default." }, + "proxy": { "label": "HTTP Proxy", "help": "Route bot traffic through an HTTP proxy" }, + "dm_policy": { "label": "DM Policy" }, + "group_policy": { "label": "Group Policy" }, + "require_mention": { "label": "Require @mention in groups" }, + "history_limit": { "label": "Group History Limit", "help": "Max pending group messages for context (0 = disabled)" }, + "dm_stream": { "label": "DM Streaming", "help": "Stream response progressively in DMs" }, + "group_stream": { "label": "Group Streaming", "help": "Stream response progressively in groups" }, + "draft_transport": { "label": "Draft Preview", "help": "Use stealth draft preview for answer stream in DMs — no notification per edit (requires DM Streaming)" }, + "reasoning_stream": { "label": "Show Reasoning", "help": "Display AI thinking as a separate message before the answer (requires streaming)" }, + "reaction_level": { "label": "Reaction Level" }, + "media_max_mb": { "label": "Max Media Size (MB)" }, + "link_preview": { "label": "Link Preview" }, + "allow_from": { "label": "Allowed Users" }, + "block_reply": { "label": "Block Reply", "help": "Deliver intermediate text during tool iterations" }, + "domain": { "label": "Domain" }, + "connection_mode": { "label": "Connection Mode", "help": "Lark Global only supports Webhook" }, + "webhook_port": { "label": "Webhook Port", "help": "0 = share main gateway port (recommended)" }, + "webhook_path": { "label": "Webhook Path", "help": "Path on main server for Lark events" }, + "webhook_url": { "label": "Webhook URL" }, + "topic_session_mode": { "label": "Topic Session Mode", "help": "Use thread root_id for session isolation" }, + "render_mode": { "label": "Render Mode" }, + "text_chunk_limit": { "label": "Text Chunk Limit", "help": "Max characters per message" }, + "group_allow_from": { "label": "Group Allowed Users", "help": "Separate allowlist for group senders" }, + "native_stream": { "label": "Native Streaming (Agents & AI Apps)", "help": "Use Slack's ChatStreamer API for native streaming. Falls back to edit-in-place if unavailable." }, + "debounce_delay": { "label": "Debounce Delay (ms)", "help": "Milliseconds to wait before dispatching rapid messages. Set 0 to disable." }, + "thread_ttl": { "label": "Thread Participation TTL (hours)", "help": "Hours before bot stops auto-replying in threads it participated in. 0 = always require @mention." } + }, + "fieldOptions": { + "block_reply": { + "inherit": "Inherit from gateway", + "true": "Enabled", + "false": "Disabled" + }, + "dm_policy": { + "pairing": "Pairing (require code)", + "open": "Open (accept all)", + "allowlist": "Allowlist only", + "disabled": "Disabled" + }, + "group_policy": { + "open": "Open (accept all)", + "pairing": "Pairing (require approval)", + "allowlist": "Allowlist only", + "disabled": "Disabled" + }, + "reaction_level": { + "off": "Off", + "minimal": "Minimal", + "full": "Full" + }, + "domain": { + "lark": "Lark (global) — webhook only", + "feishu": "Feishu (China)" + }, + "connection_mode": { + "webhook": "Webhook", + "websocket": "WebSocket (Feishu only)" + }, + "topic_session_mode": { + "disabled": "Disabled", + "enabled": "Enabled" + }, + "render_mode": { + "auto": "Auto", + "raw": "Raw", + "card": "Card" + } + }, "groupOverrides": { "title": "Group & Topic Overrides", "hint": "Override channel defaults per group chat or forum topic. Use \"*\" as group ID for wildcard defaults.", @@ -164,6 +244,12 @@ "topicLabel": "Topic: {{id}}", "addTopicPlaceholder": "Topic thread ID", "addTopic": "Add Topic", + "fieldConfig": { + "dm_stream": { "label": "DM Streaming", "help": "Stream response progressively in DMs" }, + "group_stream": { "label": "Group Streaming", "help": "Stream response progressively in groups" }, + "draft_transport": { "label": "Draft Preview", "help": "Use stealth draft preview for answer stream in DMs — no notification per edit (requires DM Streaming)" }, + "reasoning_stream": { "label": "Show Reasoning", "help": "Display AI thinking as a separate message before the answer (requires streaming)" } + }, "fields": { "groupPolicy": "Group Policy", "requireMention": "Require @mention", diff --git a/ui/web/src/i18n/locales/en/config.json b/ui/web/src/i18n/locales/en/config.json index fa02b08c..b543ada7 100644 --- a/ui/web/src/i18n/locales/en/config.json +++ b/ui/web/src/i18n/locales/en/config.json @@ -282,7 +282,14 @@ "channels.allowFrom": "Allow From", "channels.allowFromPlaceholder": "user_id_1, user_id_2, ...", "channels.dmStreaming": "DM Streaming", + "channels.dmStreamingHint": "Progressive message updates in DMs", "channels.groupStreaming": "Group Streaming", + "channels.groupStreamingHint": "Progressive message updates in groups", + "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.reactionLevel": "Reaction Level", "channels.connectionMode": "Connection Mode" } diff --git a/ui/web/src/i18n/locales/vi/channels.json b/ui/web/src/i18n/locales/vi/channels.json index 4b7d125b..c3b9a80e 100644 --- a/ui/web/src/i18n/locales/vi/channels.json +++ b/ui/web/src/i18n/locales/vi/channels.json @@ -153,6 +153,86 @@ } } }, + "fieldConfig": { + "token": { "label": "Bot Token" }, + "bot_token": { "label": "Bot Token" }, + "app_token": { "label": "Token cấp ứng dụng", "help": "Token cấp ứng dụng với scope connections:write (bắt buộc cho Socket Mode)" }, + "user_token": { "label": "Token người dùng (Tùy chọn)", "help": "Tùy chọn: Token OAuth người dùng cho danh tính bot tùy chỉnh. Để trống để dùng danh tính bot mặc định." }, + "app_id": { "label": "App ID" }, + "app_secret": { "label": "App Secret" }, + "encrypt_key": { "label": "Khóa mã hóa", "help": "Cho chế độ webhook" }, + "verification_token": { "label": "Token xác minh", "help": "Cho chế độ webhook" }, + "webhook_secret": { "label": "Webhook Secret" }, + "bridge_url": { "label": "URL Bridge" }, + "api_server": { "label": "URL máy chủ API", "help": "Máy chủ API Bot Telegram tùy chỉnh cho tải file lớn (lên đến 2GB). Để trống cho mặc định." }, + "proxy": { "label": "Proxy HTTP", "help": "Định tuyến lưu lượng bot qua proxy HTTP" }, + "dm_policy": { "label": "Chính sách DM" }, + "group_policy": { "label": "Chính sách nhóm" }, + "require_mention": { "label": "Yêu cầu @đề cập trong nhóm" }, + "history_limit": { "label": "Giới hạn lịch sử nhóm", "help": "Số tin nhắn nhóm chờ tối đa cho ngữ cảnh (0 = tắt)" }, + "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)" }, + "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" }, + "allow_from": { "label": "Người dùng được phép" }, + "block_reply": { "label": "Phản hồi khối", "help": "Gửi văn bản trung gian trong quá trình lặp công cụ" }, + "domain": { "label": "Tên miền" }, + "connection_mode": { "label": "Chế độ kết nối", "help": "Lark Global chỉ hỗ trợ Webhook" }, + "webhook_port": { "label": "Cổng webhook", "help": "0 = chia sẻ cổng gateway chính (khuyến nghị)" }, + "webhook_path": { "label": "Đường dẫn webhook", "help": "Đường dẫn trên máy chủ chính cho sự kiện Lark" }, + "webhook_url": { "label": "URL Webhook" }, + "topic_session_mode": { "label": "Chế độ phiên chủ đề", "help": "Dùng root_id của luồng để cách ly phiên" }, + "render_mode": { "label": "Chế độ hiển thị" }, + "text_chunk_limit": { "label": "Giới hạn ký tự", "help": "Số ký tự tối đa mỗi tin nhắn" }, + "group_allow_from": { "label": "Người dùng nhóm được phép", "help": "Danh sách cho phép riêng cho người gửi nhóm" }, + "native_stream": { "label": "Stream gốc (Agents & AI Apps)", "help": "Dùng API ChatStreamer của Slack cho stream gốc. Quay về chỉnh sửa tại chỗ nếu không khả dụng." }, + "debounce_delay": { "label": "Độ trễ debounce (ms)", "help": "Mili giây chờ trước khi gửi tin nhắn nhanh. Đặt 0 để tắt." }, + "thread_ttl": { "label": "TTL tham gia luồng (giờ)", "help": "Số giờ trước khi bot ngừng tự động trả lời trong các luồng đã tham gia. 0 = luôn yêu cầu @đề cập." } + }, + "fieldOptions": { + "block_reply": { + "inherit": "Kế thừa từ gateway", + "true": "Bật", + "false": "Tắt" + }, + "dm_policy": { + "pairing": "Ghép nối (yêu cầu mã)", + "open": "Mở (chấp nhận tất cả)", + "allowlist": "Chỉ danh sách cho phép", + "disabled": "Tắt" + }, + "group_policy": { + "open": "Mở (chấp nhận tất cả)", + "pairing": "Ghép nối (yêu cầu phê duyệt)", + "allowlist": "Chỉ danh sách cho phép", + "disabled": "Tắt" + }, + "reaction_level": { + "off": "Tắt", + "minimal": "Tối thiểu", + "full": "Đầy đủ" + }, + "domain": { + "lark": "Lark (quốc tế) — chỉ webhook", + "feishu": "Feishu (Trung Quốc)" + }, + "connection_mode": { + "webhook": "Webhook", + "websocket": "WebSocket (chỉ Feishu)" + }, + "topic_session_mode": { + "disabled": "Tắt", + "enabled": "Bật" + }, + "render_mode": { + "auto": "Tự động", + "raw": "Thô", + "card": "Thẻ" + } + }, "groupOverrides": { "title": "Ghi đè nhóm & chủ đề", "hint": "Ghi đè mặc định channel theo cuộc trò chuyện nhóm hoặc chủ đề diễn đàn. Dùng \"*\" làm ID nhóm cho mặc định ký tự đại diện.", @@ -164,6 +244,12 @@ "topicLabel": "Chủ đề: {{id}}", "addTopicPlaceholder": "ID luồng chủ đề", "addTopic": "Thêm chủ đề", + "fieldConfig": { + "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)" } + }, "fields": { "groupPolicy": "Chính sách nhóm", "requireMention": "Yêu cầu @đề cập", diff --git a/ui/web/src/i18n/locales/vi/config.json b/ui/web/src/i18n/locales/vi/config.json index cc9a8d37..d58ae7d2 100644 --- a/ui/web/src/i18n/locales/vi/config.json +++ b/ui/web/src/i18n/locales/vi/config.json @@ -282,7 +282,14 @@ "channels.allowFrom": "Cho phép từ", "channels.allowFromPlaceholder": "user_id_1, user_id_2, ...", "channels.dmStreaming": "Streaming tin nhắn riêng", + "channels.dmStreamingHint": "Cập nhật tin nhắn liên tục trong DM", "channels.groupStreaming": "Streaming nhóm", + "channels.groupStreamingHint": "Cập nhật tin nhắn liên tục trong nhóm", + "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.reactionLevel": "Mức phản ứng", "channels.connectionMode": "Chế độ kết nối" } diff --git a/ui/web/src/i18n/locales/zh/channels.json b/ui/web/src/i18n/locales/zh/channels.json index 75560dd9..8fa53728 100644 --- a/ui/web/src/i18n/locales/zh/channels.json +++ b/ui/web/src/i18n/locales/zh/channels.json @@ -153,6 +153,86 @@ } } }, + "fieldConfig": { + "token": { "label": "Bot Token" }, + "bot_token": { "label": "Bot Token" }, + "app_token": { "label": "应用级Token", "help": "具有 connections:write 权限的应用级Token（Socket Mode 必需）" }, + "user_token": { "label": "用户Token（可选）", "help": "可选：用于自定义Bot身份的用户OAuth Token。留空使用默认Bot身份。" }, + "app_id": { "label": "App ID" }, + "app_secret": { "label": "App Secret" }, + "encrypt_key": { "label": "加密密钥", "help": "用于Webhook模式" }, + "verification_token": { "label": "验证Token", "help": "用于Webhook模式" }, + "webhook_secret": { "label": "Webhook Secret" }, + "bridge_url": { "label": "Bridge URL" }, + "api_server": { "label": "API 服务器 URL", "help": "自定义 Telegram Bot API 服务器用于大文件上传（最大 2GB）。留空使用默认服务器。" }, + "proxy": { "label": "HTTP 代理", "help": "通过 HTTP 代理路由Bot流量" }, + "dm_policy": { "label": "私聊策略" }, + "group_policy": { "label": "群组策略" }, + "require_mention": { "label": "群组中需要 @提及" }, + "history_limit": { "label": "群组历史限制", "help": "上下文中最大待处理群组消息数（0 = 禁用）" }, + "dm_stream": { "label": "私聊流式传输", "help": "在私聊中逐步传输回复" }, + "group_stream": { "label": "群聊流式传输", "help": "在群聊中逐步传输回复" }, + "draft_transport": { "label": "草稿预览", "help": "在私聊中使用隐形草稿预览回答流 — 每次编辑无通知（需启用私聊流式传输）" }, + "reasoning_stream": { "label": "显示推理", "help": "将AI思考过程显示为单独的消息（需启用流式传输）" }, + "reaction_level": { "label": "表情回应级别" }, + "media_max_mb": { "label": "最大媒体大小 (MB)" }, + "link_preview": { "label": "链接预览" }, + "allow_from": { "label": "允许的用户" }, + "block_reply": { "label": "分块回复", "help": "在工具迭代期间发送中间文本" }, + "domain": { "label": "域名" }, + "connection_mode": { "label": "连接模式", "help": "Lark 国际版仅支持 Webhook" }, + "webhook_port": { "label": "Webhook 端口", "help": "0 = 共享主网关端口（推荐）" }, + "webhook_path": { "label": "Webhook 路径", "help": "主服务器上 Lark 事件的路径" }, + "webhook_url": { "label": "Webhook URL" }, + "topic_session_mode": { "label": "话题会话模式", "help": "使用线程 root_id 进行会话隔离" }, + "render_mode": { "label": "渲染模式" }, + "text_chunk_limit": { "label": "文本分段限制", "help": "每条消息最大字符数" }, + "group_allow_from": { "label": "群组允许的用户", "help": "群组发送者的单独白名单" }, + "native_stream": { "label": "原生流式传输 (Agents & AI Apps)", "help": "使用 Slack 的 ChatStreamer API 进行原生流式传输。不可用时回退到原地编辑。" }, + "debounce_delay": { "label": "防抖延迟 (ms)", "help": "发送快速消息前等待的毫秒数。设为 0 禁用。" }, + "thread_ttl": { "label": "线程参与 TTL（小时）", "help": "Bot 停止自动回复已参与线程的小时数。0 = 始终需要 @提及。" } + }, + "fieldOptions": { + "block_reply": { + "inherit": "继承网关设置", + "true": "启用", + "false": "禁用" + }, + "dm_policy": { + "pairing": "配对（需要验证码）", + "open": "开放（接受所有）", + "allowlist": "仅白名单", + "disabled": "禁用" + }, + "group_policy": { + "open": "开放（接受所有）", + "pairing": "配对（需要审批）", + "allowlist": "仅白名单", + "disabled": "禁用" + }, + "reaction_level": { + "off": "关闭", + "minimal": "最小", + "full": "全部" + }, + "domain": { + "lark": "Lark（国际版）— 仅 Webhook", + "feishu": "飞书（中国版）" + }, + "connection_mode": { + "webhook": "Webhook", + "websocket": "WebSocket（仅飞书）" + }, + "topic_session_mode": { + "disabled": "禁用", + "enabled": "启用" + }, + "render_mode": { + "auto": "自动", + "raw": "原始", + "card": "卡片" + } + }, "groupOverrides": { "title": "群组和话题覆盖", "hint": "按群组聊天或论坛话题覆盖Channel默认值。使用 「*」 作为群组 ID 表示通配符默认值。", @@ -164,6 +244,12 @@ "topicLabel": "话题: {{id}}", "addTopicPlaceholder": "话题线程 ID", "addTopic": "添加话题", + "fieldConfig": { + "dm_stream": { "label": "私聊流式传输", "help": "在私聊中逐步传输回复" }, + "group_stream": { "label": "群聊流式传输", "help": "在群聊中逐步传输回复" }, + "draft_transport": { "label": "草稿预览", "help": "在私聊中使用隐形草稿预览回答流 — 每次编辑无通知（需启用私聊流式传输）" }, + "reasoning_stream": { "label": "显示推理", "help": "将AI思考过程显示为单独的消息（需启用流式传输）" } + }, "fields": { "groupPolicy": "群组策略", "requireMention": "需要 @提及", diff --git a/ui/web/src/i18n/locales/zh/config.json b/ui/web/src/i18n/locales/zh/config.json index 04c878ba..045eae5b 100644 --- a/ui/web/src/i18n/locales/zh/config.json +++ b/ui/web/src/i18n/locales/zh/config.json @@ -282,7 +282,14 @@ "channels.allowFrom": "允许来自", "channels.allowFromPlaceholder": "user_id_1, user_id_2, ...", "channels.dmStreaming": "私信流式传输", + "channels.dmStreamingHint": "私聊中逐步更新消息", "channels.groupStreaming": "群组流式传输", + "channels.groupStreamingHint": "群聊中逐步更新消息", + "channels.streaming": "流式传输", + "channels.draftTransport": "草稿预览", + "channels.draftTransportHint": "私聊中静默流式预览（每次编辑无通知）", + "channels.reasoningStream": "显示推理", + "channels.reasoningStreamHint": "将AI思考过程显示为单独的消息", "channels.reactionLevel": "反应级别", "channels.connectionMode": "连接模式" } diff --git a/ui/web/src/pages/channels/channel-detail/channel-config-tab.tsx b/ui/web/src/pages/channels/channel-detail/channel-config-tab.tsx index 0d698984..3cd17bfa 100644 --- a/ui/web/src/pages/channels/channel-detail/channel-config-tab.tsx +++ b/ui/web/src/pages/channels/channel-detail/channel-config-tab.tsx @@ -77,7 +77,7 @@ export function ChannelConfigTab({ instance, onUpdate }: ChannelConfigTabProps) }, [webhookPath]); return ( -
+
.label" / "fieldConfig..help", fall back to hardcoded schema string + const label = t(`fieldConfig.${field.key}.label`, { defaultValue: field.label }); + const help = field.help ? t(`fieldConfig.${field.key}.help`, { defaultValue: field.help }) : ""; const labelSuffix = field.required && !isEdit ? " *" : ""; const editHint = isEdit && field.type === "password" ? ` ${t("form.credentialsHint")}` : ""; @@ -60,7 +63,7 @@ function FieldRenderer({ return (
onChange(e.target.value)} placeholder={field.placeholder} /> - {field.help &&

{field.help}

} + {help &&

{help}

}
); case "number": return (
- + onChange(e.target.value ? Number(e.target.value) : undefined)} placeholder={field.defaultValue !== undefined ? String(field.defaultValue) : undefined} /> - {field.help &&

{field.help}

} + {help &&

{help}

}
); @@ -96,15 +99,15 @@ function FieldRenderer({ checked={(value as boolean) ?? (field.defaultValue as boolean) ?? false} onCheckedChange={(v) => onChange(v)} /> - - {field.help && — {field.help}} + + {help && — {help}}
); case "select": return (
- + - {field.help &&

{field.help}

} + {help &&

{help}

}
); case "tags": return (
- +