opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Resolve data races in channels, logger, spawn, and subagent

Browse source
This commit is contained in:
Yasuhiro Matsumoto 2026-02-16 00:51:44 +09:00
parent 8d757fbb6f
commit c176d1faab
No known key found for this signature in database
GPG key ID: F2EA90DF2C146D1E
5 changed files with 41 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
pkg/channels/base.go
View file

@ -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)
}

10
pkg/channels/telegram.go
View file

@ -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{}

15
pkg/logger/logger.go
View file

@ -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")
}
}

14
pkg/tools/spawn.go
View file

@ -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))
}

13
pkg/tools/subagent.go
View file

@ -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)