mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-20 02:17:24 +00:00
feat(channels): add sidecar delivery behavior overrides
Closes #144. - Retire user-facing Tool Status Messages and deterministic tool-status channel text. - Add sidecar-generated Quick Acknowledgement and Intermediate Replies with provider/model/timeout/token/char caps. - Resolve delivery behavior as Channel > Agent > Workspace with agent overrides in other_config.delivery_behavior. - Preserve legacy block_reply defaults for Intermediate Replies.
This commit is contained in:
1 parent
23b18fa6cd
commit
fdba1c2138
46 files changed
+1960
-201
No files matched your search
@@ -15,6 +15,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/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
@@ -26,7 +27,7 @@ import (
|
||||
// and routes them through the scheduler/agent loop, then publishes the response back.
|
||||
// Also handles subagent announcements: routes them through the parent agent's session
|
||||
// (matching TS subagent-announce.ts pattern) so the agent can reformulate for the user.
|
||||
func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor, subagentMgr *tools.SubagentManager, usageCapSvc *usagecaps.Service) {
|
||||
func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor, subagentMgr *tools.SubagentManager, usageCapSvc *usagecaps.Service, providerReg *providers.Registry) {
|
||||
slog.Info("inbound message consumer started")
|
||||
|
||||
// Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts).
|
||||
@@ -59,6 +60,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
ContactCollector: contactCollector,
|
||||
SubagentMgr: subagentMgr,
|
||||
UsageCaps: usageCapSvc,
|
||||
ProviderReg: providerReg,
|
||||
GetAnnounceMu: getAnnounceMu,
|
||||
}
|
||||
|
||||
|
||||
@@ -7,6 +7,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/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
@@ -30,6 +31,7 @@ type ConsumerDeps struct {
|
||||
TaskRunSessions sync.Map
|
||||
SubagentMgr *tools.SubagentManager
|
||||
UsageCaps *usagecaps.Service
|
||||
ProviderReg *providers.Registry
|
||||
BgWg sync.WaitGroup
|
||||
GetAnnounceMu func(string) *sync.Mutex
|
||||
}
|
||||
@@ -242,14 +242,17 @@ func processNormalMessage(
|
||||
if lk := msg.Metadata["local_key"]; lk != "" {
|
||||
chatIDForRun = lk
|
||||
}
|
||||
blockReply := deps.ChannelMgr != nil && deps.ChannelMgr.ResolveBlockReply(msg.Channel, deps.Cfg.Gateway.BlockReply)
|
||||
chatBehavior := channels.ResolvedChatBehavior{}
|
||||
if deps.ChannelMgr != nil {
|
||||
chatBehavior = deps.ChannelMgr.ResolveChatBehavior(msg.Channel, deps.Cfg.Gateway.ChatBehavior)
|
||||
workspaceBehavior := channels.ChatBehaviorConfigWithIntermediateDefault(deps.Cfg.Gateway.ChatBehavior, deps.Cfg.Gateway.BlockReply)
|
||||
agentBehavior := channels.ParseAgentDeliveryBehaviorConfig(agentLoop.OtherConfig())
|
||||
chatBehavior = deps.ChannelMgr.ResolveChatBehaviorWithAgent(msg.Channel, workspaceBehavior, agentBehavior)
|
||||
}
|
||||
blockReply := deps.ChannelMgr != nil && chatBehavior.IntermediateReplies.Enabled
|
||||
deliveryRuntime := buildDeliveryRuntime(ctx, deps, agentLoop, chatBehavior, msg, peerKind, resolveChannelType(deps.ChannelMgr, msg.Channel), agentID)
|
||||
toolStatus := deps.Cfg.Gateway.ToolStatus == nil || *deps.Cfg.Gateway.ToolStatus // default true
|
||||
if deps.ChannelMgr != nil {
|
||||
deps.ChannelMgr.RegisterRunWithBehavior(runID, msg.Channel, chatIDForRun, messageID, outMeta, msg.TenantID, channelStream, blockReply, toolStatus, chatBehavior, reasoningDelivery)
|
||||
deps.ChannelMgr.RegisterRunWithDelivery(runID, msg.Channel, chatIDForRun, messageID, outMeta, msg.TenantID, channelStream, blockReply, toolStatus, chatBehavior, deliveryRuntime, reasoningDelivery)
|
||||
}
|
||||
|
||||
// Group-aware system prompt: help the LLM adapt tone and behavior for group chats.
|
||||
@@ -608,6 +611,60 @@ func processNormalMessage(
|
||||
}(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, msg.Content, outMeta, blockReply, chatBehavior, channelStream, ptd, msg.TenantID, agentLoop.UUID(), agentLoop.OtherConfig())
|
||||
}
|
||||
|
||||
func buildDeliveryRuntime(ctx context.Context, deps *ConsumerDeps, agentLoop agent.Agent, behavior channels.ResolvedChatBehavior, msg bus.InboundMessage, peerKind, channelType, agentKey string) channels.DeliveryRuntime {
|
||||
locale := msg.Metadata["locale"]
|
||||
if locale == "" {
|
||||
locale = "auto"
|
||||
}
|
||||
runtime := channels.DeliveryRuntime{
|
||||
Locale: locale,
|
||||
Inbound: msg.Content,
|
||||
PeerKind: peerKind,
|
||||
Channel: channelType,
|
||||
AgentName: agentKey,
|
||||
}
|
||||
if behavior.Enabled && behavior.QuickAck.Enabled {
|
||||
switch behavior.QuickAck.Mode {
|
||||
case channels.QuickAckModeLLMGenerated, channels.QuickAckModeSidecar, "":
|
||||
runtime.QuickAckGenerator = buildDeliveryGenerator(ctx, deps, agentLoop, msg.TenantID, behavior.QuickAck.Provider, behavior.QuickAck.Model)
|
||||
}
|
||||
}
|
||||
if behavior.Enabled && behavior.IntermediateReplies.Enabled {
|
||||
switch behavior.IntermediateReplies.Mode {
|
||||
case channels.IntermediateModeSidecar, channels.QuickAckModeLLMGenerated, "":
|
||||
runtime.ProgressGenerator = buildDeliveryGenerator(ctx, deps, agentLoop, msg.TenantID, behavior.IntermediateReplies.Provider, behavior.IntermediateReplies.Model)
|
||||
}
|
||||
}
|
||||
return runtime
|
||||
}
|
||||
|
||||
func buildDeliveryGenerator(ctx context.Context, deps *ConsumerDeps, agentLoop agent.Agent, tenantID uuid.UUID, providerName, model string) channels.DeliveryMessageGenerator {
|
||||
provider := agentLoop.Provider()
|
||||
resolvedProviderName := agentLoop.ProviderName()
|
||||
if providerName != "" && deps.ProviderReg != nil {
|
||||
if p, err := deps.ProviderReg.Get(ctx, providerName); err == nil {
|
||||
provider = p
|
||||
resolvedProviderName = providerName
|
||||
} else {
|
||||
slog.Warn("channel delivery: configured provider unavailable, falling back to agent provider", "provider", providerName, "error", err)
|
||||
}
|
||||
}
|
||||
if provider == nil {
|
||||
return nil
|
||||
}
|
||||
if model == "" {
|
||||
model = agentLoop.Model()
|
||||
}
|
||||
return channels.ProviderDeliveryMessageGenerator{
|
||||
Provider: provider,
|
||||
ProviderName: resolvedProviderName,
|
||||
Model: model,
|
||||
UsageCaps: deps.UsageCaps,
|
||||
TenantID: tenantID,
|
||||
AgentID: agentLoop.UUID(),
|
||||
}
|
||||
}
|
||||
|
||||
// isSafeBitrixEntityToken validates a webhook-sourced Bitrix entity token before
|
||||
// it is interpolated into the agent system prompt. Rejects empty, oversized, or
|
||||
// control-character payloads to prevent prompt-injection from a crafted portal
|
||||
|
||||
@@ -147,7 +147,7 @@ func (d *gatewayDeps) runLifecycle(
|
||||
d.channelMgr.SetContactCollector(contactCollector)
|
||||
}
|
||||
|
||||
go consumeInboundMessages(ctx, d.msgBus, d.agentRouter, d.cfg, deps.sched, d.channelMgr, deps.consumerTeamStore, deps.quotaChecker, d.pgStores.Sessions, d.pgStores.Agents, contactCollector, deps.postTurn, deps.subagentMgr, d.usageCapSvc)
|
||||
go consumeInboundMessages(ctx, d.msgBus, d.agentRouter, d.cfg, deps.sched, d.channelMgr, deps.consumerTeamStore, deps.quotaChecker, d.pgStores.Sessions, d.pgStores.Agents, contactCollector, deps.postTurn, deps.subagentMgr, d.usageCapSvc, d.providerRegistry)
|
||||
|
||||
// Webhook callback worker — delivers async webhook_calls rows to receiver callback_url.
|
||||
// Runs in both editions: Standard (PG, concurrency=4) and Lite (SQLite, concurrency=1).
|
||||
|
||||
Reference in new issue
Block a user