From d9e900cf29e6cd5e76a8d5977d2c449498738c96 Mon Sep 17 00:00:00 2001 From: leozeli Date: Wed, 4 Mar 2026 09:49:22 +0800 Subject: [PATCH] refactor: Remove steering arch and add message types --- go.mod | 4 +- pkg/agent/context.go | 19 ++ pkg/agent/loop.go | 471 ++++----------------------------- pkg/agent/message.go | 121 ++++++++- pkg/agent/message_converter.go | 55 ++++ pkg/config/config.go | 1 - pkg/session/manager.go | 286 ++++++++++++++++++-- pkg/skills/loader.go | 7 +- pkg/tools/subagent.go | 27 +- 9 files changed, 533 insertions(+), 458 deletions(-) diff --git a/go.mod b/go.mod index 9f755bbc9..1c699a724 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 6fccbaf53..5f6f2ef3f 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -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. diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 3c7995de1..abf32e2ad 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -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) - al.interruptHandler = NewBusInterruptHandler(msgBus, DefaultInterruptionConfig()) + // Initialize interrupt handler + al.interruptHandler = NewBusInterruptHandler(msgBus, DefaultInterruptionConfig()) - // Legacy TaskManager (Phase 2) - maxConcurrent := cfg.Agents.Defaults.MaxConcurrentTasks - if maxConcurrent < 0 { - maxConcurrent = 0 // Ensure no negative values, 0 = unlimited - } - al.taskManager = NewTaskManager(maxConcurrent) + // 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) - } - } + // Phase 2: Start task cleanup goroutine + go al.runTaskCleanup(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) diff --git a/pkg/agent/message.go b/pkg/agent/message.go index fa767b191..705617a60 100644 --- a/pkg/agent/message.go +++ b/pkg/agent/message.go @@ -17,9 +17,11 @@ const ( MessageTypeSystem AgentMessageType = "system" // Extended types for business semantics - MessageTypeArtifact AgentMessageType = "artifact" // LLM-generated artifacts (code, images, documents) - MessageTypeAttachment AgentMessageType = "attachment" // User-provided attachments - MessageTypeEvent AgentMessageType = "event" // System events (task started, interrupted, etc.) + 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 @@ -44,10 +46,10 @@ type AgentMessage struct { ToolCallID string `json:"tool_call_id,omitempty"` // ===== Extended Fields ===== - Type AgentMessageType `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 AgentMessageType `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 // ===== Artifact-specific Fields ===== ArtifactID string `json:"artifact_id,omitempty"` // Unique identifier for 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 +} diff --git a/pkg/agent/message_converter.go b/pkg/agent/message_converter.go index 2e8736aa2..65f42f70d 100644 --- a/pkg/agent/message_converter.go +++ b/pkg/agent/message_converter.go @@ -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 +} diff --git a/pkg/config/config.go b/pkg/config/config.go index e84f24389..e2d7f5598 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -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) diff --git a/pkg/session/manager.go b/pkg/session/manager.go index 08f0b0ad2..48c080905 100644 --- a/pkg/session/manager.go +++ b/pkg/session/manager.go @@ -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"` - Summary string `json:"summary,omitempty"` - Created time.Time `json:"created"` - Updated time.Time `json:"updated"` + Key string `json:"key"` + 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() diff --git a/pkg/skills/loader.go b/pkg/skills/loader.go index fcbcf934b..459f28bbb 100644 --- a/pkg/skills/loader.go +++ b/pkg/skills/loader.go @@ -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 } diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index 69f1a49a2..48ebe315c 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -3,6 +3,7 @@ package tools import ( "context" "fmt" + "strconv" "sync" "time" @@ -215,17 +216,37 @@ 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{ Channel: "system", SenderID: fmt.Sprintf("subagent:%s", task.ID), // Format: "original_channel:original_chat_id" for routing back - ChatID: fmt.Sprintf("%s:%s", task.OriginChannel, task.OriginChatID), - Content: announceContent, + ChatID: fmt.Sprintf("%s:%s", task.OriginChannel, task.OriginChatID), + Content: announceContent, + Metadata: metadata, }) } }