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

refactor: Remove steering arch and add message types

Browse source
This commit is contained in:
leozeli 2026-03-04 09:49:22 +08:00
parent 05034ad7c8
commit d9e900cf29
9 changed files with 533 additions and 458 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
go.mod
View file

@ -8,6 +8,7 @@ require (
github.com/bwmarrin/discordgo v0.29.0
github.com/caarlos0/env/v11 v11.3.1
github.com/chzyer/readline v1.5.1
github.com/gdamore/tcell/v2 v2.13.8
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.3
github.com/larksuite/oapi-sdk-go/v3 v3.5.3
@ -16,6 +17,7 @@ require (
github.com/mymmrac/telego v1.6.0
github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1
github.com/openai/openai-go/v3 v3.22.0
github.com/rivo/tview v0.42.0
github.com/slack-go/slack v0.17.3
github.com/spf13/cobra v1.10.2
github.com/stretchr/testify v1.11.1
@ -35,7 +37,6 @@ require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/elliotchance/orderedmap/v3 v3.1.0 // indirect
github.com/gdamore/encoding v1.0.1 // indirect
github.com/gdamore/tcell/v2 v2.13.8 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/lucasb-eyer/go-colorful v1.3.0 // indirect
github.com/mattn/go-colorable v0.1.14 // indirect
@ -44,7 +45,6 @@ require (
github.com/petermattis/goid v0.0.0-20260113132338-7c7de50cc741 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/rivo/tview v0.42.0 // indirect
github.com/rivo/uniseg v0.4.7 // indirect
github.com/rs/zerolog v1.34.0 // indirect
github.com/spf13/pflag v1.0.10 // indirect

19
pkg/agent/context.go
View file

@ -71,6 +71,25 @@ Your workspace is at: %s
- Daily Notes: %s/memory/YYYYMM/YYYYMMDD.md
- Skills: %s/skills/{skill-name}/SKILL.md
## Message Structure (Dual-Layer Model)
You may receive messages with special formatting that provides additional context:
1. **Subagent Results** - Messages from background tasks:
- Format: [Subagent 'task_name' completed after N iterations]
- These contain results from asynchronous operations you delegated
- The iteration count indicates how many tool calls the subagent made
- Use this information to understand what work was done
2. **Tool Progress** - Progress updates from long-running operations:
- Format: [Progress: XX%%] Status text
- These are intermediate updates, not final results
- Continue waiting for the final result
3. **System Messages** - Internal notifications from the system
**Note**: These structured messages help you track background work and understand execution context. When you see subagent results, you can reference the iterations to gauge task complexity.
## Important Rules
1. **ALWAYS use tools** - When you need to perform an action (schedule reminders, send messages, execute commands, etc.), you MUST call the appropriate tool. Do NOT just say you'll do it or pretend to do it.

457
pkg/agent/loop.go
View file

@ -44,14 +44,9 @@ type AgentLoop struct {
channelManager *channels.Manager
mediaStore media.MediaStore
// Legacy interrupt handling (to be deprecated)
// Task management for concurrent task tracking
interruptHandler InterruptHandler // Interrupt handler for dynamic task management
taskManager *TaskManager // Task manager for concurrent task tracking (Phase 2)
// New steering architecture (nanobot-inspired)
enableSteering bool // Opt-in flag for steering feature
interruptCheckers map[string]*InterruptionChecker // Per-session interrupt queues
checkersMu sync.RWMutex // Protects interruptCheckers map
taskManager *TaskManager // Task manager for concurrent task tracking
}
// processOptions configures how a message is processed
@ -96,24 +91,17 @@ func NewAgentLoop(
state: stateManager,
summarizing: sync.Map{},
fallback: fallbackChain,
// New steering architecture (nanobot-inspired)
enableSteering: cfg.Agents.Defaults.EnableSteering,
interruptCheckers: make(map[string]*InterruptionChecker),
}
// Legacy components (DEPRECATED - only initialized when new steering is disabled)
if !cfg.Agents.Defaults.EnableSteering {
// Legacy interrupt handler (Phase 1.5)
// Initialize interrupt handler
al.interruptHandler = NewBusInterruptHandler(msgBus, DefaultInterruptionConfig())
// Legacy TaskManager (Phase 2)
// Initialize TaskManager for concurrent task tracking
maxConcurrent := cfg.Agents.Defaults.MaxConcurrentTasks
if maxConcurrent < 0 {
maxConcurrent = 0 // Ensure no negative values, 0 = unlimited
}
al.taskManager = NewTaskManager(maxConcurrent)
}
return al
}
@ -268,24 +256,13 @@ func (al *AgentLoop) Run(ctx context.Context) error {
}
}
// Legacy: Start task cleanup and steering loop only when new steering is disabled
if !al.enableSteering {
// Phase 2: Start task cleanup goroutine
go al.runTaskCleanup(ctx)
// Phase 2: Start steering loop if enabled
if al.cfg.Agents.Defaults.EnableSteeringLoop {
go al.runSteeringLoop(ctx)
}
}
for al.running.Load() {
select {
case <-ctx.Done():
// Legacy: Wait for running tasks to complete (only if using TaskManager)
if !al.enableSteering && al.taskManager != nil {
al.waitForRunningTasks(5 * time.Second)
}
return nil
default:
msg, ok := al.bus.ConsumeInbound(ctx)
@ -293,39 +270,6 @@ func (al *AgentLoop) Run(ctx context.Context) error {
continue
}
// ===== NEW: Steering Architecture - Check for active session =====
if al.enableSteering {
// Get session key for this message (needs routing resolution)
sessionKey := al.getSessionKeyForMessage(msg)
// Check if this session has an active checker (task is running)
if al.hasActiveChecker(sessionKey) {
// Session is active, signal interruption instead of creating new task
checker := al.getOrCreateChecker(sessionKey)
signaled := checker.Signal(msg)
if signaled {
logger.InfoCF("agent", "Steering: signaled interruption for active session",
map[string]any{
"session_key": sessionKey,
"channel": msg.Channel,
"chat_id": msg.ChatID,
"content_preview": utils.Truncate(msg.Content, 60),
})
continue // Don't process as new message
} else {
// Grace period expired, treat as new message
logger.WarnCF("agent", "Steering: session checker exists but grace period expired, processing as new message",
map[string]any{
"session_key": sessionKey,
"channel": msg.Channel,
"chat_id": msg.ChatID,
})
// Fall through to process as new message
}
}
}
// Phase 2: Process message asynchronously
go func(msg bus.InboundMessage) {
defer func() {
@ -465,31 +409,6 @@ func (al *AgentLoop) waitForRunningTasks(timeout time.Duration) {
}
}
// runSteeringLoop monitors for interrupt signals and cancels tasks (Phase 2 Step 5)
func (al *AgentLoop) runSteeringLoop(ctx context.Context) {
// Get interval from config, default to 500ms
intervalMs := al.cfg.Agents.Defaults.SteeringLoopIntervalMs
if intervalMs <= 0 {
intervalMs = 500
}
ticker := time.NewTicker(time.Duration(intervalMs) * time.Millisecond)
defer ticker.Stop()
logger.InfoCF("agent", "Steering loop started",
map[string]any{"interval_ms": intervalMs})
for {
select {
case <-ctx.Done():
logger.InfoCF("agent", "Steering loop stopped", nil)
return
case <-ticker.C:
al.checkAndHandleInterrupts(ctx)
}
}
}
// checkAndHandleInterrupts checks for interrupt signals and handles them (Phase 2 Step 5)
func (al *AgentLoop) checkAndHandleInterrupts(ctx context.Context) {
if al.interruptHandler == nil {
@ -556,104 +475,6 @@ func (al *AgentLoop) RegisterTool(tool tools.Tool) {
}
}
// ===== Steering Architecture: InterruptionChecker Management =====
// getOrCreateChecker gets or creates an interruption checker for a session.
// Thread-safe with double-checked locking pattern.
func (al *AgentLoop) getOrCreateChecker(sessionKey string) *InterruptionChecker {
// Fast path: read lock
al.checkersMu.RLock()
checker, exists := al.interruptCheckers[sessionKey]
al.checkersMu.RUnlock()
if exists {
return checker
}
// Slow path: write lock
al.checkersMu.Lock()
defer al.checkersMu.Unlock()
// Double-check after acquiring write lock
if checker, exists := al.interruptCheckers[sessionKey]; exists {
return checker
}
// Create new checker
checker = NewInterruptionChecker()
al.interruptCheckers[sessionKey] = checker
logger.DebugCF("agent", "Created interruption checker for session",
map[string]any{"session_key": sessionKey})
return checker
}
// formatInterruptionInjection formats pending interruption messages for injection into conversation.
// This follows nanobot's pattern of providing context to the LLM about the interruption.
func formatInterruptionInjection(pending []bus.InboundMessage) string {
if len(pending) == 0 {
return ""
}
var combined strings.Builder
for i, msg := range pending {
if i > 0 {
combined.WriteString("\n\n---\n\n")
}
combined.WriteString(msg.Content)
}
injection := "[The user just sent a new message while you were working. " +
"Read it and decide: continue current work, switch to the new request, or address both.]\n\n" +
combined.String()
return injection
}
// removeChecker removes a checker when session completes.
// This prevents memory leaks for long-running processes.
func (al *AgentLoop) removeChecker(sessionKey string) {
al.checkersMu.Lock()
defer al.checkersMu.Unlock()
if _, exists := al.interruptCheckers[sessionKey]; exists {
delete(al.interruptCheckers, sessionKey)
logger.DebugCF("agent", "Removed interruption checker for session",
map[string]any{"session_key": sessionKey})
}
}
// hasActiveChecker checks if a session has an active interruption checker.
// This indicates the session is currently processing a message.
func (al *AgentLoop) hasActiveChecker(sessionKey string) bool {
al.checkersMu.RLock()
defer al.checkersMu.RUnlock()
_, exists := al.interruptCheckers[sessionKey]
return exists
}
// getSessionKeyForMessage resolves the session key for a message using routing logic.
// This is needed to check if the session has an active task before creating a new one.
func (al *AgentLoop) getSessionKeyForMessage(msg bus.InboundMessage) string {
// If message already has an agent-scoped session key, use it
if msg.SessionKey != "" && strings.HasPrefix(msg.SessionKey, "agent:") {
return msg.SessionKey
}
// Otherwise, resolve via routing
route := al.registry.ResolveRoute(routing.RouteInput{
Channel: msg.Channel,
AccountID: msg.Metadata["account_id"],
Peer: extractPeer(msg),
ParentPeer: extractParentPeer(msg),
GuildID: msg.Metadata["guild_id"],
TeamID: msg.Metadata["team_id"],
})
return route.SessionKey
}
func (al *AgentLoop) SetChannelManager(cm *channels.Manager) {
al.channelManager = cm
}
@ -781,9 +602,6 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
)
// New steering architecture: Direct processing without task management
if al.enableSteering {
return al.processMessageDirect(ctx, msg)
}
// Legacy: Phase 2 task management
priority := 5 // Default priority
@ -826,59 +644,6 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
return response, nil
}
// processMessageDirect handles message processing for new steering architecture (no task management)
func (al *AgentLoop) processMessageDirect(ctx context.Context, msg bus.InboundMessage) (string, error) {
// Route system messages to processSystemMessage
if msg.Channel == "system" {
return al.processSystemMessage(ctx, msg)
}
// Check for commands
if response, handled := al.handleCommand(ctx, msg); handled {
return response, nil
}
// Route to determine agent and session key
route := al.registry.ResolveRoute(routing.RouteInput{
Channel: msg.Channel,
AccountID: msg.Metadata["account_id"],
Peer: extractPeer(msg),
ParentPeer: extractParentPeer(msg),
GuildID: msg.Metadata["guild_id"],
TeamID: msg.Metadata["team_id"],
})
agent, ok := al.registry.GetAgent(route.AgentID)
if !ok {
agent = al.registry.GetDefaultAgent()
}
if agent == nil {
return "", fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID)
}
// Reset message-tool state for this round
if tool, ok := agent.Tools.Get("message"); ok {
if mt, ok := tool.(tools.ContextualTool); ok {
mt.SetContext(msg.Channel, msg.ChatID)
}
}
// Use routed session key, but honor pre-set agent-scoped keys
sessionKey := route.SessionKey
if msg.SessionKey != "" && strings.HasPrefix(msg.SessionKey, "agent:") {
sessionKey = msg.SessionKey
}
return al.runAgentLoop(ctx, agent, processOptions{
SessionKey: sessionKey,
Channel: msg.Channel,
ChatID: msg.ChatID,
UserMessage: msg.Content,
EnableSummary: true,
SendResponse: false,
})
}
// processMessageWithTask handles the actual message processing logic with task context (LEGACY)
func (al *AgentLoop) processMessageWithTask(ctx context.Context, task *Task, msg bus.InboundMessage) (string, error) {
// Route system messages to processSystemMessage
@ -1013,45 +778,6 @@ func (al *AgentLoop) runAgentLoop(
}
// ===== NEW: Steering Architecture - Setup checker for this session =====
if al.enableSteering {
// Create checker to signal this session is active
checker := al.getOrCreateChecker(opts.SessionKey)
// Cleanup with grace period to handle race conditions
defer func() {
// Set grace period before checking for late arrivals
checker.SetGracePeriod(2 * time.Second)
// Wait briefly for any race-condition messages
time.Sleep(150 * time.Millisecond)
// Check one final time for pending interruptions
finalPending := checker.DrainAll()
if len(finalPending) > 0 {
logger.InfoCF("agent", "Steering: found interruptions during grace period, reprocessing",
map[string]any{
"session_key": opts.SessionKey,
"pending_count": len(finalPending),
"channel": opts.Channel,
"chat_id": opts.ChatID,
})
// Re-trigger processing by publishing as new inbound message
// Combine all pending messages
injectionContent := formatInterruptionInjection(finalPending)
al.bus.PublishInbound(ctx, bus.InboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
SessionKey: opts.SessionKey,
Content: injectionContent,
Metadata: make(map[string]string), // Empty metadata for re-triggered message
})
}
// Now safe to remove checker
al.removeChecker(opts.SessionKey)
}()
}
// 0a. Update interrupt handler context
if busHandler, ok := al.interruptHandler.(*BusInterruptHandler); ok {
@ -1418,48 +1144,6 @@ func (al *AgentLoop) runLLMIteration(
// Check if no tool calls - but first check for pending interruptions
if len(response.ToolCalls) == 0 {
// NEW: Check for pending interruptions before finishing
if al.enableSteering {
checker := al.getOrCreateChecker(opts.SessionKey)
pending := checker.DrainAll()
if len(pending) > 0 {
logger.InfoCF("agent", "Steering: LLM finished but has pending interruptions, injecting",
map[string]any{
"session_key": opts.SessionKey,
"pending_count": len(pending),
"iteration": iteration,
})
// Save the assistant's response first
assistantMsg := providers.Message{
Role: "assistant",
Content: response.Content,
}
messages = append(messages, assistantMsg)
agent.Sessions.AddMessage(opts.SessionKey, "assistant", response.Content)
// Send the response to user
if !constants.IsInternalChannel(opts.Channel) {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
Content: response.Content,
})
}
// Format and inject interruption
injectionContent := formatInterruptionInjection(pending)
injectionMsg := providers.Message{
Role: "user",
Content: injectionContent,
}
messages = append(messages, injectionMsg)
agent.Sessions.AddMessage(opts.SessionKey, "user", injectionContent)
// Continue to handle the interruption
continue
}
}
// No interruptions, finish normally
finalContent = response.Content
@ -1472,49 +1156,6 @@ func (al *AgentLoop) runLLMIteration(
// FINAL SAFETY CHECK: One more check for race-condition interruptions
// This catches messages that arrived while we were processing the final response
if al.enableSteering {
time.Sleep(100 * time.Millisecond) // Brief wait for any in-flight messages
checker := al.getOrCreateChecker(opts.SessionKey)
lastMinutePending := checker.DrainAll()
if len(lastMinutePending) > 0 {
logger.InfoCF("agent", "Steering: caught last-minute interruptions after final response",
map[string]any{
"session_key": opts.SessionKey,
"pending_count": len(lastMinutePending),
"iteration": iteration,
})
// Save assistant's response first
assistantMsg := providers.Message{
Role: "assistant",
Content: response.Content,
}
messages = append(messages, assistantMsg)
agent.Sessions.AddMessage(opts.SessionKey, "assistant", response.Content)
// Send response to user
if !constants.IsInternalChannel(opts.Channel) {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
Content: response.Content,
})
}
// Inject last-minute interruptions
injectionContent := formatInterruptionInjection(lastMinutePending)
injectionMsg := providers.Message{
Role: "user",
Content: injectionContent,
}
messages = append(messages, injectionMsg)
agent.Sessions.AddMessage(opts.SessionKey, "user", injectionContent)
// Continue to handle these interruptions
continue
}
}
break
}
@ -1705,56 +1346,6 @@ func (al *AgentLoop) runLLMIteration(
}
// ===== NEW: Steering Architecture - Check for interruptions after tool execution =====
if al.enableSteering {
checker := al.getOrCreateChecker(opts.SessionKey)
pending := checker.DrainAll()
if len(pending) > 0 {
logger.InfoCF("agent", "Steering: injecting interruption messages",
map[string]any{
"session_key": opts.SessionKey,
"pending_count": len(pending),
"iteration": iteration,
})
// Send progress update to user showing tool results before handling interruption
if !constants.IsInternalChannel(opts.Channel) {
// Build a brief summary of what just completed
completedTools := []string{}
for _, msg := range messages {
if msg.Role == "assistant" && len(msg.ToolCalls) > 0 {
for _, tc := range msg.ToolCalls {
completedTools = append(completedTools, tc.Name)
}
}
}
progressMsg := fmt.Sprintf("⚡ Completed: %s\n📥 Processing new request...",
utils.Truncate(fmt.Sprint(completedTools), 100))
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
Content: progressMsg,
})
}
// Format and inject interruption
injectionContent := formatInterruptionInjection(pending)
injectionMsg := providers.Message{
Role: "user",
Content: injectionContent,
}
messages = append(messages, injectionMsg)
// Save injection to session for context continuity
agent.Sessions.AddMessage(opts.SessionKey, "user", injectionContent)
// Continue to next iteration with injected message
// The LLM will decide how to handle both the original task and the new request
continue
}
}
}
return finalContent, iteration, nil
@ -2129,6 +1720,46 @@ func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage)
return fmt.Sprintf("Unknown list target: %s", args[0]), true
}
case "/clear":
// Clear session history to reset conversation state
route := al.registry.ResolveRoute(routing.RouteInput{
Channel: msg.Channel,
AccountID: msg.Metadata["account_id"],
Peer: extractPeer(msg),
ParentPeer: extractParentPeer(msg),
GuildID: msg.Metadata["guild_id"],
TeamID: msg.Metadata["team_id"],
})
agent, _ := al.registry.GetAgent(route.AgentID)
if agent == nil {
agent = al.registry.GetDefaultAgent()
}
if agent == nil {
return "❌ 没有可用的代理。\nNo agent available.", true
}
sessionKey := route.SessionKey
if msg.SessionKey != "" && strings.HasPrefix(msg.SessionKey, "agent:") {
sessionKey = msg.SessionKey
}
// Clear the session history (deletes from memory and disk)
clearedCount := agent.Sessions.ClearHistory(sessionKey)
logger.InfoCF("agent", "User cleared session history",
map[string]any{
"channel": msg.Channel,
"chat_id": msg.ChatID,
"session_key": sessionKey,
"agent_id": route.AgentID,
"cleared_msgs": clearedCount,
})
return fmt.Sprintf("🧹 已清除会话历史（%d 条消息）。\n✨ 会话已重置，可以开始新的对话。\n\nCleared session history (%d messages). Session reset, ready for new conversation.",
clearedCount, clearedCount), true
case "/stop", "/cancel", "/abort":
// Phase 2: Cancel running tasks for the current session
runningTasks := al.taskManager.GetRunningTasksForSession(msg.Channel, msg.ChatID)

107
pkg/agent/message.go
View file

@ -20,6 +20,8 @@ const (
MessageTypeArtifact AgentMessageType = "artifact" // LLM-generated artifacts (code, images, documents)
MessageTypeAttachment AgentMessageType = "attachment" // User-provided attachments
MessageTypeEvent AgentMessageType = "event" // System events (task started, interrupted, etc.)
MessageTypeSubagentResult AgentMessageType = "subagent_result" // Results from subagent execution
MessageTypeToolProgress AgentMessageType = "tool_progress" // Progress updates from long-running tools
)
// ArtifactType categorizes the type of artifact
@ -63,6 +65,21 @@ type AgentMessage struct {
// ===== Event-specific Fields =====
EventType string `json:"event_type,omitempty"` // Event category (task_started, interrupted, etc.)
EventData map[string]any `json:"event_data,omitempty"` // Event-specific data
// ===== Subagent-specific Fields (Enhanced) =====
SubagentID string `json:"subagent_id,omitempty"` // Subagent task ID
SubagentLabel string `json:"subagent_label,omitempty"` // Subagent task label
SubagentStatus string `json:"subagent_status,omitempty"` // Status: running, completed, failed, canceled
Iterations int `json:"iterations,omitempty"` // Tool loop iterations executed
// ===== Progress-specific Fields =====
Progress float64 `json:"progress,omitempty"` // Progress percentage (0-100)
ProgressText string `json:"progress_text,omitempty"` // Human-readable progress message
// ===== Context and Routing Fields =====
OriginChannel string `json:"origin_channel,omitempty"` // Source channel for routing
OriginChatID string `json:"origin_chat_id,omitempty"` // Source chat ID for routing
AgentID string `json:"agent_id,omitempty"` // Agent that created this message
}
// ToLLMMessage converts an AgentMessage to a standard providers.Message
@ -78,7 +95,7 @@ func (am *AgentMessage) ToLLMMessage() providers.Message {
}
// ToLLMMessageWithContext converts an AgentMessage to a providers.Message,
// but includes contextual information about artifacts/attachments in the content.
// but includes contextual information about artifacts/attachments/subagents in the content.
func (am *AgentMessage) ToLLMMessageWithContext() providers.Message {
msg := am.ToLLMMessage()
@ -100,6 +117,24 @@ func (am *AgentMessage) ToLLMMessageWithContext() providers.Message {
}
}
// Add subagent result context if present
if am.Type == MessageTypeSubagentResult && am.SubagentID != "" {
if msg.Content == "" {
msg.Content = am.formatSubagentReference()
} else {
msg.Content = am.formatSubagentReference() + "\n\n" + msg.Content
}
}
// Add progress context if present
if am.Type == MessageTypeToolProgress && am.ProgressText != "" {
if msg.Content == "" {
msg.Content = am.formatProgressReference()
} else {
msg.Content = am.formatProgressReference() + "\n\n" + msg.Content
}
}
return msg
}
@ -286,3 +321,73 @@ func (am *AgentMessage) WithSessionID(sessionID string) *AgentMessage {
am.SessionID = sessionID
return am
}
// NewSubagentResultMessage creates a new subagent result message
func NewSubagentResultMessage(subagentID, label, status, content string, iterations int) *AgentMessage {
return &AgentMessage{
Role: "tool",
Content: content,
Type: MessageTypeSubagentResult,
SubagentID: subagentID,
SubagentLabel: label,
SubagentStatus: status,
Iterations: iterations,
Timestamp: time.Now(),
Metadata: map[string]any{
"source": "subagent",
"task_id": subagentID,
"task_label": label,
"status": status,
"iterations": iterations,
},
}
}
// NewProgressMessage creates a new progress update message
func NewProgressMessage(progressText string, progress float64) *AgentMessage {
return &AgentMessage{
Role: "assistant",
Content: progressText,
Type: MessageTypeToolProgress,
Progress: progress,
ProgressText: progressText,
Timestamp: time.Now(),
}
}
// WithOrigin sets the origin channel and chat ID (builder pattern)
func (am *AgentMessage) WithOrigin(channel, chatID string) *AgentMessage {
am.OriginChannel = channel
am.OriginChatID = chatID
return am
}
// WithAgentID sets the agent ID (builder pattern)
func (am *AgentMessage) WithAgentID(agentID string) *AgentMessage {
am.AgentID = agentID
return am
}
// formatSubagentReference creates a human-readable reference to subagent execution
func (am *AgentMessage) formatSubagentReference() string {
ref := "[Subagent"
if am.SubagentLabel != "" {
ref += " '" + am.SubagentLabel + "'"
}
if am.SubagentStatus != "" {
ref += " " + am.SubagentStatus
}
if am.Iterations > 0 {
ref += " after " + string(rune(am.Iterations)) + " iterations"
}
ref += "]"
return ref
}
// formatProgressReference creates a human-readable progress indicator
func (am *AgentMessage) formatProgressReference() string {
if am.Progress > 0 {
return "[Progress: " + string(rune(int(am.Progress))) + "%] " + am.ProgressText
}
return am.ProgressText
}

55
pkg/agent/message_converter.go
View file

@ -2,6 +2,7 @@ package agent
import (
"fmt"
"time"
"github.com/sipeed/picoclaw/pkg/providers"
)
@ -276,3 +277,57 @@ func MergeMessageHistories(histories ...[]*AgentMessage) []*AgentMessage {
// Note: For production use, consider more sophisticated deduplication
return result
}
// ============================================================================
// Phase 2: Session Integration - Conversion Functions (AgentMessage ↔ providers.Message)
// ============================================================================
// FromLLMMessage converts a providers.Message to AgentMessage
// This is primarily used for migrating legacy session data
func FromLLMMessage(msg providers.Message) *AgentMessage {
agentMsg := &AgentMessage{
Role: msg.Role,
Content: msg.Content,
ReasoningContent: msg.ReasoningContent,
ToolCalls: msg.ToolCalls,
ToolCallID: msg.ToolCallID,
Timestamp: time.Now(),
Metadata: make(map[string]any),
}
// Infer AgentMessageType from role
switch msg.Role {
case "user":
agentMsg.Type = MessageTypeUser
case "assistant":
agentMsg.Type = MessageTypeAssistant
case "tool":
agentMsg.Type = MessageTypeTool
case "system":
agentMsg.Type = MessageTypeSystem
default:
agentMsg.Type = MessageTypeUser // Default to user
}
return agentMsg
}
// FromLLMMessages converts a slice of providers.Message to AgentMessage slice
// Used for batch migration of legacy session histories
func FromLLMMessages(messages []providers.Message) []*AgentMessage {
result := make([]*AgentMessage, len(messages))
for i, msg := range messages {
result[i] = FromLLMMessage(msg)
}
return result
}
// ToLLMMessages converts a slice of AgentMessage to providers.Message slice
// Used when passing session history to LLM providers
func ToLLMMessages(messages []*AgentMessage) []providers.Message {
result := make([]providers.Message, len(messages))
for i, msg := range messages {
result[i] = msg.ToLLMMessageWithContext()
}
return result
}

1
pkg/config/config.go
View file

@ -182,7 +182,6 @@ type AgentDefaults struct {
MaxToolIterations int `json:"max_tool_iterations" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOOL_ITERATIONS"`
// Steering architecture (nanobot-inspired, opt-in)
EnableSteering bool `json:"enable_steering,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_ENABLE_STEERING"` // Enable message injection during tool execution
// Legacy: Phase 2 concurrent task management (to be deprecated)
MaxConcurrentTasks int `json:"max_concurrent_tasks,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_CONCURRENT_TASKS"` // Maximum concurrent tasks (0=unlimited)

278
pkg/session/manager.go
View file

@ -2,21 +2,62 @@ package session
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/providers"
)
// AgentMessage represents a message with extended metadata
// This is a local copy to avoid import cycle with pkg/agent
type AgentMessage struct {
// Core Fields (compatible with providers.Message)
Role string `json:"role"`
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content,omitempty"`
ToolCalls []providers.ToolCall `json:"tool_calls,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
// Extended Fields
Type string `json:"type"` // Semantic message type
Metadata map[string]any `json:"metadata,omitempty"` // Arbitrary metadata
Timestamp time.Time `json:"timestamp"` // Message creation time
SessionID string `json:"session_id,omitempty"` // Associated session
// Type-specific fields (artifact, attachment, event, subagent, progress, etc.)
ArtifactID string `json:"artifact_id,omitempty"`
ArtifactType string `json:"artifact_type,omitempty"`
ArtifactMIME string `json:"artifact_mime,omitempty"`
ArtifactSize int64 `json:"artifact_size,omitempty"`
AttachmentURL string `json:"attachment_url,omitempty"`
AttachmentSize int64 `json:"attachment_size,omitempty"`
AttachmentFilename string `json:"attachment_filename,omitempty"`
EventType string `json:"event_type,omitempty"`
EventData map[string]any `json:"event_data,omitempty"`
SubagentID string `json:"subagent_id,omitempty"`
SubagentLabel string `json:"subagent_label,omitempty"`
SubagentStatus string `json:"subagent_status,omitempty"`
Iterations int `json:"iterations,omitempty"`
Progress float64 `json:"progress,omitempty"`
ProgressText string `json:"progress_text,omitempty"`
OriginChannel string `json:"origin_channel,omitempty"`
OriginChatID string `json:"origin_chat_id,omitempty"`
AgentID string `json:"agent_id,omitempty"`
}
// Session stores conversation history using AgentMessage for rich metadata support
type Session struct {
Key string `json:"key"`
Messages []providers.Message `json:"messages"`
Messages []*AgentMessage `json:"messages"` // Phase 2: Changed to AgentMessage
Summary string `json:"summary,omitempty"`
Created time.Time `json:"created"`
Updated time.Time `json:"updated"`
Version int `json:"version"` // Schema version for migration tracking
}
type SessionManager struct {
@ -50,25 +91,140 @@ func (sm *SessionManager) GetOrCreate(key string) *Session {
session = &Session{
Key: key,
Messages: []providers.Message{},
Messages: []*AgentMessage{}, // Phase 2: Use AgentMessage
Created: time.Now(),
Updated: time.Now(),
Version: 1, // Current schema version
}
sm.sessions[key] = session
return session
}
func (sm *SessionManager) AddMessage(sessionKey, role, content string) {
sm.AddFullMessage(sessionKey, providers.Message{
Role: role,
Content: content,
})
// ============================================================================
// Conversion Helpers (to avoid import cycle with pkg/agent)
// ============================================================================
// fromLLMMessage converts a providers.Message to AgentMessage
func fromLLMMessage(msg providers.Message) *AgentMessage {
agentMsg := &AgentMessage{
Role: msg.Role,
Content: msg.Content,
ReasoningContent: msg.ReasoningContent,
ToolCalls: msg.ToolCalls,
ToolCallID: msg.ToolCallID,
Timestamp: time.Now(),
Metadata: make(map[string]any),
}
// Infer type from role
switch msg.Role {
case "user":
agentMsg.Type = "user"
case "assistant":
agentMsg.Type = "assistant"
case "tool":
agentMsg.Type = "tool"
case "system":
agentMsg.Type = "system"
default:
agentMsg.Type = "user"
}
return agentMsg
}
// AddFullMessage adds a complete message with tool calls and tool call ID to the session.
// This is used to save the full conversation flow including tool calls and tool results.
// fromLLMMessages converts a slice of providers.Message to AgentMessage slice
func fromLLMMessages(messages []providers.Message) []*AgentMessage {
result := make([]*AgentMessage, len(messages))
for i, msg := range messages {
result[i] = fromLLMMessage(msg)
}
return result
}
// toLLMMessage converts an AgentMessage to providers.Message
func toLLMMessage(msg *AgentMessage) providers.Message {
llmMsg := providers.Message{
Role: msg.Role,
Content: msg.Content,
ReasoningContent: msg.ReasoningContent,
ToolCalls: msg.ToolCalls,
ToolCallID: msg.ToolCallID,
}
// Add context for special message types
if msg.Type != "" && msg.Type != msg.Role {
// Add metadata context to content if it's a special type
switch msg.Type {
case "subagent_result":
if msg.SubagentLabel != "" || msg.SubagentStatus != "" {
prefix := fmt.Sprintf("[Subagent Result: %s, Status: %s]\n", msg.SubagentLabel, msg.SubagentStatus)
llmMsg.Content = prefix + llmMsg.Content
}
case "artifact":
if msg.ArtifactType != "" {
prefix := fmt.Sprintf("[Artifact: %s]\n", msg.ArtifactType)
llmMsg.Content = prefix + llmMsg.Content
}
case "attachment":
if msg.AttachmentFilename != "" {
prefix := fmt.Sprintf("[Attachment: %s]\n", msg.AttachmentFilename)
llmMsg.Content = prefix + llmMsg.Content
}
}
}
return llmMsg
}
// toLLMMessages converts a slice of AgentMessage to providers.Message slice
func toLLMMessages(messages []*AgentMessage) []providers.Message {
result := make([]providers.Message, len(messages))
for i, msg := range messages {
result[i] = toLLMMessage(msg)
}
return result
}
// AddMessage creates and adds an AgentMessage from role and content
// This is the primary method for adding simple messages to sessions
func (sm *SessionManager) AddMessage(sessionKey, role, content string) {
// Create AgentMessage with proper type inference
msg := &AgentMessage{
Role: role,
Content: content,
Timestamp: time.Now(),
Metadata: make(map[string]any),
}
// Infer type from role
switch role {
case "user":
msg.Type = "user"
case "assistant":
msg.Type = "assistant"
case "tool":
msg.Type = "tool"
case "system":
msg.Type = "system"
default:
msg.Type = "user"
}
sm.AddAgentMessage(sessionKey, msg)
}
// AddFullMessage adds a complete providers.Message to the session
// Converts it to AgentMessage for storage (backward compatibility)
func (sm *SessionManager) AddFullMessage(sessionKey string, msg providers.Message) {
agentMsg := fromLLMMessage(msg)
sm.AddAgentMessage(sessionKey, agentMsg)
}
// AddAgentMessage adds an AgentMessage directly to the session
// This is the core method that all other add methods delegate to
func (sm *SessionManager) AddAgentMessage(sessionKey string, msg *AgentMessage) {
sm.mu.Lock()
defer sm.mu.Unlock()
@ -76,8 +232,9 @@ func (sm *SessionManager) AddFullMessage(sessionKey string, msg providers.Messag
if !ok {
session = &Session{
Key: sessionKey,
Messages: []providers.Message{},
Messages: []*AgentMessage{},
Created: time.Now(),
Version: 1,
}
sm.sessions[sessionKey] = session
}
@ -86,16 +243,26 @@ func (sm *SessionManager) AddFullMessage(sessionKey string, msg providers.Messag
session.Updated = time.Now()
}
// GetHistory returns the session history as providers.Message slice
// Converts AgentMessage to providers.Message for backward compatibility with LLM calls
func (sm *SessionManager) GetHistory(key string) []providers.Message {
agentHistory := sm.GetAgentHistory(key)
return toLLMMessages(agentHistory)
}
// GetAgentHistory returns the raw AgentMessage slice for a session
// This provides access to full metadata and extended fields
func (sm *SessionManager) GetAgentHistory(key string) []*AgentMessage {
sm.mu.RLock()
defer sm.mu.RUnlock()
session, ok := sm.sessions[key]
if !ok {
return []providers.Message{}
return []*AgentMessage{}
}
history := make([]providers.Message, len(session.Messages))
// Return a copy to prevent external modification
history := make([]*AgentMessage, len(session.Messages))
copy(history, session.Messages)
return history
}
@ -122,6 +289,32 @@ func (sm *SessionManager) SetSummary(key string, summary string) {
}
}
// ClearHistory clears all messages from a session, resetting it to an empty state
// It also deletes the session from memory and removes the session file from disk
func (sm *SessionManager) ClearHistory(key string) int {
sm.mu.Lock()
defer sm.mu.Unlock()
session, ok := sm.sessions[key]
if !ok {
return 0
}
clearedCount := len(session.Messages)
// Remove session from memory
delete(sm.sessions, key)
// Delete session file from disk if storage is configured
if sm.storage != "" {
filename := sanitizeFilename(key)
sessionPath := filepath.Join(sm.storage, filename+".json")
_ = os.Remove(sessionPath) // Ignore error if file doesn't exist
}
return clearedCount
}
func (sm *SessionManager) TruncateHistory(key string, keepLast int) {
sm.mu.Lock()
defer sm.mu.Unlock()
@ -132,7 +325,7 @@ func (sm *SessionManager) TruncateHistory(key string, keepLast int) {
}
if keepLast <= 0 {
session.Messages = []providers.Message{}
session.Messages = []*AgentMessage{}
session.Updated = time.Now()
return
}
@ -182,12 +375,13 @@ func (sm *SessionManager) Save(key string) error {
Summary: stored.Summary,
Created: stored.Created,
Updated: stored.Updated,
Version: stored.Version,
}
if len(stored.Messages) > 0 {
snapshot.Messages = make([]providers.Message, len(stored.Messages))
snapshot.Messages = make([]*AgentMessage, len(stored.Messages))
copy(snapshot.Messages, stored.Messages)
} else {
snapshot.Messages = []providers.Message{}
snapshot.Messages = []*AgentMessage{}
}
sm.mu.RUnlock()
@ -233,6 +427,7 @@ func (sm *SessionManager) Save(key string) error {
return nil
}
// loadSessions loads session files with automatic migration from legacy format
func (sm *SessionManager) loadSessions() error {
files, err := os.ReadDir(sm.storage)
if err != nil {
@ -251,30 +446,77 @@ func (sm *SessionManager) loadSessions() error {
sessionPath := filepath.Join(sm.storage, file.Name())
data, err := os.ReadFile(sessionPath)
if err != nil {
logger.WarnF(fmt.Sprintf("Failed to read session file: %s", file.Name()), map[string]any{"error": err})
continue
}
var session Session
if err := json.Unmarshal(data, &session); err != nil {
// Try loading as legacy format ([]providers.Message)
var legacySession struct {
Key string `json:"key"`
Messages []providers.Message `json:"messages"`
Summary string `json:"summary,omitempty"`
Created time.Time `json:"created"`
Updated time.Time `json:"updated"`
}
err2 := json.Unmarshal(data, &legacySession)
if err2 == nil {
// Successfully loaded legacy format - migrate to new format
session = Session{
Key: legacySession.Key,
Messages: fromLLMMessages(legacySession.Messages),
Summary: legacySession.Summary,
Created: legacySession.Created,
Updated: legacySession.Updated,
Version: 1,
}
logger.Info(fmt.Sprintf("Migrated legacy session to new format: %s", session.Key))
// Save migrated session immediately
sm.sessions[session.Key] = &session
if saveErr := sm.Save(session.Key); saveErr != nil {
logger.WarnF(fmt.Sprintf("Failed to save migrated session: %s", session.Key), map[string]any{"error": saveErr})
}
continue
}
// Both attempts failed
logger.WarnF(fmt.Sprintf("Failed to load session file: %s", file.Name()), map[string]any{"new_format_error": err, "legacy_format_error": err2})
continue
}
// Successfully loaded new format
// Ensure version is set for sessions that might not have it
if session.Version == 0 {
session.Version = 1
}
sm.sessions[session.Key] = &session
}
return nil
}
// SetHistory updates the messages of a session.
// SetHistory updates the messages of a session from providers.Message slice
// Converts to AgentMessage for storage (backward compatibility)
func (sm *SessionManager) SetHistory(key string, history []providers.Message) {
agentHistory := fromLLMMessages(history)
sm.SetAgentHistory(key, agentHistory)
}
// SetAgentHistory updates the messages of a session with AgentMessage slice
// This is the core method for bulk history updates
func (sm *SessionManager) SetAgentHistory(key string, history []*AgentMessage) {
sm.mu.Lock()
defer sm.mu.Unlock()
session, ok := sm.sessions[key]
if ok {
// Create a deep copy to strictly isolate internal state
// from the caller's slice.
msgs := make([]providers.Message, len(history))
msgs := make([]*AgentMessage, len(history))
copy(msgs, history)
session.Messages = msgs
session.Updated = time.Now()

7
pkg/skills/loader.go
View file

@ -86,10 +86,13 @@ func (sl *SkillsLoader) ListSkills() []SkillInfo {
return
}
for _, d := range dirs {
if !d.IsDir() {
// Check if entry is a directory or a symlink pointing to a directory
entryPath := filepath.Join(dir, d.Name())
fileInfo, err := os.Stat(entryPath) // Stat follows symlinks
if err != nil || !fileInfo.IsDir() {
continue
}
skillFile := filepath.Join(dir, d.Name(), "SKILL.md")
skillFile := filepath.Join(entryPath, "SKILL.md")
if _, err := os.Stat(skillFile); err != nil {
continue
}

23
pkg/tools/subagent.go
View file

@ -3,6 +3,7 @@ package tools
import (
"context"
"fmt"
"strconv"
"sync"
"time"
@ -215,9 +216,28 @@ After completing the task, provide a clear summary of what was done.`
}
}
// Send announce message back to main agent
// Send announce message back to main agent with structured metadata
// Note: We build metadata directly to avoid circular import with pkg/agent
if sm.bus != nil {
// Build structured metadata for AgentMessage reconstruction in loop.go
metadata := make(map[string]string)
metadata["agent_message_type"] = "subagent_result"
metadata["subagent_id"] = task.ID
metadata["subagent_label"] = task.Label
metadata["subagent_status"] = task.Status
// Only set iterations if loopResult is not nil
if loopResult != nil {
metadata["iterations"] = strconv.Itoa(loopResult.Iterations)
} else {
metadata["iterations"] = "0"
}
metadata["origin_channel"] = task.OriginChannel
metadata["origin_chat_id"] = task.OriginChatID
metadata["timestamp"] = strconv.FormatInt(time.Now().UnixMilli(), 10)
// Format content for backward compatibility
announceContent := fmt.Sprintf("Task '%s' completed.\n\nResult:\n%s", task.Label, task.Result)
pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer pubCancel()
sm.bus.PublishInbound(pubCtx, bus.InboundMessage{
@ -226,6 +246,7 @@ After completing the task, provide a clear summary of what was done.`
// Format: "original_channel:original_chat_id" for routing back
ChatID: fmt.Sprintf("%s:%s", task.OriginChannel, task.OriginChatID),
Content: announceContent,
Metadata: metadata,
})
}
}