From c176d1faab88591da65e439bc9ac071ac1f2adb0 Mon Sep 17 00:00:00 2001 From: Yasuhiro Matsumoto Date: Mon, 16 Feb 2026 00:51:44 +0900 Subject: [PATCH] Resolve data races in channels, logger, spawn, and subagent --- pkg/channels/base.go | 8 ++++---- pkg/channels/telegram.go | 10 +++++----- pkg/logger/logger.go | 15 ++++++++++----- pkg/tools/spawn.go | 14 +++++++++++++- pkg/tools/subagent.go | 13 +++++++++---- 5 files changed, 41 insertions(+), 19 deletions(-) diff --git a/pkg/channels/base.go b/pkg/channels/base.go index 8d2d9a65b..43dc29cf8 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync/atomic" "github.com/sipeed/picoclaw/pkg/bus" ) @@ -20,7 +21,7 @@ type Channel interface { type BaseChannel struct { config interface{} bus *bus.MessageBus - running bool + running atomic.Bool name string allowList []string } @@ -31,7 +32,6 @@ func NewBaseChannel(name string, config interface{}, bus *bus.MessageBus, allowL bus: bus, name: name, allowList: allowList, - running: false, } } @@ -40,7 +40,7 @@ func (c *BaseChannel) Name() string { } func (c *BaseChannel) IsRunning() bool { - return c.running + return c.running.Load() } func (c *BaseChannel) IsAllowed(senderID string) bool { @@ -104,5 +104,5 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st } func (c *BaseChannel) setRunning(running bool) { - c.running = running + c.running.Store(running) } diff --git a/pkg/channels/telegram.go b/pkg/channels/telegram.go index 5601d508c..a39180a93 100644 --- a/pkg/channels/telegram.go +++ b/pkg/channels/telegram.go @@ -28,8 +28,8 @@ type TelegramChannel struct { *BaseChannel bot *telego.Bot commands TelegramCommander - config *config.Config - chatIDs map[string]int64 + config config.TelegramConfig + chatIDs sync.Map transcriber *voice.GroqTranscriber placeholders sync.Map // chatID -> messageID stopThinking sync.Map // chatID -> thinkingCancel @@ -72,8 +72,8 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann BaseChannel: base, commands: NewTelegramCommands(bot, cfg), bot: bot, - config: cfg, - chatIDs: make(map[string]int64), + config: telegramCfg, + chatIDs: sync.Map{}, transcriber: nil, placeholders: sync.Map{}, stopThinking: sync.Map{}, @@ -210,7 +210,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes } chatID := message.Chat.ID - c.chatIDs[senderID] = chatID + c.chatIDs.Store(senderID, chatID) content := "" mediaPaths := []string{} diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 22f66829f..cd71b440e 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -97,7 +97,12 @@ func DisableFileLogging() { } func logMessage(level LogLevel, component string, message string, fields map[string]interface{}) { - if level < currentLevel { + mu.RLock() + lvl := currentLevel + file := logger.file + mu.RUnlock() + + if level < lvl { return } @@ -109,17 +114,17 @@ func logMessage(level LogLevel, component string, message string, fields map[str Fields: fields, } - if pc, file, line, ok := runtime.Caller(2); ok { + if pc, f, line, ok := runtime.Caller(2); ok { fn := runtime.FuncForPC(pc) if fn != nil { - entry.Caller = fmt.Sprintf("%s:%d (%s)", file, line, fn.Name()) + entry.Caller = fmt.Sprintf("%s:%d (%s)", f, line, fn.Name()) } } - if logger.file != nil { + if file != nil { jsonData, err := json.Marshal(entry) if err == nil { - logger.file.WriteString(string(jsonData) + "\n") + file.WriteString(string(jsonData) + "\n") } } diff --git a/pkg/tools/spawn.go b/pkg/tools/spawn.go index 42dd36a33..26cf69c9a 100644 --- a/pkg/tools/spawn.go +++ b/pkg/tools/spawn.go @@ -3,9 +3,11 @@ package tools import ( "context" "fmt" + "sync" ) type SpawnTool struct { + mu sync.Mutex manager *SubagentManager originChannel string originChatID string @@ -22,7 +24,9 @@ func NewSpawnTool(manager *SubagentManager) *SpawnTool { // SetCallback implements AsyncTool interface for async completion notification func (t *SpawnTool) SetCallback(cb AsyncCallback) { + t.mu.Lock() t.callback = cb + t.mu.Unlock() } func (t *SpawnTool) Name() string { @@ -51,8 +55,10 @@ func (t *SpawnTool) Parameters() map[string]interface{} { } func (t *SpawnTool) SetContext(channel, chatID string) { + t.mu.Lock() t.originChannel = channel t.originChatID = chatID + t.mu.Unlock() } func (t *SpawnTool) Execute(ctx context.Context, args map[string]interface{}) *ToolResult { @@ -67,8 +73,14 @@ func (t *SpawnTool) Execute(ctx context.Context, args map[string]interface{}) *T return ErrorResult("Subagent manager not configured") } + t.mu.Lock() + originChannel := t.originChannel + originChatID := t.originChatID + callback := t.callback + t.mu.Unlock() + // Pass callback to manager for async completion notification - result, err := t.manager.Spawn(ctx, task, label, t.originChannel, t.originChatID, t.callback) + result, err := t.manager.Spawn(ctx, task, label, originChannel, originChatID, callback) if err != nil { return ErrorResult(fmt.Sprintf("failed to spawn subagent: %v", err)) } diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index efa1d33aa..a44b7b744 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -89,9 +89,6 @@ func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel } func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, callback AsyncCallback) { - task.Status = "running" - task.Created = time.Now().UnixMilli() - // Build system prompt for subagent systemPrompt := `You are a subagent. Complete the given task independently and report the result. You have access to tools - use them as needed to complete your task. @@ -209,6 +206,7 @@ func (sm *SubagentManager) ListTasks() []*SubagentTask { // Unlike SpawnTool which runs tasks asynchronously, SubagentTool waits for completion // and returns the result directly in the ToolResult. type SubagentTool struct { + mu sync.Mutex manager *SubagentManager originChannel string originChatID string @@ -248,11 +246,18 @@ func (t *SubagentTool) Parameters() map[string]interface{} { } func (t *SubagentTool) SetContext(channel, chatID string) { + t.mu.Lock() + defer t.mu.Unlock() t.originChannel = channel t.originChatID = chatID } func (t *SubagentTool) Execute(ctx context.Context, args map[string]interface{}) *ToolResult { + t.mu.Lock() + originChannel := t.originChannel + originChatID := t.originChatID + t.mu.Unlock() + task, ok := args["task"].(string) if !ok { return ErrorResult("task is required").WithError(fmt.Errorf("task parameter is required")) @@ -292,7 +297,7 @@ func (t *SubagentTool) Execute(ctx context.Context, args map[string]interface{}) "max_tokens": 4096, "temperature": 0.7, }, - }, messages, t.originChannel, t.originChatID) + }, messages, originChannel, originChatID) if err != nil { return ErrorResult(fmt.Sprintf("Subagent execution failed: %v", err)).WithError(err)