diff --git a/cmd/gateway.go b/cmd/gateway.go index 9d24ec2b..ecd541bc 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -22,6 +22,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/cron" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/gateway/methods" + mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" "github.com/nextlevelbuilder/goclaw/internal/pairing" "github.com/nextlevelbuilder/goclaw/internal/permissions" "github.com/nextlevelbuilder/goclaw/internal/providers" @@ -32,7 +33,6 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/internal/store/file" "github.com/nextlevelbuilder/goclaw/internal/store/pg" - mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" "github.com/nextlevelbuilder/goclaw/internal/tools" "github.com/nextlevelbuilder/goclaw/internal/tracing" "github.com/nextlevelbuilder/goclaw/pkg/browser" @@ -229,20 +229,24 @@ func runGateway() { if len(items) > 1 { label = fmt.Sprintf("%d tasks", len(items)) } + batchMeta := map[string]string{ + "origin_channel": meta.OriginChannel, + "origin_peer_kind": meta.OriginPeerKind, + "parent_agent": meta.ParentAgent, + "subagent_label": label, + "origin_trace_id": meta.OriginTraceID, + "origin_root_span_id": meta.OriginRootSpanID, + } + if meta.OriginLocalKey != "" { + batchMeta["origin_local_key"] = meta.OriginLocalKey + } msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: senderID, ChatID: meta.OriginChatID, Content: content, UserID: meta.OriginUserID, - Metadata: map[string]string{ - "origin_channel": meta.OriginChannel, - "origin_peer_kind": meta.OriginPeerKind, - "parent_agent": meta.ParentAgent, - "subagent_label": label, - "origin_trace_id": meta.OriginTraceID, - "origin_root_span_id": meta.OriginRootSpanID, - }, + Metadata: batchMeta, }) }, func(parentID string) int { @@ -737,19 +741,15 @@ func runGateway() { } } - // Register create_forum_topic tool (lazy bot resolution via channel manager). - toolsReg.Register(tools.NewCreateForumTopicTool(func() tools.ForumTopicCreator { - for _, name := range channelMgr.GetEnabledChannels() { - ch, ok := channelMgr.GetChannel(name) - if !ok { - continue - } - if fc, ok := ch.(tools.ForumTopicCreator); ok { - return fc - } - } - return nil - })) + // TODO: create_forum_topic tool — disabled for now, re-enable when needed. + // toolsReg.Register(tools.NewCreateForumTopicTool(func() tools.ForumTopicCreator { + // for _, name := range channelMgr.GetEnabledChannels() { + // ch, ok := channelMgr.GetChannel(name) + // if !ok { continue } + // if fc, ok := ch.(tools.ForumTopicCreator); ok { return fc } + // } + // return nil + // })) // Register channels RPC methods (after channelMgr is initialized with all channels) methods.NewChannelsMethods(channelMgr).Register(server.Router()) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 27d73be2..40fb3bc3 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -193,6 +193,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Channel: msg.Channel, ChatID: msg.ChatID, PeerKind: peerKind, + LocalKey: msg.Metadata["local_key"], UserID: userID, SenderID: msg.SenderID, RunID: runID, @@ -325,6 +326,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "subagent:") { origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] + origLocalKey := msg.Metadata["origin_local_key"] parentAgent := msg.Metadata["parent_agent"] if parentAgent == "" { parentAgent = "default" @@ -341,6 +343,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Use SAME session as user's original chat so agent has context. sessionKey := sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + // Override session key for forum topics / DM threads (same logic as main lane). + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + slog.Info("subagent announce → scheduler (subagent lane)", "subagent", msg.SenderID, "label", msg.Metadata["subagent_label"], @@ -362,6 +367,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID) } + // Build outbound metadata for topic/thread routing. + outMeta := buildAnnounceOutMeta(origLocalKey) + // Schedule through subagent lane outCh := sched.Schedule(ctx, scheduler.LaneSubagent, agent.RunRequest{ SessionKey: sessionKey, @@ -370,6 +378,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, + LocalKey: origLocalKey, UserID: announceUserID, RunID: fmt.Sprintf("announce-%s", msg.SenderID), Stream: false, @@ -378,14 +387,15 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) // Handle result asynchronously to not block the consumer loop - go func(origCh, chatID, senderID, label string) { + go func(origCh, chatID, senderID, label string, meta map[string]string) { outcome := <-outCh if outcome.Err != nil { slog.Error("subagent announce: agent run failed", "error", outcome.Err) msgBus.PublishOutbound(bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: formatAgentError(outcome.Err), + Channel: origCh, + ChatID: chatID, + Content: formatAgentError(outcome.Err), + Metadata: meta, }) return } @@ -406,9 +416,10 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents announceContent = "" // suppress NO_REPLY text but still send media } outMsg := bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: announceContent, + Channel: origCh, + ChatID: chatID, + Content: announceContent, + Metadata: meta, } for _, mr := range outcome.Result.Media { outMsg.Media = append(outMsg.Media, bus.MediaAttachment{ @@ -417,7 +428,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"]) + }(origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"], outMeta) continue } @@ -426,6 +437,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "delegate:") { origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] + origLocalKey := msg.Metadata["origin_local_key"] parentAgent := msg.Metadata["parent_agent"] if parentAgent == "" { parentAgent = "default" @@ -441,6 +453,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents sessionKey := sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + // Override session key for forum topics / DM threads. + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + slog.Info("delegate announce → scheduler (delegate lane)", "delegation", msg.SenderID, "target", msg.Metadata["target_agent"], @@ -461,6 +476,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents parentRootSpanID, _ = uuid.Parse(sid) } + // Build outbound metadata for topic/thread routing. + outMeta := buildAnnounceOutMeta(origLocalKey) + outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, @@ -468,6 +486,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, + LocalKey: origLocalKey, UserID: announceUserID, RunID: fmt.Sprintf("delegate-announce-%s", msg.Metadata["delegation_id"]), Stream: false, @@ -475,14 +494,15 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents ParentRootSpanID: parentRootSpanID, }) - go func(origCh, chatID, senderID string) { + go func(origCh, chatID, senderID string, meta map[string]string) { outcome := <-outCh if outcome.Err != nil { slog.Error("delegate announce: agent run failed", "error", outcome.Err) msgBus.PublishOutbound(bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: formatAgentError(outcome.Err), + Channel: origCh, + ChatID: chatID, + Content: formatAgentError(outcome.Err), + Metadata: meta, }) return } @@ -496,9 +516,10 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents announceContent = "" // suppress NO_REPLY text but still send media } outMsg := bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: announceContent, + Channel: origCh, + ChatID: chatID, + Content: announceContent, + Metadata: meta, } for _, mr := range outcome.Result.Media { outMsg.Media = append(outMsg.Media, bus.MediaAttachment{ @@ -507,7 +528,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(origChannel, msg.ChatID, msg.SenderID) + }(origChannel, msg.ChatID, msg.SenderID, outMeta) continue } @@ -516,6 +537,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "handoff:") { origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] + origLocalKey := msg.Metadata["origin_local_key"] targetAgent := msg.AgentID if targetAgent == "" { targetAgent = cfg.ResolveDefaultAgentID() @@ -530,6 +552,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents } sessionKey := sessions.BuildScopedSessionKey(targetAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, targetAgent, origChannel, msg.ChatID, origPeerKind) slog.Info("handoff announce → scheduler (delegate lane)", "handoff", msg.SenderID, @@ -542,18 +565,21 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID) } + outMeta := buildAnnounceOutMeta(origLocalKey) + outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, + LocalKey: origLocalKey, UserID: announceUserID, RunID: fmt.Sprintf("handoff-%s", msg.Metadata["handoff_id"]), Stream: false, }) - go func(origCh, chatID string) { + go func(origCh, chatID string, meta map[string]string) { outcome := <-outCh if outcome.Err != nil { slog.Error("handoff announce: agent run failed", "error", outcome.Err) @@ -563,11 +589,12 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents return } msgBus.PublishOutbound(bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: outcome.Result.Content, + Channel: origCh, + ChatID: chatID, + Content: outcome.Result.Content, + Metadata: meta, }) - }(origChannel, msg.ChatID) + }(origChannel, msg.ChatID, outMeta) continue } @@ -576,6 +603,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "teammate:") { origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] + origLocalKey := msg.Metadata["origin_local_key"] targetAgent := msg.AgentID // team_message sets AgentID to the target agent key if targetAgent == "" { targetAgent = cfg.ResolveDefaultAgentID() @@ -590,6 +618,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents } sessionKey := sessions.BuildScopedSessionKey(targetAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, targetAgent, origChannel, msg.ChatID, origPeerKind) slog.Info("teammate message → scheduler (delegate lane)", "from", msg.SenderID, @@ -602,18 +631,21 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID) } + outMeta := buildAnnounceOutMeta(origLocalKey) + outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, + LocalKey: origLocalKey, UserID: announceUserID, RunID: fmt.Sprintf("teammate-%s-%s", msg.Metadata["from_agent"], msg.Metadata["to_agent"]), Stream: false, }) - go func(origCh, chatID, senderID string) { + go func(origCh, chatID, senderID string, meta map[string]string) { outcome := <-outCh if outcome.Err != nil { slog.Error("teammate message: agent run failed", "error", outcome.Err) @@ -626,11 +658,12 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Deliver response to origin channel (same as delegate/subagent announce). // This allows the lead to respond to users after receiving teammate updates. msgBus.PublishOutbound(bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: outcome.Result.Content, + Channel: origCh, + ChatID: chatID, + Content: outcome.Result.Content, + Metadata: meta, }) - }(origChannel, msg.ChatID, msg.SenderID) + }(origChannel, msg.ChatID, msg.SenderID, outMeta) continue } @@ -723,3 +756,41 @@ func resolveAgentRoute(cfg *config.Config, channel, chatID, peerKind string) str return cfg.ResolveDefaultAgentID() } + +// overrideSessionKeyFromLocalKey extracts topic/thread ID from the composite +// local_key and returns the correct session key for forum topics or DM threads. +// If localKey is empty or has no suffix, the original sessionKey is returned unchanged. +func overrideSessionKeyFromLocalKey(sessionKey, localKey, agentID, channel, chatID, peerKind string) string { + if localKey == "" { + return sessionKey + } + if idx := strings.Index(localKey, ":topic:"); idx > 0 && peerKind == string(sessions.PeerGroup) { + var topicID int + fmt.Sscanf(localKey[idx+7:], "%d", &topicID) + if topicID > 0 { + return sessions.BuildGroupTopicSessionKey(agentID, channel, chatID, topicID) + } + } else if idx := strings.Index(localKey, ":thread:"); idx > 0 && peerKind == string(sessions.PeerDirect) { + var threadID int + fmt.Sscanf(localKey[idx+8:], "%d", &threadID) + if threadID > 0 { + return sessions.BuildDMThreadSessionKey(agentID, channel, chatID, threadID) + } + } + return sessionKey +} + +// buildAnnounceOutMeta builds outbound metadata for announce messages so that +// Send() can route replies to the correct forum topic or DM thread. +func buildAnnounceOutMeta(localKey string) map[string]string { + if localKey == "" { + return nil + } + meta := map[string]string{"local_key": localKey} + if idx := strings.Index(localKey, ":topic:"); idx > 0 { + meta["message_thread_id"] = localKey[idx+7:] + } else if idx := strings.Index(localKey, ":thread:"); idx > 0 { + meta["message_thread_id"] = localKey[idx+8:] + } + return meta +} diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 4fe3c801..f5f65ee8 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -247,6 +247,7 @@ type RunRequest struct { ExtraSystemPrompt string // optional: injected into system prompt (skills, subagent context, etc.) SkillFilter []string // per-request skill override: nil=use agent default, []=no skills, ["x","y"]=whitelist HistoryLimit int // max user turns to keep in context (0=unlimited, from channel config) + LocalKey string // composite key with topic/thread suffix for routing (e.g. "-100123:topic:42") ParentTraceID uuid.UUID // if set, reuse parent trace instead of creating new (announce runs) ParentRootSpanID uuid.UUID // if set, nest announce agent span under this parent span TraceName string // override trace name (default: "chat ") @@ -331,6 +332,12 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { } } + // Inject local key into tool context so delegation/subagent tools can + // propagate topic/thread routing info back through announce messages. + if req.LocalKey != "" { + ctx = tools.WithToolLocalKey(ctx, req.LocalKey) + } + runStart := time.Now().UTC() result, err := l.runLoop(ctx, req) diff --git a/internal/channels/manager.go b/internal/channels/manager.go index fa5ea520..b8e08e09 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -5,6 +5,8 @@ import ( "fmt" "log/slog" "os" + "regexp" + "strings" "sync" "github.com/nextlevelbuilder/goclaw/internal/bus" @@ -132,6 +134,21 @@ func (m *Manager) dispatchOutbound(ctx context.Context) { "channel", msg.Channel, "error", err, ) + // Try to send a text-only error notification back to the chat. + // Only for media failures — text-only failures likely mean the chat + // is inaccessible (kicked, blocked, etc.) so retrying won't help. + if len(msg.Media) > 0 { + notifyMsg := bus.OutboundMessage{ + Channel: msg.Channel, + ChatID: msg.ChatID, + Content: formatChannelSendError(err), + Metadata: sendErrorMeta(msg.Metadata), + } + if err2 := channel.Send(ctx, notifyMsg); err2 != nil { + slog.Warn("failed to send error notification", + "channel", msg.Channel, "error", err2) + } + } } // Clean up temporary media files after successful (or failed) send. @@ -376,3 +393,61 @@ func extractPayloadString(payload interface{}, key string) string { } return "" } + +// --- Send error notification helpers --- + +// telegramAPIDescRe extracts the human-readable description from Telegram Bot API errors. +// Example: `telego: sendPhoto: api: 400 "Bad Request: not enough rights to send photos to the chat"` +// +// → "not enough rights to send photos to the chat" +var telegramAPIDescRe = regexp.MustCompile(`"Bad Request:\s*(.+?)"`) + +// formatChannelSendError converts a channel.Send error into a user-friendly message. +// Never exposes raw library/HTTP details. +func formatChannelSendError(err error) string { + raw := err.Error() + lower := strings.ToLower(raw) + + // Telegram "Bad Request: " — extract description + if m := telegramAPIDescRe.FindStringSubmatch(raw); len(m) == 2 { + return fmt.Sprintf("⚠️ Send failed: %s", m[1]) + } + + // Common Telegram API errors (non-Bad Request) + switch { + case strings.Contains(lower, "not enough rights"): + return "⚠️ Send failed: bot doesn't have permission to send this type of message." + case strings.Contains(lower, "chat not found"): + return "⚠️ Send failed: chat not found." + case strings.Contains(lower, "bot was blocked"): + return "⚠️ Send failed: bot was blocked by the user." + case strings.Contains(lower, "user is deactivated"): + return "⚠️ Send failed: user account is deactivated." + case strings.Contains(lower, "too many requests") || strings.Contains(lower, "flood"): + return "⚠️ Send failed: rate limited by Telegram. Please try again later." + case strings.Contains(lower, "file is too big") || strings.Contains(lower, "wrong file"): + return "⚠️ Send failed: file is too large or invalid for Telegram." + } + + // Generic fallback — don't expose internals + return "⚠️ Failed to deliver message. Check bot logs for details." +} + +// sendErrorMeta copies only the routing fields from outbound metadata. +// Strips reply_to_message_id, placeholder_key, audio_as_voice, etc. +// that could cause unintended side effects on the error notification. +func sendErrorMeta(orig map[string]string) map[string]string { + if orig == nil { + return nil + } + meta := make(map[string]string) + for _, k := range []string{"local_key", "message_thread_id"} { + if v := orig[k]; v != "" { + meta[k] = v + } + } + if len(meta) == 0 { + return nil + } + return meta +} diff --git a/internal/tools/announce_queue.go b/internal/tools/announce_queue.go index 22766ee2..cc455b09 100644 --- a/internal/tools/announce_queue.go +++ b/internal/tools/announce_queue.go @@ -22,6 +22,7 @@ type AnnounceMetadata struct { OriginChannel string OriginChatID string OriginPeerKind string + OriginLocalKey string // composite key with topic/thread suffix for routing OriginUserID string ParentAgent string OriginTraceID string // parent trace UUID for announce linking diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index 8a9b8f88..a45fd723 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -17,6 +17,7 @@ const ( ctxChannel toolContextKey = "tool_channel" ctxChatID toolContextKey = "tool_chat_id" ctxPeerKind toolContextKey = "tool_peer_kind" + ctxLocalKey toolContextKey = "tool_local_key" // composite key with topic/thread suffix for routing ctxSandboxKey toolContextKey = "tool_sandbox_key" ctxAsyncCB toolContextKey = "tool_async_cb" ctxWorkspace toolContextKey = "tool_workspace" @@ -50,6 +51,17 @@ func ToolPeerKindFromCtx(ctx context.Context) string { return v } +// WithToolLocalKey injects the composite local key (e.g. "-100123:topic:42") into context. +// Used by delegation/subagent to preserve topic routing info for announce-back. +func WithToolLocalKey(ctx context.Context, localKey string) context.Context { + return context.WithValue(ctx, ctxLocalKey, localKey) +} + +func ToolLocalKeyFromCtx(ctx context.Context) string { + v, _ := ctx.Value(ctxLocalKey).(string) + return v +} + func WithToolSandboxKey(ctx context.Context, key string) context.Context { return context.WithValue(ctx, ctxSandboxKey, key) } diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index 532dc689..d82f7d63 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -36,6 +36,7 @@ type DelegationTask struct { OriginChannel string `json:"-"` OriginChatID string `json:"-"` OriginPeerKind string `json:"-"` + OriginLocalKey string `json:"-"` // composite key with topic/thread suffix for routing // Trace context for announce linking (same pattern as SubagentTask) OriginTraceID uuid.UUID `json:"-"` @@ -296,22 +297,26 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) }) } + announceMeta := map[string]string{ + "origin_channel": task.OriginChannel, + "origin_peer_kind": task.OriginPeerKind, + "parent_agent": task.SourceAgentKey, + "delegation_id": task.ID, + "target_agent": task.TargetAgentKey, + "origin_trace_id": task.OriginTraceID.String(), + "origin_root_span_id": task.OriginRootSpanID.String(), + } + if task.OriginLocalKey != "" { + announceMeta["origin_local_key"] = task.OriginLocalKey + } announceMsg := bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("delegate:%s", task.ID), ChatID: task.OriginChatID, Content: formatDelegateAnnounce(task, artifacts, runErr, elapsed), UserID: task.UserID, - Metadata: map[string]string{ - "origin_channel": task.OriginChannel, - "origin_peer_kind": task.OriginPeerKind, - "parent_agent": task.SourceAgentKey, - "delegation_id": task.ID, - "target_agent": task.TargetAgentKey, - "origin_trace_id": task.OriginTraceID.String(), - "origin_root_span_id": task.OriginRootSpanID.String(), - }, - Media: artifacts.Media, + Metadata: announceMeta, + Media: artifacts.Media, } dm.msgBus.PublishInbound(announceMsg) } @@ -436,6 +441,7 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO channel := ToolChannelFromCtx(ctx) chatID := ToolChatIDFromCtx(ctx) peerKind := ToolPeerKindFromCtx(ctx) + localKey := ToolLocalKeyFromCtx(ctx) delegationID := uuid.NewString()[:12] task := &DelegationTask{ @@ -454,6 +460,7 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO OriginChannel: channel, OriginChatID: chatID, OriginPeerKind: peerKind, + OriginLocalKey: localKey, OriginTraceID: tracing.TraceIDFromContext(ctx), OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), TeamTaskID: opts.TeamTaskID, diff --git a/internal/tools/handoff_tool.go b/internal/tools/handoff_tool.go index fe849d7d..911ed74f 100644 --- a/internal/tools/handoff_tool.go +++ b/internal/tools/handoff_tool.go @@ -183,6 +183,15 @@ func (t *HandoffTool) executeTransfer(ctx context.Context, args map[string]inter } content += "\n\nPlease greet the user and continue the conversation." + handoffMeta := map[string]string{ + "origin_channel": channel, + "origin_peer_kind": peerKind, + "handoff_id": handoffID, + "from_agent": sourceAgent.AgentKey, + } + if localKey := ToolLocalKeyFromCtx(ctx); localKey != "" { + handoffMeta["origin_local_key"] = localKey + } t.msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("handoff:%s", handoffID), @@ -190,12 +199,7 @@ func (t *HandoffTool) executeTransfer(ctx context.Context, args map[string]inter Content: content, AgentID: targetAgentKey, UserID: userID, - Metadata: map[string]string{ - "origin_channel": channel, - "origin_peer_kind": peerKind, - "handoff_id": handoffID, - "from_agent": sourceAgent.AgentKey, - }, + Metadata: handoffMeta, }) } diff --git a/internal/tools/policy.go b/internal/tools/policy.go index f12c86a6..cf0d0b41 100644 --- a/internal/tools/policy.go +++ b/internal/tools/policy.go @@ -17,7 +17,7 @@ var toolGroups = map[string][]string{ "sessions": {"sessions_list", "sessions_history", "sessions_send", "sessions_spawn", "session_status"}, "ui": {"browser", "canvas"}, "automation": {"cron", "gateway"}, - "messaging": {"message"}, + "messaging": {"message", "create_forum_topic"}, "nodes": {"nodes"}, // Composite group: all goclaw native tools (excludes provider plugins). // Matching TS group:goclaw. @@ -26,6 +26,7 @@ var toolGroups = map[string][]string{ "agents_list", "sessions_list", "sessions_history", "sessions_send", "sessions_spawn", "session_status", "memory_search", "memory_get", "web_search", "web_fetch", "read_image", "create_image", + "create_forum_topic", }, } diff --git a/internal/tools/subagent.go b/internal/tools/subagent.go index db570735..cdcd2549 100644 --- a/internal/tools/subagent.go +++ b/internal/tools/subagent.go @@ -54,6 +54,7 @@ type SubagentTask struct { OriginChannel string `json:"originChannel,omitempty"` OriginChatID string `json:"originChatId,omitempty"` OriginPeerKind string `json:"originPeerKind,omitempty"` // "direct" or "group" (for session key building) + OriginLocalKey string `json:"originLocalKey,omitempty"` // composite key with topic/thread suffix for routing OriginUserID string `json:"originUserId,omitempty"` // parent's userID for per-user scoping propagation CreatedAt int64 `json:"createdAt"` CompletedAt int64 `json:"completedAt,omitempty"` @@ -192,6 +193,7 @@ func (sm *SubagentManager) Spawn( OriginChannel: channel, OriginChatID: chatID, OriginPeerKind: peerKind, + OriginLocalKey: ToolLocalKeyFromCtx(ctx), OriginUserID: store.UserIDFromContext(ctx), OriginTraceID: tracing.TraceIDFromContext(ctx), OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), @@ -241,6 +243,7 @@ func (sm *SubagentManager) RunSync( Depth: depth + 1, OriginChannel: channel, OriginChatID: chatID, + OriginLocalKey: ToolLocalKeyFromCtx(ctx), OriginUserID: store.UserIDFromContext(ctx), OriginTraceID: tracing.TraceIDFromContext(ctx), OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), diff --git a/internal/tools/subagent_exec.go b/internal/tools/subagent_exec.go index 8e84f18e..5911ebaf 100644 --- a/internal/tools/subagent_exec.go +++ b/internal/tools/subagent_exec.go @@ -35,6 +35,7 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call OriginChannel: task.OriginChannel, OriginChatID: task.OriginChatID, OriginPeerKind: task.OriginPeerKind, + OriginLocalKey: task.OriginLocalKey, OriginUserID: task.OriginUserID, ParentAgent: task.ParentID, OriginTraceID: task.OriginTraceID.String(), @@ -50,21 +51,25 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call remainingActive := sm.CountRunningForParent(task.ParentID) announceContent := FormatBatchedAnnounce([]AnnounceQueueItem{item}, remainingActive) + announceMeta := map[string]string{ + "origin_channel": task.OriginChannel, + "origin_peer_kind": task.OriginPeerKind, + "parent_agent": task.ParentID, + "subagent_id": task.ID, + "subagent_label": task.Label, + "origin_trace_id": task.OriginTraceID.String(), + "origin_root_span_id": task.OriginRootSpanID.String(), + } + if task.OriginLocalKey != "" { + announceMeta["origin_local_key"] = task.OriginLocalKey + } sm.msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("subagent:%s", task.ID), ChatID: task.OriginChatID, Content: announceContent, UserID: task.OriginUserID, - Metadata: map[string]string{ - "origin_channel": task.OriginChannel, - "origin_peer_kind": task.OriginPeerKind, - "parent_agent": task.ParentID, - "subagent_id": task.ID, - "subagent_label": task.Label, - "origin_trace_id": task.OriginTraceID.String(), - "origin_root_span_id": task.OriginRootSpanID.String(), - }, + Metadata: announceMeta, }) } } diff --git a/internal/tools/team_message_tool.go b/internal/tools/team_message_tool.go index 62409137..4ebd2b95 100644 --- a/internal/tools/team_message_tool.go +++ b/internal/tools/team_message_tool.go @@ -212,6 +212,15 @@ func (t *TeamMessageTool) publishTeammateMessage(fromKey, toKey, text string, ct originChannel := ToolChannelFromCtx(ctx) originPeerKind := ToolPeerKindFromCtx(ctx) + teamMeta := map[string]string{ + "origin_channel": originChannel, + "origin_peer_kind": originPeerKind, + "from_agent": fromKey, + "to_agent": toKey, + } + if localKey := ToolLocalKeyFromCtx(ctx); localKey != "" { + teamMeta["origin_local_key"] = localKey + } t.manager.msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("teammate:%s", fromKey), @@ -219,11 +228,6 @@ func (t *TeamMessageTool) publishTeammateMessage(fromKey, toKey, text string, ct Content: fmt.Sprintf("[Team message from %s]: %s", fromKey, text), UserID: userID, AgentID: toKey, - Metadata: map[string]string{ - "origin_channel": originChannel, - "origin_peer_kind": originPeerKind, - "from_agent": fromKey, - "to_agent": toKey, - }, + Metadata: teamMeta, }) }