feat(channels): add human-like chat behavior

Squash merge PR #99 after resolving conflicts with current dev. PR CI run 26703381807 passed release-versioning, go, and web.
This commit is contained in:
Duy /zuey/ authored and GitHub committed 2026-05-31 11:44:56 +07:00
1 parent 3569f8d681
commit f0f39ce31f
51 files changed
+1786 -205

No files matched your search

+12 -5
View File
@@ -13,6 +13,7 @@ import (
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
@@ -66,11 +67,11 @@ type Channel struct {
// access_token — no shared admin secret is required. mcpServerID is
// resolved once at Start() via mcpStore.GetServerByName and then
// cached — avoids looking up the server on every inbound message.
mcpStore store.MCPServerStore
mcpClient *mcpClient
mcpServerID uuid.UUID
mcpProvMu sync.Mutex
mcpDebounce map[mcpDebounceKey]time.Time
mcpStore store.MCPServerStore
mcpClient *mcpClient
mcpServerID uuid.UUID
mcpProvMu sync.Mutex
mcpDebounce map[mcpDebounceKey]time.Time
// User-facing degradation notice state. When provisionIfMissing fails
// in an UNEXPECTED way (HTTP failure, persist failure, not one of the
@@ -120,6 +121,12 @@ func (c *Channel) PortalName() string { return c.cfg.Portal }
// Config returns a copy of the instance config. Exported for tests.
func (c *Channel) Config() bitrixInstanceConfig { return c.cfg }
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.cfg.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.cfg.ChatBehavior }
// IsOpenChannelBot reports whether this channel was registered as a Bitrix24
// Open Channel bot (TYPE "O"), i.e. a customer-facing bot attached to an
// Open Channel queue. Standard internal bots (TYPE "B") return false.
+7 -5
View File
@@ -8,6 +8,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/store"
)
@@ -29,9 +30,9 @@ type bitrixCreds struct{}
// sent to imbot.register).
type bitrixInstanceConfig struct {
// Resource link (required)
Portal string `json:"portal"` // bitrix_portals.name scoped by tenant_id
BotCode string `json:"bot_code"` // stable key passed to imbot.register / LookupRegisteredBot
BotName string `json:"bot_name"` // display name
Portal string `json:"portal"` // bitrix_portals.name scoped by tenant_id
BotCode string `json:"bot_code"` // stable key passed to imbot.register / LookupRegisteredBot
BotName string `json:"bot_name"` // display name
BotAvatar string `json:"bot_avatar,omitempty"` // optional URL; factory resolves and base64-encodes at Start()
// BotType — forwarded verbatim to imbot.register TYPE param.
@@ -72,8 +73,9 @@ type bitrixInstanceConfig struct {
ReactionLevel string `json:"reaction_level,omitempty"` // off|minimal|full
// Misc
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
// Webhook endpoint override. Bitrix24 imbot.register requires absolute
// URLs for EVENT_MESSAGE_ADD etc. GoClaw has no global GOCLAW_PUBLIC_URL
+7
View File
@@ -20,6 +20,7 @@ import (
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
@@ -164,6 +165,12 @@ type BlockReplyChannel interface {
BlockReplyEnabled() *bool
}
// ChatBehaviorChannel is optionally implemented by channels that override
// gateway-level human-like delivery behavior. Nil means inherit the gateway default.
type ChatBehaviorChannel interface {
ChatBehaviorConfig() *config.ChatBehaviorConfig
}
// 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
+193
View File
@@ -0,0 +1,193 @@
package channels
import (
"strings"
"github.com/nextlevelbuilder/goclaw/internal/config"
)
const (
defaultQuickAckDelayMs = 1000
defaultFinalSplitMin = 1200
defaultFinalSplitMax = 3
defaultFinalSplitDelay = 500
defaultAckTemplate = "Got it. Working on it..."
)
type ResolvedChatBehavior struct {
Enabled bool
QuickAck ResolvedQuickAckConfig
FinalSplit ResolvedFinalSplitConfig
}
type ResolvedQuickAckConfig struct {
Enabled bool
MinDelayMs int
Templates []string
}
type ResolvedFinalSplitConfig struct {
Enabled bool
MinChars int
MaxMessages int
DelayMs int
}
type ChatBehaviorPreviewOptions struct {
Content string
IsStreaming bool
HasToolCalls bool
}
type ChatBehaviorPreview struct {
Resolved ResolvedChatBehavior `json:"resolved"`
Ack AckPreview `json:"ack"`
Split SplitPreview `json:"split"`
}
type AckPreview struct {
ShouldSend bool `json:"shouldSend"`
Content string `json:"content,omitempty"`
}
type SplitPreview struct {
Parts []string `json:"parts"`
}
func ResolveChatBehavior(global, override *config.ChatBehaviorConfig) ResolvedChatBehavior {
resolved := ResolvedChatBehavior{
QuickAck: ResolvedQuickAckConfig{
MinDelayMs: defaultQuickAckDelayMs,
Templates: []string{defaultAckTemplate},
},
FinalSplit: ResolvedFinalSplitConfig{
MinChars: defaultFinalSplitMin,
MaxMessages: defaultFinalSplitMax,
DelayMs: defaultFinalSplitDelay,
},
}
applyChatBehavior(&resolved, global)
applyChatBehavior(&resolved, override)
if !resolved.Enabled {
resolved.QuickAck.Enabled = false
resolved.FinalSplit.Enabled = false
}
if resolved.FinalSplit.MaxMessages < 1 {
resolved.FinalSplit.MaxMessages = 1
}
return resolved
}
func applyChatBehavior(dst *ResolvedChatBehavior, src *config.ChatBehaviorConfig) {
if src == nil {
return
}
if src.Enabled != nil {
dst.Enabled = *src.Enabled
}
if src.QuickAck != nil {
if src.QuickAck.Enabled != nil {
dst.QuickAck.Enabled = *src.QuickAck.Enabled
}
if src.QuickAck.MinDelayMs != nil {
dst.QuickAck.MinDelayMs = max(0, *src.QuickAck.MinDelayMs)
}
if len(src.QuickAck.Templates) > 0 {
dst.QuickAck.Templates = cleanTemplates(src.QuickAck.Templates)
}
}
if src.FinalSplit != nil {
if src.FinalSplit.Enabled != nil {
dst.FinalSplit.Enabled = *src.FinalSplit.Enabled
}
if src.FinalSplit.MinChars != nil {
dst.FinalSplit.MinChars = max(0, *src.FinalSplit.MinChars)
}
if src.FinalSplit.MaxMessages != nil {
dst.FinalSplit.MaxMessages = max(1, *src.FinalSplit.MaxMessages)
}
if src.FinalSplit.DelayMs != nil {
dst.FinalSplit.DelayMs = max(0, *src.FinalSplit.DelayMs)
}
}
}
func cleanTemplates(values []string) []string {
out := make([]string, 0, len(values))
for _, v := range values {
if s := strings.TrimSpace(v); s != "" {
out = append(out, s)
}
}
if len(out) == 0 {
return []string{defaultAckTemplate}
}
return out
}
func PreviewChatBehavior(global, override *config.ChatBehaviorConfig, opts ChatBehaviorPreviewOptions) ChatBehaviorPreview {
resolved := ResolveChatBehavior(global, override)
preview := ChatBehaviorPreview{
Resolved: resolved,
Split: SplitPreview{Parts: SplitFinalMessages(opts.Content, resolved.FinalSplit)},
}
if ShouldSendQuickAck(resolved, opts.IsStreaming) {
preview.Ack = AckPreview{ShouldSend: true, Content: resolved.QuickAck.Templates[0]}
}
return preview
}
func ShouldSendQuickAck(behavior ResolvedChatBehavior, streaming bool) bool {
return behavior.Enabled && behavior.QuickAck.Enabled && !streaming
}
func SplitFinalMessages(content string, cfg ResolvedFinalSplitConfig) []string {
if content == "" {
return nil
}
if !cfg.Enabled || len(content) < cfg.MinChars || cfg.MaxMessages <= 1 || hasUnsafeSplitMarkdown(content) {
return []string{content}
}
parts := splitParagraphs(content)
if len(parts) <= 1 || len(parts) > cfg.MaxMessages {
return []string{content}
}
return parts
}
func splitParagraphs(content string) []string {
raw := strings.Split(content, "\n\n")
parts := make([]string, 0, len(raw))
for _, part := range raw {
p := strings.TrimSpace(part)
if p == "" {
continue
}
parts = append(parts, p)
}
return parts
}
func hasUnsafeSplitMarkdown(content string) bool {
lines := strings.SplitSeq(content, "\n")
for line := range lines {
trimmed := strings.TrimSpace(line)
switch {
case strings.HasPrefix(trimmed, "```"):
return true
case strings.HasPrefix(trimmed, ">"):
return true
case strings.HasPrefix(trimmed, "- ") || strings.HasPrefix(trimmed, "* "):
return true
case len(trimmed) > 3 && trimmed[0] >= '0' && trimmed[0] <= '9' && strings.HasPrefix(trimmed[1:], ". "):
return true
case strings.Contains(trimmed, "|") && (strings.Contains(trimmed, "---") || strings.Count(trimmed, "|") >= 2):
return true
case strings.HasPrefix(trimmed, "{") || strings.HasPrefix(trimmed, "[") || strings.HasPrefix(trimmed, "<"):
return true
case strings.HasPrefix(trimmed, "http://") || strings.HasPrefix(trimmed, "https://"):
return true
}
}
return false
}
@@ -0,0 +1,101 @@
package channels
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)
func TestHandleAgentEvent_QuickAckNonStreamingOnly(t *testing.T) {
behavior := ResolvedChatBehavior{
Enabled: true,
QuickAck: ResolvedQuickAckConfig{
Enabled: true,
MinDelayMs: 0,
Templates: []string{"On it."},
},
}
mb := bus.New()
mgr := NewManager(mb)
mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"})
mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", map[string]string{"local_key": "chat-1/topic"}, uuid.Nil, false, false, true, behavior)
mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
got, ok := mb.SubscribeOutbound(ctx)
if !ok {
t.Fatal("expected quick acknowledgement outbound message")
}
if got.Content != "On it." || got.ChatID != "chat-1" || got.Metadata["local_key"] != "chat-1/topic" {
t.Fatalf("quick ack outbound = %+v, want content and routing metadata", got)
}
mb = bus.New()
mgr = NewManager(mb)
mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"})
mgr.RegisterRunWithBehavior("run-2", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, behavior)
mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-2", nil)
ctx, cancel = context.WithTimeout(context.Background(), 25*time.Millisecond)
defer cancel()
if got, ok := mb.SubscribeOutbound(ctx); ok {
t.Fatalf("streaming run emitted quick ack: %+v", got)
}
}
func TestUnregisterRun_CancelsPendingQuickAck(t *testing.T) {
mb := bus.New()
mgr := NewManager(mb)
mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"})
mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, false, false, true, ResolvedChatBehavior{
Enabled: true,
QuickAck: ResolvedQuickAckConfig{
Enabled: true,
MinDelayMs: 50,
Templates: []string{"On it."},
},
})
mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil)
mgr.UnregisterRun("run-1")
ctx, cancel := context.WithTimeout(context.Background(), 75*time.Millisecond)
defer cancel()
if got, ok := mb.SubscribeOutbound(ctx); ok {
t.Fatalf("unregistered run emitted quick ack: %+v", got)
}
}
func TestCancelQuickAck_BlocksInFlightSend(t *testing.T) {
mb := bus.New()
mgr := NewManager(mb)
rc := &RunContext{
ChannelName: "test",
ChatID: "chat-1",
ChatBehavior: ResolvedChatBehavior{
Enabled: true,
QuickAck: ResolvedQuickAckConfig{
Enabled: true,
Templates: []string{"On it."},
},
},
}
mgr.cancelQuickAck(rc)
mgr.sendQuickAck(rc)
ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond)
defer cancel()
if got, ok := mb.SubscribeOutbound(ctx); ok {
t.Fatalf("cancelled quick ack emitted message: %+v", got)
}
}
+137
View File
@@ -0,0 +1,137 @@
package channels
import (
"context"
"reflect"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/config"
)
func TestResolveChatBehavior_InheritsGlobalAndChannelOverride(t *testing.T) {
global := &config.ChatBehaviorConfig{
Enabled: new(true),
QuickAck: &config.QuickAckConfig{
Enabled: new(true),
MinDelayMs: new(750),
Templates: []string{"On it."},
},
FinalSplit: &config.FinalSplitConfig{
Enabled: new(true),
MinChars: new(1200),
MaxMessages: new(3),
DelayMs: new(400),
},
}
override := &config.ChatBehaviorConfig{
QuickAck: &config.QuickAckConfig{Enabled: new(false)},
FinalSplit: &config.FinalSplitConfig{
MaxMessages: new(2),
},
}
got := ResolveChatBehavior(global, override)
if !got.Enabled {
t.Fatal("Enabled = false, want true")
}
if got.QuickAck.Enabled {
t.Fatal("QuickAck.Enabled = true, want channel override false")
}
if got.QuickAck.MinDelayMs != 750 {
t.Fatalf("QuickAck.MinDelayMs = %d, want 750", got.QuickAck.MinDelayMs)
}
if got.FinalSplit.MaxMessages != 2 {
t.Fatalf("FinalSplit.MaxMessages = %d, want override 2", got.FinalSplit.MaxMessages)
}
if got.FinalSplit.MinChars != 1200 || got.FinalSplit.DelayMs != 400 {
t.Fatalf("FinalSplit inherited fields = %+v, want min=1200 delay=400", got.FinalSplit)
}
}
func TestSplitFinalMessages_ConservativeParagraphSplit(t *testing.T) {
cfg := ResolvedFinalSplitConfig{Enabled: true, MinChars: 20, MaxMessages: 3}
text := "First part is useful.\n\nSecond part is also useful.\n\nThird part closes it."
got := SplitFinalMessages(text, cfg)
want := []string{"First part is useful.", "Second part is also useful.", "Third part closes it."}
if !reflect.DeepEqual(got, want) {
t.Fatalf("SplitFinalMessages() = %#v, want %#v", got, want)
}
}
func TestSplitFinalMessages_DoesNotSplitUnsafeMarkdown(t *testing.T) {
cfg := ResolvedFinalSplitConfig{Enabled: true, MinChars: 10, MaxMessages: 3}
cases := map[string]string{
"fenced code": "Intro.\n\n```go\nfmt.Println(\"hi\")\n```\n\nDone.",
"table": "A | B\n--- | ---\n1 | 2\n\nDone.",
"list": "Intro.\n\n- one\n- two\n\nDone.",
"quote": "Intro.\n\n> quoted\n> text\n\nDone.",
"json": "Intro.\n\n{\"ok\": true}\n\nDone.",
"url paragraph": "Intro.\n\nhttps://example.com/a/b?c=d\n\nDone.",
}
for name, text := range cases {
t.Run(name, func(t *testing.T) {
got := SplitFinalMessages(text, cfg)
if len(got) != 1 || got[0] != text {
t.Fatalf("SplitFinalMessages() = %#v, want original single message", got)
}
})
}
}
func TestPreviewChatBehavior_NoSideEffects(t *testing.T) {
global := &config.ChatBehaviorConfig{
Enabled: new(true),
QuickAck: &config.QuickAckConfig{Enabled: new(true), Templates: []string{"Working."}},
FinalSplit: &config.FinalSplitConfig{Enabled: new(true), MinChars: new(10), MaxMessages: new(2)},
}
got := PreviewChatBehavior(global, nil, ChatBehaviorPreviewOptions{
Content: "Part one is long.\n\nPart two is long.",
IsStreaming: false,
HasToolCalls: true,
})
if !got.Ack.ShouldSend || got.Ack.Content != "Working." {
t.Fatalf("Ack preview = %+v, want send Working.", got.Ack)
}
if len(got.Split.Parts) != 2 {
t.Fatalf("Split parts = %#v, want two parts", got.Split.Parts)
}
}
func TestManagerResolveChatBehavior_UsesChannelOverride(t *testing.T) {
global := &config.ChatBehaviorConfig{
Enabled: new(true),
QuickAck: &config.QuickAckConfig{Enabled: new(true), Templates: []string{"global"}},
}
override := &config.ChatBehaviorConfig{
QuickAck: &config.QuickAckConfig{Enabled: new(true), Templates: []string{"channel"}},
}
mgr := NewManager(bus.New())
mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test", behavior: override})
got := mgr.ResolveChatBehavior("test", global)
if got.QuickAck.Templates[0] != "channel" {
t.Fatalf("QuickAck template = %q, want channel override", got.QuickAck.Templates[0])
}
}
type chatBehaviorTestChannel struct {
name string
behavior *config.ChatBehaviorConfig
}
func (c *chatBehaviorTestChannel) Name() string { return c.name }
func (c *chatBehaviorTestChannel) Type() string { return c.name }
func (c *chatBehaviorTestChannel) Start(context.Context) error { return nil }
func (c *chatBehaviorTestChannel) Stop(context.Context) error { return nil }
func (c *chatBehaviorTestChannel) Send(context.Context, bus.OutboundMessage) error { return nil }
func (c *chatBehaviorTestChannel) IsRunning() bool { return true }
func (c *chatBehaviorTestChannel) IsAllowed(string) bool { return true }
func (c *chatBehaviorTestChannel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.behavior }
+7 -4
View File
@@ -26,12 +26,12 @@ type Channel struct {
*channels.BaseChannel
session *discordgo.Session
config config.DiscordConfig
botUserID string // populated on start
placeholders sync.Map // placeholderKey string → messageID string
typingCtrls sync.Map // channelID string → *typing.Controller
botUserID string // populated on start
placeholders sync.Map // placeholderKey string → messageID string
typingCtrls sync.Map // channelID string → *typing.Controller
agentStore store.AgentStore // for agent key lookup (nil = writer commands disabled)
configPermStore store.ConfigPermissionStore // for group file writer management (nil = writer commands disabled)
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
// pairingService, pairingDebounce, approvedGroups, groupHistory, historyLimit, requireMention
// are inherited from channels.BaseChannel.
}
@@ -108,6 +108,9 @@ func (c *Channel) Start(_ context.Context) error {
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.config.ChatBehavior }
// SetPendingCompaction configures LLM-based auto-compaction for pending messages.
func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) {
if gh := c.GroupHistory(); gh != nil {
+14 -12
View File
@@ -18,18 +18,19 @@ type discordCreds struct {
// discordInstanceConfig maps the non-secret config JSONB from the channel_instances table.
type discordInstanceConfig struct {
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
MediaMaxBytes int64 `json:"media_max_bytes,omitempty"`
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
STTTenantID string `json:"stt_tenant_id,omitempty"`
STTTimeoutSeconds int `json:"stt_timeout_seconds,omitempty"`
VoiceAgentID string `json:"voice_agent_id,omitempty"`
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
MediaMaxBytes int64 `json:"media_max_bytes,omitempty"`
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
STTTenantID string `json:"stt_tenant_id,omitempty"`
STTTimeoutSeconds int `json:"stt_timeout_seconds,omitempty"`
VoiceAgentID string `json:"voice_agent_id,omitempty"`
}
// Factory creates a Discord channel from DB instance data (no extra stores).
@@ -82,6 +83,7 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
RequireMention: ic.RequireMention,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
MediaMaxBytes: ic.MediaMaxBytes,
STTProxyURL: ic.STTProxyURL,
STTAPIKey: ic.STTAPIKey,
+61 -3
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log/slog"
"strings"
"time"
"github.com/google/uuid"
@@ -38,6 +39,10 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
ctx = store.WithTenantID(ctx, rc.TenantID)
}
if eventType == protocol.AgentEventRunStarted {
m.scheduleQuickAck(rc)
}
// Forward to StreamingChannel (only when streaming is enabled for this run).
// Without this gate, channels that implement StreamingChannel but have streaming
// disabled (e.g. group_stream=false) would create stream messages AND emit
@@ -284,6 +289,11 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
return // streaming already delivered via chunks
}
m.cancelQuickAck(rc)
rc.mu.Lock()
rc.blockReplySent = true
rc.mu.Unlock()
// Build outbound metadata: copy routing fields but strip reply_to_message_id
// (block replies are standalone) and placeholder_key (reserve for final message).
// feishu_reply_target_id MUST be preserved so intermediate block replies for
@@ -354,10 +364,59 @@ 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)
}
}
func (m *Manager) scheduleQuickAck(rc *RunContext) {
if !ShouldSendQuickAck(rc.ChatBehavior, rc.Streaming) {
return
}
delay := time.Duration(rc.ChatBehavior.QuickAck.MinDelayMs) * time.Millisecond
if delay <= 0 {
m.sendQuickAck(rc)
return
}
rc.mu.Lock()
if rc.ackTimer == nil && !rc.ackSent && !rc.blockReplySent {
rc.ackTimer = time.AfterFunc(delay, func() {
m.sendQuickAck(rc)
})
}
rc.mu.Unlock()
}
func (m *Manager) cancelQuickAck(rc *RunContext) {
rc.mu.Lock()
rc.ackCancelled = true
if rc.ackTimer != nil {
rc.ackTimer.Stop()
rc.ackTimer = nil
}
rc.mu.Unlock()
}
func (m *Manager) sendQuickAck(rc *RunContext) {
rc.mu.Lock()
if rc.ackCancelled || rc.ackSent || rc.blockReplySent || !ShouldSendQuickAck(rc.ChatBehavior, rc.Streaming) || len(rc.ChatBehavior.QuickAck.Templates) == 0 {
rc.mu.Unlock()
return
}
content := rc.ChatBehavior.QuickAck.Templates[0]
rc.ackSent = true
rc.ackTimer = nil
rc.mu.Unlock()
m.bus.PublishOutbound(bus.OutboundMessage{
Channel: rc.ChannelName,
ChatID: rc.ChatID,
Content: content,
Metadata: copyRoutingMeta(rc.Metadata),
TenantID: rc.TenantID,
})
}
// extractPayloadString extracts a string field from a payload (map[string]string or map[string]interface{}).
func extractPayloadString(payload any, key string) string {
switch p := payload.(type) {
@@ -371,7 +430,6 @@ func extractPayloadString(payload any, key string) string {
return ""
}
// toolStatusMap maps builtin tool names to user-friendly status messages.
var toolStatusMap = map[string]string{
// Filesystem
@@ -400,8 +458,8 @@ var toolStatusMap = map[string]string{
// Browser
"browser": "🌐 Browsing...",
// Delegation & teams
"spawn": "👥 Delegating task...",
"team_tasks": "📋 Managing team tasks...",
"spawn": "👥 Delegating task...",
"team_tasks": "📋 Managing team tasks...",
// Sessions
"sessions_list": "📋 Listing sessions...",
"session_status": "📋 Checking session...",
+25 -22
View File
@@ -21,28 +21,29 @@ type feishuCreds struct {
// feishuInstanceConfig maps the non-secret config JSONB from the channel_instances table.
type feishuInstanceConfig struct {
Domain string `json:"domain,omitempty"`
ConnectionMode string `json:"connection_mode,omitempty"`
WebhookPort int `json:"webhook_port,omitempty"`
WebhookPath string `json:"webhook_path,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
GroupAllowFrom []string `json:"group_allow_from,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
TopicSessionMode string `json:"topic_session_mode,omitempty"`
TextChunkLimit int `json:"text_chunk_limit,omitempty"`
MediaMaxMB int `json:"media_max_mb,omitempty"`
RenderMode string `json:"render_mode,omitempty"`
Streaming *bool `json:"streaming,omitempty"`
ReactionLevel string `json:"reaction_level,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
STTTenantID string `json:"stt_tenant_id,omitempty"`
STTTimeoutSeconds int `json:"stt_timeout_seconds,omitempty"`
VoiceAgentID string `json:"voice_agent_id,omitempty"`
Domain string `json:"domain,omitempty"`
ConnectionMode string `json:"connection_mode,omitempty"`
WebhookPort int `json:"webhook_port,omitempty"`
WebhookPath string `json:"webhook_path,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
GroupAllowFrom []string `json:"group_allow_from,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
TopicSessionMode string `json:"topic_session_mode,omitempty"`
TextChunkLimit int `json:"text_chunk_limit,omitempty"`
MediaMaxMB int `json:"media_max_mb,omitempty"`
RenderMode string `json:"render_mode,omitempty"`
Streaming *bool `json:"streaming,omitempty"`
ReactionLevel string `json:"reaction_level,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
STTTenantID string `json:"stt_tenant_id,omitempty"`
STTTimeoutSeconds int `json:"stt_timeout_seconds,omitempty"`
VoiceAgentID string `json:"voice_agent_id,omitempty"`
}
// Factory creates a Feishu/Lark channel from DB instance data.
@@ -89,6 +90,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
ReactionLevel: ic.ReactionLevel,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
STTProxyURL: ic.STTProxyURL,
STTAPIKey: ic.STTAPIKey,
STTTenantID: ic.STTTenantID,
@@ -160,6 +162,7 @@ func FactoryWithPendingStoreAndAudio(pendingStore store.PendingMessageStore, aud
ReactionLevel: ic.ReactionLevel,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
STTProxyURL: ic.STTProxyURL,
STTAPIKey: ic.STTAPIKey,
STTTenantID: ic.STTTenantID,
+7 -4
View File
@@ -40,10 +40,10 @@ type Channel struct {
cfg config.FeishuConfig
client *LarkClient
botOpenID string
senderCache sync.Map // open_id → *senderCacheEntry
dedup sync.Map // message_id → struct{}
reactions sync.Map // chatID → *reactionState
docCache *docCache // LRU+TTL cache for Lark docx raw_content lookups
senderCache sync.Map // open_id → *senderCacheEntry
dedup sync.Map // message_id → struct{}
reactions sync.Map // chatID → *reactionState
docCache *docCache // LRU+TTL cache for Lark docx raw_content lookups
agentStore store.AgentStore // optional — agent key → UUID lookup for writer commands
configPermStore store.ConfigPermissionStore // optional — group file writer ACL for /addwriter et al.
groupAllowList []string // Feishu-specific: per-group sender allowlist (separate from BaseChannel allowList)
@@ -160,6 +160,9 @@ func (c *Channel) Start(ctx context.Context) error {
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.cfg.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.cfg.ChatBehavior }
// SetPendingCompaction configures LLM-based auto-compaction for pending messages.
func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) {
if gh := c.GroupHistory(); gh != nil {
+6
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log/slog"
"sync"
"time"
"github.com/google/uuid"
@@ -36,7 +37,12 @@ type RunContext struct {
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
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)
+4
View File
@@ -11,6 +11,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/store"
)
@@ -332,6 +333,9 @@ func (ch *Channel) sendPrivateReply(ctx context.Context, senderID, conversationI
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (ch *Channel) BlockReplyEnabled() *bool { return ch.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (ch *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return ch.config.ChatBehavior }
// WebhookHandler returns the shared webhook path and global router as handler.
// Only the first pancake instance mounts the route; others return ("", nil).
func (ch *Channel) WebhookHandler() (string, http.Handler) {
+11 -6
View File
@@ -3,7 +3,11 @@
// A single Pancake API key gives access to all connected platforms — no per-platform OAuth needed.
package pancake
import "encoding/json"
import (
"encoding/json"
"github.com/nextlevelbuilder/goclaw/internal/config"
)
// pancakeCreds holds encrypted credentials stored in channel_instances.credentials.
type pancakeCreds struct {
@@ -31,11 +35,12 @@ type pancakeInstanceConfig struct {
Filter string `json:"filter"` // "all" | "keyword" (default: all)
Keywords []string `json:"keywords"` // required when filter = "keyword"
} `json:"comment_reply_options"`
PrivateReplyMessage string `json:"private_reply_message,omitempty"` // custom DM text; defaults to built-in message. Supports {{commenter_name}} / {{post_title}} vars.
AutoReactOptions *AutoReactOptions `json:"auto_react_options,omitempty"`
PostContextCacheTTL string `json:"post_context_cache_ttl,omitempty"` // e.g. "30m"; defaults to 15m
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
PrivateReplyMessage string `json:"private_reply_message,omitempty"` // custom DM text; defaults to built-in message. Supports {{commenter_name}} / {{post_title}} vars.
AutoReactOptions *AutoReactOptions `json:"auto_react_options,omitempty"`
PostContextCacheTTL string `json:"post_context_cache_ttl,omitempty"` // e.g. "30m"; defaults to 15m
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
}
// AutoReactOptions holds per-page scope filters for Facebook auto-react.
+7
View File
@@ -38,6 +38,13 @@ func CopyFinalRoutingMeta(src map[string]string) map[string]string {
return copySelectedMeta(src, finalReplyMetaKeys)
}
// CopyFollowupRoutingMeta copies routing metadata for additional outbound
// messages after the first final reply. It preserves thread/topic routing but
// does not reuse placeholder or reply-to metadata reserved for the first reply.
func CopyFollowupRoutingMeta(src map[string]string) map[string]string {
return copyRoutingMeta(src)
}
// copyRoutingMeta copies only the subset safe for intermediate block replies,
// retries, and placeholder updates.
func copyRoutingMeta(src map[string]string) map[string]string {
+31 -2
View File
@@ -1,12 +1,22 @@
package channels
import "github.com/google/uuid"
import (
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/config"
)
// --- Run tracking for streaming/reaction event forwarding ---
// RegisterRun associates a run ID with a channel context so agent events
// (chunks, tool calls, completion) can be forwarded to the originating channel.
func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, metadata map[string]string, tenantID uuid.UUID, streaming, blockReply, toolStatus bool) {
m.RegisterRunWithBehavior(runID, channelName, chatID, messageID, metadata, tenantID, streaming, blockReply, toolStatus, ResolvedChatBehavior{})
}
// 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) {
m.runs.Store(runID, &RunContext{
ChannelName: channelName,
ChatID: chatID,
@@ -16,12 +26,17 @@ func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, meta
Streaming: streaming,
BlockReplyEnabled: blockReply,
ToolStatusEnabled: toolStatus,
ChatBehavior: chatBehavior,
})
}
// UnregisterRun removes a run tracking entry.
func (m *Manager) UnregisterRun(runID string) {
m.runs.Delete(runID)
if val, ok := m.runs.LoadAndDelete(runID); ok {
if rc, ok := val.(*RunContext); ok {
m.cancelQuickAck(rc)
}
}
}
// IsStreamingChannel checks if a named channel implements StreamingChannel
@@ -56,3 +71,17 @@ func (m *Manager) ResolveBlockReply(channelName string, globalDefault *bool) boo
}
return globalDefault != nil && *globalDefault
}
// ResolveChatBehavior checks per-channel override, then falls back to gateway config.
func (m *Manager) ResolveChatBehavior(channelName string, globalDefault *config.ChatBehaviorConfig) ResolvedChatBehavior {
var override *config.ChatBehaviorConfig
m.mu.RLock()
ch, exists := m.channels[channelName]
m.mu.RUnlock()
if exists {
if bc, ok := ch.(ChatBehaviorChannel); ok {
override = bc.ChatBehaviorConfig()
}
}
return ResolveChatBehavior(globalDefault, override)
}
+15 -12
View File
@@ -19,18 +19,19 @@ type slackCreds struct {
// slackInstanceConfig maps the non-secret config JSONB from the channel_instances table.
type slackInstanceConfig struct {
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
AllowFrom []string `json:"allow_from,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"`
NativeStream *bool `json:"native_stream,omitempty"`
ReactionLevel string `json:"reaction_level,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
DebounceDelay *int `json:"debounce_delay,omitempty"`
ThreadTTL *int `json:"thread_ttl,omitempty"`
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
AllowFrom []string `json:"allow_from,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"`
NativeStream *bool `json:"native_stream,omitempty"`
ReactionLevel string `json:"reaction_level,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
DebounceDelay *int `json:"debounce_delay,omitempty"`
ThreadTTL *int `json:"thread_ttl,omitempty"`
}
// Factory creates a Slack channel from DB instance data.
@@ -72,6 +73,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
NativeStream: ic.NativeStream,
ReactionLevel: ic.ReactionLevel,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
DebounceDelay: ic.DebounceDelay,
ThreadTTL: ic.ThreadTTL,
}
@@ -129,6 +131,7 @@ func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.Ch
NativeStream: ic.NativeStream,
ReactionLevel: ic.ReactionLevel,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
DebounceDelay: ic.DebounceDelay,
ThreadTTL: ic.ThreadTTL,
}
+4
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
@@ -54,6 +55,9 @@ func (c *Channel) HandleMessage(senderID, chatID, content string, mediaPaths []s
// BlockReplyEnabled returns the per-channel block_reply override.
func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.config.ChatBehavior }
// resolveDisplayName fetches and caches the Slack display name for a user ID.
func (c *Channel) resolveDisplayName(userID string) string {
c.userCacheMu.RLock()
+3
View File
@@ -378,6 +378,9 @@ func (c *Channel) ReasoningStreamEnabled() bool {
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.config.ChatBehavior }
// SetPendingCompaction configures LLM-based auto-compaction for pending messages.
func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) {
if gh := c.GroupHistory(); gh != nil {
+34 -32
View File
@@ -20,24 +20,25 @@ 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"`
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)
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).
@@ -96,25 +97,26 @@ 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,
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,
ForceIPv4: ic.ForceIPv4,
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).
+14 -8
View File
@@ -14,12 +14,13 @@ import (
// whatsappInstanceConfig maps the non-secret config JSONB from the channel_instances table.
type whatsappInstanceConfig struct {
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,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"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
}
// FactoryWithDB returns a ChannelFactory with DB access for whatsmeow auth state.
@@ -44,14 +45,18 @@ func FactoryWithDBAudio(db *sql.DB, pendingStore store.PendingMessageStore, dial
// Detect old bridge_url config and give clear migration error.
if len(cfg) > 0 {
var legacy struct{ BridgeURL string `json:"bridge_url"` }
var legacy struct {
BridgeURL string `json:"bridge_url"`
}
if json.Unmarshal(cfg, &legacy) == nil && legacy.BridgeURL != "" {
return nil, fmt.Errorf("whatsapp: bridge_url is no longer supported — " +
"WhatsApp now runs natively via whatsmeow. Remove bridge_url from config")
}
}
if len(creds) > 0 {
var legacy struct{ BridgeURL string `json:"bridge_url"` }
var legacy struct {
BridgeURL string `json:"bridge_url"`
}
if json.Unmarshal(creds, &legacy) == nil && legacy.BridgeURL != "" {
return nil, fmt.Errorf("whatsapp: bridge_url is no longer supported — " +
"WhatsApp now runs natively via whatsmeow. Remove bridge_url from credentials")
@@ -66,6 +71,7 @@ func FactoryWithDBAudio(db *sql.DB, pendingStore store.PendingMessageStore, dial
RequireMention: ic.RequireMention,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
}
// DB instances default to "pairing" for groups (secure by default).
if waCfg.GroupPolicy == "" {
+11 -8
View File
@@ -35,14 +35,14 @@ func init() {
// Auth state is stored in PostgreSQL (standard) or SQLite (desktop).
type Channel struct {
*channels.BaseChannel
client *whatsmeow.Client
container *sqlstore.Container
config config.WhatsAppConfig
mu sync.Mutex
ctx context.Context
cancel context.CancelFunc
parentCtx context.Context // stored from Start() for Reauth() context chain
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
client *whatsmeow.Client
container *sqlstore.Container
config config.WhatsAppConfig
mu sync.Mutex
ctx context.Context
cancel context.CancelFunc
parentCtx context.Context // stored from Start() for Reauth() context chain
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
builtinToolStore store.BuiltinToolStore // reads stt settings (whatsapp_enabled) per voice message; nil = opt-out
// QR state
@@ -146,6 +146,9 @@ func (c *Channel) Start(ctx context.Context) error {
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.config.ChatBehavior }
// Stop gracefully shuts down the WhatsApp channel.
func (c *Channel) Stop(_ context.Context) error {
slog.Info("stopping whatsapp channel")
+7 -5
View File
@@ -18,11 +18,12 @@ type zaloCreds struct {
// zaloInstanceConfig maps the non-secret config JSONB from the channel_instances table.
type zaloInstanceConfig struct {
DMPolicy string `json:"dm_policy,omitempty"`
WebhookURL string `json:"webhook_url,omitempty"`
MediaMaxMB int `json:"media_max_mb,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
DMPolicy string `json:"dm_policy,omitempty"`
WebhookURL string `json:"webhook_url,omitempty"`
MediaMaxMB int `json:"media_max_mb,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
}
// Factory creates a Zalo OA channel from DB instance data.
@@ -55,6 +56,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
WebhookSecret: c.WebhookSecret,
MediaMaxMB: ic.MediaMaxMB,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
}
ch, err := New(zCfg, msgBus, pairingSvc)
@@ -71,6 +71,9 @@ func New(cfg config.ZaloPersonalConfig, msgBus *bus.MessageBus, pairingSvc store
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.config.ChatBehavior }
// session returns the current session snapshot (thread-safe).
func (c *Channel) session() *protocol.Session {
c.mu.RLock()
+12 -9
View File
@@ -13,20 +13,21 @@ import (
// zaloCreds maps the credentials JSON from the channel_instances table.
type zaloCreds struct {
IMEI string `json:"imei"`
IMEI string `json:"imei"`
Cookie *protocol.CookieUnion `json:"cookie"`
UserAgent string `json:"userAgent"`
Language *string `json:"language,omitempty"`
UserAgent string `json:"userAgent"`
Language *string `json:"language,omitempty"`
}
// zaloInstanceConfig maps the config JSONB from the channel_instances table.
type zaloInstanceConfig struct {
DMPolicy string `json:"dm_policy,omitempty"`
GroupPolicy string `json:"group_policy,omitempty"`
RequireMention *bool `json:"require_mention,omitempty"`
HistoryLimit int `json:"history_limit,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,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"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
ChatBehavior *config.ChatBehaviorConfig `json:"chat_behavior,omitempty"`
}
// Factory creates a Zalo Personal channel from DB instance data.
@@ -62,6 +63,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
RequireMention: ic.RequireMention,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
}
ch, err := New(zaloCfg, msgBus, pairingSvc, nil)
@@ -112,6 +114,7 @@ func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.Ch
RequireMention: ic.RequireMention,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
}
ch, err := New(zaloCfg, msgBus, pairingSvc, pendingStore)
+20 -15
View File
@@ -39,13 +39,14 @@ var apiBase = "https://bot-api.zaloplatforms.com"
// Channel connects to the Zalo OA Bot API.
type Channel struct {
*channels.BaseChannel
token string
dmPolicy string
mediaMaxMB int
blockReply *bool
stopCh chan struct{}
client *http.Client
pollClient *http.Client
token string
dmPolicy string
mediaMaxMB int
blockReply *bool
chatBehavior *config.ChatBehaviorConfig
stopCh chan struct{}
client *http.Client
pollClient *http.Client
// pairingService, pairingDebounce are inherited from channels.BaseChannel.
}
@@ -69,14 +70,15 @@ func New(cfg config.ZaloConfig, msgBus *bus.MessageBus, pairingSvc store.Pairing
}
ch := &Channel{
BaseChannel: base,
token: cfg.Token,
dmPolicy: dmPolicy,
mediaMaxMB: mediaMax,
blockReply: cfg.BlockReply,
stopCh: make(chan struct{}),
client: &http.Client{Timeout: 60 * time.Second},
pollClient: &http.Client{Timeout: 0},
BaseChannel: base,
token: cfg.Token,
dmPolicy: dmPolicy,
mediaMaxMB: mediaMax,
blockReply: cfg.BlockReply,
chatBehavior: cfg.ChatBehavior,
stopCh: make(chan struct{}),
client: &http.Client{Timeout: 60 * time.Second},
pollClient: &http.Client{Timeout: 0},
}
ch.SetPairingService(pairingSvc)
return ch, nil
@@ -85,6 +87,9 @@ func New(cfg config.ZaloConfig, msgBus *bus.MessageBus, pairingSvc store.Pairing
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (c *Channel) BlockReplyEnabled() *bool { return c.blockReply }
// ChatBehaviorConfig returns the per-channel chat_behavior override.
func (c *Channel) ChatBehaviorConfig() *config.ChatBehaviorConfig { return c.chatBehavior }
// Start begins polling for Zalo updates.
func (c *Channel) Start(ctx context.Context) error {
slog.Info("starting zalo bot (polling mode)")
+49 -18
View File
@@ -1,5 +1,28 @@
package config
// ChatBehaviorConfig controls optional human-like channel delivery behavior.
// Pointer fields allow per-channel overrides to inherit gateway defaults.
type ChatBehaviorConfig struct {
Enabled *bool `json:"enabled,omitempty"`
QuickAck *QuickAckConfig `json:"quick_ack,omitempty"`
FinalSplit *FinalSplitConfig `json:"final_split,omitempty"`
}
// QuickAckConfig controls one short acknowledgement before longer non-streaming runs.
type QuickAckConfig struct {
Enabled *bool `json:"enabled,omitempty"`
MinDelayMs *int `json:"min_delay_ms,omitempty"`
Templates []string `json:"templates,omitempty"`
}
// FinalSplitConfig controls semantic splitting of final channel replies.
type FinalSplitConfig struct {
Enabled *bool `json:"enabled,omitempty"`
MinChars *int `json:"min_chars,omitempty"`
MaxMessages *int `json:"max_messages,omitempty"`
DelayMs *int `json:"delay_ms,omitempty"`
}
// PendingCompactionConfig configures LLM-based compaction of pending group messages.
// When a group accumulates more than Threshold pending messages, older messages are
// summarized by an LLM and replaced with a compact summary, keeping KeepRecent raw messages.
@@ -42,6 +65,7 @@ type TelegramConfig struct {
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.
@@ -103,6 +127,7 @@ type DiscordConfig struct {
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)
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // max media download size (default 25MB)
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
@@ -126,6 +151,7 @@ type SlackConfig struct {
NativeStream *bool `json:"native_stream,omitempty"` // use Slack ChatStreamer API if available (default false)
ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full"
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
DebounceDelay *int `json:"debounce_delay,omitempty"` // ms delay before dispatching rapid messages (default 300, 0=disabled)
ThreadTTL *int `json:"thread_ttl,omitempty"` // hours before thread participation expires (default 24, 0=disabled — always require @mention)
MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // max file download size in bytes (default 20MB)
@@ -140,6 +166,7 @@ type WhatsAppConfig struct {
RequireMention *bool `json:"require_mention,omitempty"` // only respond in groups when bot is @mentioned (default false)
HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 200, 0=disabled)
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
}
type ZaloConfig struct {
@@ -149,8 +176,9 @@ type ZaloConfig struct {
DMPolicy string `json:"dm_policy,omitempty"` // "pairing" (default), "allowlist", "open", "disabled"
WebhookURL string `json:"webhook_url,omitempty"`
WebhookSecret string `json:"webhook_secret,omitempty"`
MediaMaxMB int `json:"media_max_mb,omitempty"` // default 5
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
MediaMaxMB int `json:"media_max_mb,omitempty"` // default 5
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
}
type ZaloPersonalConfig struct {
@@ -162,6 +190,7 @@ type ZaloPersonalConfig struct {
HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled)
CredentialsPath string `json:"credentials_path,omitempty"` // path to saved cookies JSON
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
}
type FeishuConfig struct {
@@ -186,7 +215,8 @@ type FeishuConfig struct {
Streaming *bool `json:"streaming,omitempty"` // default true
ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full" — typing emoji reactions
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // override gateway chat behavior (nil = inherit)
STTProxyURL string `json:"stt_proxy_url,omitempty"`
STTAPIKey string `json:"stt_api_key,omitempty"`
STTTenantID string `json:"stt_tenant_id,omitempty"`
@@ -362,21 +392,22 @@ type QuotaConfig struct {
// GatewayConfig controls the gateway server.
type GatewayConfig struct {
Host string `json:"host"`
Port int `json:"port"`
Token string `json:"token,omitempty"` // bearer token for WS/HTTP auth
OwnerIDs []string `json:"owner_ids,omitempty"` // sender IDs considered "owner"
AllowedOrigins []string `json:"allowed_origins,omitempty"` // WebSocket CORS whitelist (empty = allow all)
MaxMessageChars int `json:"max_message_chars,omitempty"` // max user message characters (default 32000)
RateLimitRPM int `json:"rate_limit_rpm,omitempty"` // rate limit: requests per minute per user (default 20, 0 = disabled)
InjectionAction string `json:"injection_action,omitempty"` // prompt injection action: "log", "warn" (default), "block", "off"
InboundDebounceMs int `json:"inbound_debounce_ms,omitempty"` // silence-window in ms that merges rapid channel/Web Chat messages from the same sender/session; 0 disables for text but media-bearing messages still honor a built-in media floor so multi-attachment bursts (#63) coalesce into a single agent run. Agents may override via per-agent agent_config.inbound_debounce_ms.
Quota *QuotaConfig `json:"quota,omitempty"` // per-user/group request quotas
BlockReply *bool `json:"block_reply,omitempty"` // deliver intermediate text during tool iterations (default false)
ToolStatus *bool `json:"tool_status,omitempty"` // show tool name in streaming preview during tool execution (default true)
TaskRecoveryIntervalSec int `json:"task_recovery_interval_sec,omitempty"` // team task recovery ticker interval in seconds (default 300 = 5min)
BackgroundProvider string `json:"background_provider,omitempty"` // LLM provider for background workers (vault enrichment, consolidation)
BackgroundModel string `json:"background_model,omitempty"` // LLM model for background workers
Host string `json:"host"`
Port int `json:"port"`
Token string `json:"token,omitempty"` // bearer token for WS/HTTP auth
OwnerIDs []string `json:"owner_ids,omitempty"` // sender IDs considered "owner"
AllowedOrigins []string `json:"allowed_origins,omitempty"` // WebSocket CORS whitelist (empty = allow all)
MaxMessageChars int `json:"max_message_chars,omitempty"` // max user message characters (default 32000)
RateLimitRPM int `json:"rate_limit_rpm,omitempty"` // rate limit: requests per minute per user (default 20, 0 = disabled)
InjectionAction string `json:"injection_action,omitempty"` // prompt injection action: "log", "warn" (default), "block", "off"
InboundDebounceMs int `json:"inbound_debounce_ms,omitempty"` // silence-window in ms that merges rapid channel/Web Chat messages from the same sender/session; 0 disables for text but media-bearing messages still honor a built-in media floor so multi-attachment bursts (#63) coalesce into a single agent run. Agents may override via per-agent agent_config.inbound_debounce_ms.
Quota *QuotaConfig `json:"quota,omitempty"` // per-user/group request quotas
BlockReply *bool `json:"block_reply,omitempty"` // deliver intermediate text during tool iterations (default false)
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // human-like channel delivery behavior (default disabled)
ToolStatus *bool `json:"tool_status,omitempty"` // show tool name in streaming preview during tool execution (default true)
TaskRecoveryIntervalSec int `json:"task_recovery_interval_sec,omitempty"` // team task recovery ticker interval in seconds (default 300 = 5min)
BackgroundProvider string `json:"background_provider,omitempty"` // LLM provider for background workers (vault enrichment, consolidation)
BackgroundModel string `json:"background_model,omitempty"` // LLM model for background workers
}
// ToolsConfig controls tool availability, policy, and web search.
+85
View File
@@ -0,0 +1,85 @@
package methods
import (
"context"
"encoding/json"
"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"
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)
// ChatBehaviorMethods handles dashboard previews for channel delivery behavior.
type ChatBehaviorMethods struct {
cfg *config.Config
channelMgr *channels.Manager
}
func NewChatBehaviorMethods(cfg *config.Config, channelMgr *channels.Manager) *ChatBehaviorMethods {
return &ChatBehaviorMethods{cfg: cfg, channelMgr: channelMgr}
}
func (m *ChatBehaviorMethods) Register(router *gateway.MethodRouter) {
router.Register(protocol.MethodChatBehaviorPreview, m.requireMasterScope(m.requireOwner(m.handlePreview)))
}
func (m *ChatBehaviorMethods) requireOwner(next gateway.MethodHandler) gateway.MethodHandler {
return func(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) {
if !client.IsOwner() {
locale := store.LocaleFromContext(ctx)
client.SendResponse(protocol.NewErrorResponse(
req.ID,
protocol.ErrUnauthorized,
i18n.T(locale, i18n.MsgPermissionDenied, req.Method),
))
return
}
next(ctx, client, req)
}
}
func (m *ChatBehaviorMethods) requireMasterScope(next gateway.MethodHandler) gateway.MethodHandler {
return func(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) {
if !store.IsMasterScope(ctx) {
locale := store.LocaleFromContext(ctx)
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgConfigMasterScopeOnly)))
return
}
next(ctx, client, req)
}
}
func (m *ChatBehaviorMethods) handlePreview(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) {
var params struct {
Channel string `json:"channel"`
Content string `json:"content"`
IsStreaming bool `json:"isStreaming"`
HasToolCalls bool `json:"hasToolCalls"`
Config *config.ChatBehaviorConfig `json:"config"`
}
if req.Params != nil {
_ = json.Unmarshal(req.Params, &params)
}
var resolved channels.ResolvedChatBehavior
if params.Config != nil {
resolved = channels.ResolveChatBehavior(params.Config, nil)
} else if m.channelMgr != nil {
resolved = m.channelMgr.ResolveChatBehavior(params.Channel, m.cfg.Gateway.ChatBehavior)
} else {
resolved = channels.ResolveChatBehavior(m.cfg.Gateway.ChatBehavior, nil)
}
preview := channels.ChatBehaviorPreview{
Resolved: resolved,
Split: channels.SplitPreview{
Parts: channels.SplitFinalMessages(params.Content, resolved.FinalSplit),
},
}
if channels.ShouldSendQuickAck(resolved, params.IsStreaming) {
preview.Ack = channels.AckPreview{ShouldSend: true, Content: resolved.QuickAck.Templates[0]}
}
client.SendResponse(protocol.NewOKResponse(req.ID, preview))
}
+1
View File
@@ -211,6 +211,7 @@ func isAdminMethod(method string) bool {
protocol.MethodConfigPatch,
protocol.MethodConfigSchema,
protocol.MethodConfigDefaults,
protocol.MethodChatBehaviorPreview,
protocol.MethodConfigPermissionsList,
protocol.MethodConfigPermissionsCheck,
protocol.MethodConfigPermissionsGrant,