diff --git a/cmd/picoclaw/main.go b/cmd/picoclaw/main.go index 86463c661..61a0c1d94 100644 --- a/cmd/picoclaw/main.go +++ b/cmd/picoclaw/main.go @@ -26,10 +26,10 @@ import ( "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/cron" "github.com/sipeed/picoclaw/pkg/devices" + picofantasy "github.com/sipeed/picoclaw/pkg/fantasy" "github.com/sipeed/picoclaw/pkg/heartbeat" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/migrate" - "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" @@ -508,14 +508,21 @@ func agentCmd() { os.Exit(1) } - provider, err := providers.CreateProvider(cfg) + fantasyProvider, err := picofantasy.CreateProvider(cfg) if err != nil { fmt.Printf("Error creating provider: %v\n", err) os.Exit(1) } + ctx := context.Background() + languageModel, err := fantasyProvider.LanguageModel(ctx, picofantasy.ModelID(cfg)) + if err != nil { + fmt.Printf("Error creating language model: %v\n", err) + os.Exit(1) + } + msgBus := bus.NewMessageBus() - agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + agentLoop := agent.NewAgentLoop(cfg, msgBus, languageModel) // Print agent startup info (only for interactive mode) startupInfo := agentLoop.GetStartupInfo() @@ -527,7 +534,6 @@ func agentCmd() { }) if message != "" { - ctx := context.Background() response, err := agentLoop.ProcessDirect(ctx, message, sessionKey) if err != nil { fmt.Printf("Error: %v\n", err) @@ -643,14 +649,20 @@ func gatewayCmd() { os.Exit(1) } - provider, err := providers.CreateProvider(cfg) + fantasyProvider, err := picofantasy.CreateProvider(cfg) if err != nil { fmt.Printf("Error creating provider: %v\n", err) os.Exit(1) } + languageModel, err := fantasyProvider.LanguageModel(context.Background(), picofantasy.ModelID(cfg)) + if err != nil { + fmt.Printf("Error creating language model: %v\n", err) + os.Exit(1) + } + msgBus := bus.NewMessageBus() - agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + agentLoop := agent.NewAgentLoop(cfg, msgBus, languageModel) // Print agent startup info fmt.Println("\nšŸ“¦ Agent Status:") diff --git a/pkg/agent/context.go b/pkg/agent/context.go index cf5ce2913..bcbda68e5 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -1,6 +1,7 @@ package agent import ( + "context" "fmt" "os" "path/filepath" @@ -9,7 +10,8 @@ import ( "time" "github.com/sipeed/picoclaw/pkg/logger" - "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/memory" + "github.com/sipeed/picoclaw/pkg/messages" "github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/tools" ) @@ -17,7 +19,8 @@ import ( type ContextBuilder struct { workspace string skillsLoader *skills.SkillsLoader - memory *MemoryStore + memory *MemoryStore // Legacy file-based memory + memoryStore memory.Memory // New 3-tier MemGPT memory (may be nil) tools *tools.ToolRegistry // Direct reference to tool registry } @@ -48,6 +51,11 @@ func (cb *ContextBuilder) SetToolsRegistry(registry *tools.ToolRegistry) { cb.tools = registry } +// SetMemoryStore sets the 3-tier MemGPT memory store for working context injection. +func (cb *ContextBuilder) SetMemoryStore(ms memory.Memory) { + cb.memoryStore = ms +} + func (cb *ContextBuilder) getIdentity() string { now := time.Now().Format("2006-01-02 15:04 (Monday)") workspacePath, _ := filepath.Abs(filepath.Join(cb.workspace)) @@ -128,12 +136,20 @@ The following skills extend your capabilities. To use a skill, read its SKILL.md %s`, skillsSummary)) } - // Memory context + // Legacy file-based memory context memoryContext := cb.memory.GetMemoryContext() if memoryContext != "" { parts = append(parts, "# Memory\n\n"+memoryContext) } + // 3-tier MemGPT working context injection + if cb.memoryStore != nil { + wcSection := cb.buildWorkingContextSection() + if wcSection != "" { + parts = append(parts, wcSection) + } + } + // Join with "---" separator return strings.Join(parts, "\n\n---\n\n") } @@ -157,8 +173,46 @@ func (cb *ContextBuilder) LoadBootstrapFiles() string { return result } -func (cb *ContextBuilder) BuildMessages(history []providers.Message, summary string, currentMessage string, media []string, channel, chatID string) []providers.Message { - messages := []providers.Message{} +// buildWorkingContextSection returns the working context section for the system prompt. +// It includes the hot-tier working context buffer and memory usage instructions. +func (cb *ContextBuilder) buildWorkingContextSection() string { + if cb.memoryStore == nil { + return "" + } + + // Use a background context for system prompt building (non-blocking) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + var parts []string + + // Inject working context (hot tier) + wc, err := cb.memoryStore.GetWorkingContext(ctx, "picoclaw", "default") + if err == nil && wc != "" { + parts = append(parts, "## Working Context\n\n"+wc) + } + + // Memory system instructions + parts = append(parts, `## Memory System + +You have a 3-tier memory system accessible via the **memory** tool: + +- **Working Context** (hot): Always loaded. Use memory tool to update it with durable facts. +- **Recall** (warm): Scored memory items. Search returns the most relevant. +- **Archival** (cold): Large documents chunked and embedded for retrieval. + +Use the memory tool to: +- **search**: Find relevant memories across all tiers. +- **write**: Store important facts, preferences, decisions. +- **status**: Check memory pressure and usage. + +Store important user preferences, key decisions, and facts you want to remember long-term.`) + + return strings.Join(parts, "\n\n") +} + +func (cb *ContextBuilder) BuildMessages(history []messages.Message, summary string, currentMessage string, media []string, channel, chatID string) []messages.Message { + msgs := []messages.Message{} systemPrompt := cb.BuildSystemPrompt() @@ -189,49 +243,41 @@ func (cb *ContextBuilder) BuildMessages(history []providers.Message, summary str systemPrompt += "\n\n## Summary of Previous Conversation\n\n" + summary } - //This fix prevents the session memory from LLM failure due to elimination of toolu_IDs required from LLM - // --- INICIO DEL FIX --- - //Diegox-17 - for len(history) > 0 && (history[0].Role == "tool") { - logger.DebugCF("agent", "Removing orphaned tool message from history to prevent LLM error", - map[string]interface{}{"role": history[0].Role}) - history = history[1:] - } - //Diegox-17 - // --- FIN DEL FIX --- + // Note: Orphaned tool messages are now prevented at the source by + // tool-call-aware truncation in SessionManager.TruncateHistory(). - messages = append(messages, providers.Message{ + msgs = append(msgs, messages.Message{ Role: "system", Content: systemPrompt, }) - messages = append(messages, history...) + msgs = append(msgs, history...) - messages = append(messages, providers.Message{ + msgs = append(msgs, messages.Message{ Role: "user", Content: currentMessage, }) - return messages + return msgs } -func (cb *ContextBuilder) AddToolResult(messages []providers.Message, toolCallID, toolName, result string) []providers.Message { - messages = append(messages, providers.Message{ +func (cb *ContextBuilder) AddToolResult(msgs []messages.Message, toolCallID, toolName, result string) []messages.Message { + msgs = append(msgs, messages.Message{ Role: "tool", Content: result, ToolCallID: toolCallID, }) - return messages + return msgs } -func (cb *ContextBuilder) AddAssistantMessage(messages []providers.Message, content string, toolCalls []map[string]interface{}) []providers.Message { - msg := providers.Message{ +func (cb *ContextBuilder) AddAssistantMessage(msgs []messages.Message, content string, toolCalls []map[string]interface{}) []messages.Message { + msg := messages.Message{ Role: "assistant", Content: content, } // Always add assistant message, whether or not it has tool calls - messages = append(messages, msg) - return messages + msgs = append(msgs, msg) + return msgs } func (cb *ContextBuilder) loadSkills() string { diff --git a/pkg/agent/integration_test.go b/pkg/agent/integration_test.go new file mode 100644 index 000000000..d9f88e592 --- /dev/null +++ b/pkg/agent/integration_test.go @@ -0,0 +1,560 @@ +package agent + +import ( + "context" + "fmt" + "os" + "strings" + "testing" + "time" + + fantasy "charm.land/fantasy" + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/tools" +) + +// --- Mock language model that simulates tool calls --- + +// toolCallingModel simulates an LLM that requests tool calls on first round, +// then produces a final text response incorporating tool results. +type toolCallingModel struct { + callCount int +} + +func (m *toolCallingModel) Generate(_ context.Context, call fantasy.Call) (*fantasy.Response, error) { + m.callCount++ + + // First call: check if any tool results already exist in the prompt. + // If no tool results found, request a tool call. + hasToolResults := false + for _, msg := range call.Prompt { + for _, part := range msg.Content { + if part.GetType() == fantasy.ContentTypeToolResult { + hasToolResults = true + } + } + } + + if !hasToolResults && len(call.Tools) > 0 { + // Request a tool call + return &fantasy.Response{ + Content: fantasy.ResponseContent{ + fantasy.ToolCallContent{ + ToolCallID: "call-1", + ToolName: "echo", + Input: `{"text": "hello from tool"}`, + }, + }, + FinishReason: fantasy.FinishReasonToolCalls, + }, nil + } + + // After tool results: produce final response + return &fantasy.Response{ + Content: fantasy.ResponseContent{ + fantasy.TextContent{Text: "Integration test response with tool output"}, + }, + FinishReason: fantasy.FinishReasonStop, + }, nil +} + +func (m *toolCallingModel) Stream(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) { + resp, err := m.Generate(context.Background(), call) + if err != nil { + return nil, err + } + + return func(yield func(fantasy.StreamPart) bool) { + // Check if response has tool calls + hasToolCalls := false + for _, c := range resp.Content { + if c.GetType() == fantasy.ContentTypeToolCall { + hasToolCalls = true + } + } + + if hasToolCalls { + // Emit tool calls as stream parts + for _, c := range resp.Content { + if tc, ok := c.(fantasy.ToolCallContent); ok { + if !yield(fantasy.StreamPart{ + Type: fantasy.StreamPartTypeToolCall, + ID: tc.ToolCallID, + ToolCallName: tc.ToolName, + ToolCallInput: tc.Input, + }) { + return + } + } + } + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonToolCalls}) { + return + } + } else { + // Emit text as proper stream sequence + text := resp.Content.Text() + if text != "" { + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text-0"}) { + return + } + // Emit text in chunks to simulate real streaming + for i := 0; i < len(text); i += 10 { + end := i + 10 + if end > len(text) { + end = len(text) + } + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text-0", Delta: text[i:end]}) { + return + } + } + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text-0"}) { + return + } + } + yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}) + } + }, nil +} + +func (m *toolCallingModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *toolCallingModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *toolCallingModel) Provider() string { return "mock" } +func (m *toolCallingModel) Model() string { return "mock-tool-model" } + +// --- Simple echo tool for integration testing --- + +type echoTool struct{} + +func (t *echoTool) Name() string { return "echo" } +func (t *echoTool) Description() string { return "Echo the given text back" } +func (t *echoTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "text": map[string]interface{}{ + "type": "string", + "description": "Text to echo", + }, + }, + "required": []string{"text"}, + } +} + +func (t *echoTool) Execute(_ context.Context, args map[string]interface{}) *tools.ToolResult { + text, _ := args["text"].(string) + return &tools.ToolResult{ + ForLLM: "Echo: " + text, + ForUser: "", + Silent: true, + IsError: false, + } +} + +// --- Integration Tests --- + +// TestIntegration_FullAgentLoop_SimpleResponse tests the full agent loop +// with a simple mock model that returns text directly (no tool calls). +func TestIntegration_FullAgentLoop_SimpleResponse(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("Hello from Fantasy agent") + al := NewAgentLoop(cfg, msgBus, model) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + msg := bus.InboundMessage{ + Channel: "test", + SenderID: "user1", + ChatID: "chat1", + Content: "Say hello", + SessionKey: "test-session-simple", + } + + response, err := al.processMessage(ctx, msg) + if err != nil { + t.Fatalf("processMessage failed: %v", err) + } + + if response != "Hello from Fantasy agent" { + t.Errorf("Expected 'Hello from Fantasy agent', got: %s", response) + } + + // Verify session has messages saved + history := al.sessions.GetHistory("test-session-simple") + if len(history) == 0 { + t.Error("Expected session history to have messages") + } + + // First message should be the user's + foundUser := false + for _, m := range history { + if m.Role == "user" && m.Content == "Say hello" { + foundUser = true + break + } + } + if !foundUser { + t.Error("Expected user message in session history") + } +} + +// TestIntegration_FullAgentLoop_WithToolCalls tests the full agent loop +// including tool call execution and response incorporation. +func TestIntegration_FullAgentLoop_WithToolCalls(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-tools-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "mock-tool-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := &toolCallingModel{} + al := NewAgentLoop(cfg, msgBus, model) + + // Register the echo tool + al.RegisterTool(&echoTool{}) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + msg := bus.InboundMessage{ + Channel: "test", + SenderID: "user1", + ChatID: "chat1", + Content: "Use the echo tool", + SessionKey: "test-session-tools", + } + + response, err := al.processMessage(ctx, msg) + if err != nil { + t.Fatalf("processMessage failed: %v", err) + } + + // The model returns "Integration test response with tool output" after tool execution + if !strings.Contains(response, "Integration test response") { + t.Errorf("Expected response to contain 'Integration test response', got: %s", response) + } + + // Model should have been called at least twice (tool call + final response) + if model.callCount < 2 { + t.Errorf("Expected model to be called at least 2 times, got: %d", model.callCount) + } +} + +// TestIntegration_ProcessDirect tests the ProcessDirect method +// which is used by CLI mode for one-shot message processing. +func TestIntegration_ProcessDirect(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-direct-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("Direct CLI response") + al := NewAgentLoop(cfg, msgBus, model) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + response, err := al.ProcessDirect(ctx, "Direct message", "direct-session") + if err != nil { + t.Fatalf("ProcessDirect failed: %v", err) + } + + if response != "Direct CLI response" { + t.Errorf("Expected 'Direct CLI response', got: %s", response) + } +} + +// --- Streaming-specific mock --- + +// streamingModel simulates an LLM with proper streaming token emission. +// It emits tokens one word at a time via Stream() and also supports Generate() +// for non-streaming fallback. +type streamingModel struct { + words []string +} + +func newStreamingModel(text string) *streamingModel { + return &streamingModel{words: strings.Fields(text)} +} + +func (m *streamingModel) Generate(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) { + fullText := strings.Join(m.words, " ") + return &fantasy.Response{ + Content: fantasy.ResponseContent{fantasy.TextContent{Text: fullText}}, + FinishReason: fantasy.FinishReasonStop, + }, nil +} + +func (m *streamingModel) Stream(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + words := m.words + return func(yield func(fantasy.StreamPart) bool) { + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "s-0"}) { + return + } + for i, w := range words { + delta := w + if i < len(words)-1 { + delta += " " + } + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "s-0", Delta: delta}) { + return + } + } + if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "s-0"}) { + return + } + yield(fantasy.StreamPart{ + Type: fantasy.StreamPartTypeFinish, + FinishReason: fantasy.FinishReasonStop, + Usage: fantasy.Usage{InputTokens: 10, OutputTokens: int64(len(words)), TotalTokens: 10 + int64(len(words))}, + }) + }, nil +} + +func (m *streamingModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *streamingModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *streamingModel) Provider() string { return "mock" } +func (m *streamingModel) Model() string { return "streaming-mock" } + +// TestIntegration_Streaming_TextDeltas tests that the streaming agent loop +// publishes text deltas to the bus and returns the complete text. +func TestIntegration_Streaming_TextDeltas(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-stream-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "streaming-mock", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newStreamingModel("Hello from streaming agent response") + al := NewAgentLoop(cfg, msgBus, model) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Collect stream deltas in background + var deltas []string + var deltaDone = make(chan struct{}) + go func() { + defer close(deltaDone) + for { + msg, ok := msgBus.SubscribeOutbound(ctx) + if !ok { + return + } + if msg.StreamDelta { + deltas = append(deltas, msg.Content) + } + } + }() + + // Process with streaming + response, err := al.ProcessDirectStreaming(ctx, "Stream me", "stream-session", "test", "chat-1") + if err != nil { + t.Fatalf("ProcessDirectStreaming failed: %v", err) + } + + // Cancel to stop delta collector + cancel() + <-deltaDone + + // Verify complete response + if response != "Hello from streaming agent response" { + t.Errorf("Expected 'Hello from streaming agent response', got: %s", response) + } + + // Verify stream deltas were published + if len(deltas) == 0 { + t.Error("Expected stream deltas to be published to bus") + } + + // Reconstruct full text from deltas + fullFromDeltas := strings.Join(deltas, "") + if fullFromDeltas != "Hello from streaming agent response" { + t.Errorf("Delta reconstruction mismatch: got '%s'", fullFromDeltas) + } + + // Verify session was saved + history := al.sessions.GetHistory("stream-session") + if len(history) < 2 { // user + assistant + t.Errorf("Expected at least 2 messages in session, got %d", len(history)) + } +} + +// TestIntegration_Streaming_WithToolCalls tests streaming with a model +// that requests tool calls before producing a final streamed response. +func TestIntegration_Streaming_WithToolCalls(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-stream-tools-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "mock-tool-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := &toolCallingModel{} + al := NewAgentLoop(cfg, msgBus, model) + al.RegisterTool(&echoTool{}) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Collect deltas + var deltas []string + var deltaDone = make(chan struct{}) + go func() { + defer close(deltaDone) + for { + msg, ok := msgBus.SubscribeOutbound(ctx) + if !ok { + return + } + if msg.StreamDelta { + deltas = append(deltas, msg.Content) + } + } + }() + + response, err := al.ProcessDirectStreaming(ctx, "Use the echo tool (streaming)", "stream-tools-session", "test", "chat-1") + cancel() + <-deltaDone + + if err != nil { + t.Fatalf("ProcessDirectStreaming with tools failed: %v", err) + } + + if !strings.Contains(response, "Integration test response") { + t.Errorf("Expected response containing 'Integration test response', got: %s", response) + } + + // Model should have been called at least twice (tool call step + final response step) + if model.callCount < 2 { + t.Errorf("Expected at least 2 model calls, got %d", model.callCount) + } +} + +// TestIntegration_MultipleMessages tests sequential message processing +// to verify session history accumulation. +func TestIntegration_MultipleMessages(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-integration-multi-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("Response") + al := NewAgentLoop(cfg, msgBus, model) + + sessionKey := "multi-msg-session" + ctx := context.Background() + + // Send 3 messages + for i := 0; i < 3; i++ { + msg := bus.InboundMessage{ + Channel: "test", + SenderID: "user1", + ChatID: "chat1", + Content: fmt.Sprintf("Message %d", i+1), + SessionKey: sessionKey, + } + + _, err := al.processMessage(ctx, msg) + if err != nil { + t.Fatalf("processMessage #%d failed: %v", i+1, err) + } + } + + // Verify session history grew + history := al.sessions.GetHistory(sessionKey) + if len(history) < 6 { // At least 3 user messages + 3 assistant messages + t.Errorf("Expected at least 6 messages in history, got: %d", len(history)) + } +} diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 73e837119..0ea9b74a0 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -8,7 +8,6 @@ package agent import ( "context" - "encoding/json" "fmt" "os" "path/filepath" @@ -17,11 +16,15 @@ import ( "sync/atomic" "time" + fantasy "charm.land/fantasy" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" + picofantasy "github.com/sipeed/picoclaw/pkg/fantasy" "github.com/sipeed/picoclaw/pkg/logger" - "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/memory/delegate" + memstore "github.com/sipeed/picoclaw/pkg/memory/store" + "github.com/sipeed/picoclaw/pkg/messages" "github.com/sipeed/picoclaw/pkg/session" "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" @@ -29,18 +32,21 @@ import ( ) type AgentLoop struct { - bus *bus.MessageBus - provider providers.LLMProvider - workspace string - model string - contextWindow int // Maximum context window size in tokens - maxIterations int - sessions *session.SessionManager - state *state.Manager - contextBuilder *ContextBuilder - tools *tools.ToolRegistry - running atomic.Bool - summarizing sync.Map // Tracks which sessions are currently being summarized + bus *bus.MessageBus + languageModel fantasy.LanguageModel + workspace string + model string + contextWindow int // Maximum context window size in tokens + maxIterations int + sessions *session.SessionManager + state *state.Manager + contextBuilder *ContextBuilder + tools *tools.ToolRegistry + memoryStore *memstore.MemoryStore // 3-tier MemGPT memory (nil if init failed) + running atomic.Bool + summarizing sync.Map // Tracks which sessions are currently being summarized + summarizeFailures sync.Map // Tracks consecutive summarization failures per session (string -> int) + cfg *config.Config // Stored for subagent factory access } // processOptions configures how a message is processed @@ -53,6 +59,7 @@ type processOptions struct { EnableSummary bool // Whether to trigger summarization SendResponse bool // Whether to send response via bus NoHistory bool // If true, don't load session history (for heartbeat) + Streaming bool // If true, stream token deltas to bus via OnTextDelta } // createToolRegistry creates a tool registry with common tools. @@ -101,7 +108,7 @@ func createToolRegistry(workspace string, restrict bool, cfg *config.Config, msg return registry } -func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers.LLMProvider) *AgentLoop { +func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, model fantasy.LanguageModel) *AgentLoop { workspace := cfg.WorkspacePath() os.MkdirAll(workspace, 0755) @@ -111,7 +118,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers toolsRegistry := createToolRegistry(workspace, restrict, cfg, msgBus) // Create subagent manager with its own tool registry - subagentManager := tools.NewSubagentManager(provider, cfg.Agents.Defaults.Model, workspace, msgBus) + subagentManager := tools.NewSubagentManager(model, cfg.Agents.Defaults.Model, workspace, msgBus) subagentTools := createToolRegistry(workspace, restrict, cfg, msgBus) // Subagent doesn't need spawn/subagent tools to avoid recursion subagentManager.SetTools(subagentTools) @@ -133,18 +140,66 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers contextBuilder := NewContextBuilder(workspace) contextBuilder.SetToolsRegistry(toolsRegistry) + // Initialize 3-tier MemGPT memory system + var ms *memstore.MemoryStore + memDBPath := filepath.Join(workspace, "memory", "picoclaw.db") + os.MkdirAll(filepath.Dir(memDBPath), 0755) + + del, err := delegate.NewLibSQLDelegate(memDBPath) + if err != nil { + logger.WarnCF("agent", "Failed to create memory delegate, memory system disabled", + map[string]interface{}{"error": err.Error()}) + } else { + if err := del.Init(context.Background()); err != nil { + logger.WarnCF("agent", "Failed to init memory schema, memory system disabled", + map[string]interface{}{"error": err.Error()}) + del.Close() + } else { + chunker := memstore.NewMarkdownChunker(memstore.DefaultMarkdownChunkerConfig()) + ms = memstore.New(del, chunker, nil, memstore.Config{ + ContextWindowTokens: cfg.Agents.Defaults.MaxTokens, + OffloadThresholdTokens: 4000, + }) + contextBuilder.SetMemoryStore(ms) + + // Register the memory tool + memTool := NewMemGPTTool(ms, "picoclaw", "default") + toolsRegistry.Register(memTool) + + logger.InfoCF("agent", "3-tier memory system initialized", + map[string]interface{}{"db_path": memDBPath}) + } + } + + // Register meta-tools for progressive disclosure (tool_search + tool_call) + toolsRegistry.RegisterMetaTools() + + // If memory tool is a gateway, mark it visible in progressive mode + if ms != nil { + toolsRegistry.MarkGateway("memory") + } + + // Apply progressive disclosure config + if cfg.Tools.ProgressiveDisclosure { + toolsRegistry.SetProgressiveDisclosure(true) + logger.InfoCF("agent", "Progressive tool disclosure enabled", + map[string]interface{}{"gateway_tools": toolsRegistry.ListVisible()}) + } + return &AgentLoop{ bus: msgBus, - provider: provider, + languageModel: model, workspace: workspace, model: cfg.Agents.Defaults.Model, - contextWindow: cfg.Agents.Defaults.MaxTokens, // Restore context window for summarization + contextWindow: cfg.Agents.Defaults.MaxTokens, maxIterations: cfg.Agents.Defaults.MaxToolIterations, sessions: sessionsManager, state: stateManager, contextBuilder: contextBuilder, tools: toolsRegistry, + memoryStore: ms, summarizing: sync.Map{}, + cfg: cfg, } } @@ -192,6 +247,9 @@ func (al *AgentLoop) Run(ctx context.Context) error { func (al *AgentLoop) Stop() { al.running.Store(false) + if al.memoryStore != nil { + al.memoryStore.Close() + } } func (al *AgentLoop) RegisterTool(tool tools.Tool) { @@ -226,6 +284,29 @@ func (al *AgentLoop) ProcessDirectWithChannel(ctx context.Context, content, sess return al.processMessage(ctx, msg) } +// ProcessDirectStreaming processes a message with streaming token delivery. +// Text deltas are published to the bus as StreamDelta messages in real time. +func (al *AgentLoop) ProcessDirectStreaming(ctx context.Context, content, sessionKey, channel, chatID string) (string, error) { + msg := bus.InboundMessage{ + Channel: channel, + SenderID: "user", + ChatID: chatID, + Content: content, + SessionKey: sessionKey, + } + + return al.runAgentLoop(ctx, processOptions{ + SessionKey: msg.SessionKey, + Channel: msg.Channel, + ChatID: msg.ChatID, + UserMessage: msg.Content, + DefaultResponse: "I've completed processing but have no response to give.", + EnableSummary: true, + SendResponse: false, + Streaming: true, + }) +} + // ProcessHeartbeat processes a heartbeat request without session history. // Each heartbeat is independent and doesn't accumulate context. func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) { @@ -327,11 +408,14 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe } // runAgentLoop is the core message processing logic. -// It handles context building, LLM calls, tool execution, and response handling. +// It handles context building, Fantasy agent creation, tool execution, and response handling. +// When opts.Streaming is true, delegates to runAgentLoopStreaming for real-time token delivery. func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (string, error) { + if opts.Streaming { + return al.runAgentLoopStreaming(ctx, opts) + } // 0. Record last channel for heartbeat notifications (skip internal channels) if opts.Channel != "" && opts.ChatID != "" { - // Don't record internal channels (cli, system, subagent) if !constants.IsInternalChannel(opts.Channel) { channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID) if err := al.RecordLastChannel(channelKey); err != nil { @@ -344,13 +428,13 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str al.updateToolContexts(opts.Channel, opts.ChatID) // 2. Build messages (skip history for heartbeat) - var history []providers.Message + var history []messages.Message var summary string if !opts.NoHistory { history = al.sessions.GetHistory(opts.SessionKey) summary = al.sessions.GetSummary(opts.SessionKey) } - messages := al.contextBuilder.BuildMessages( + builtMsgs := al.contextBuilder.BuildMessages( history, summary, opts.UserMessage, @@ -362,30 +446,89 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str // 3. Save user message to session al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage) - // 4. Run LLM iteration loop - finalContent, iteration, err := al.runLLMIteration(ctx, messages, opts) - if err != nil { - return "", err + // 4. Split built messages into system prompt, conversation history, and current user prompt. + // BuildMessages returns: [system, ...history, user] + systemPrompt := "" + var historyMsgs []messages.Message + userPrompt := opts.UserMessage + + if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" { + systemPrompt = builtMsgs[0].Content + // History is everything between system and last user message. + if len(builtMsgs) > 2 { + historyMsgs = builtMsgs[1 : len(builtMsgs)-1] + } } - // If last tool had ForUser content and we already sent it, we might not need to send final response - // This is controlled by the tool's Silent flag and ForUser content + // 5. Convert history to Fantasy message format + fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs) - // 5. Handle empty response + // 6. Build adapted tools from PicoClaw registry (with optional offloading) + adaptCfg := picofantasy.AdaptedToolsConfig{ + MemStore: al.memoryStore, + AgentID: "picoclaw", + SessionKey: opts.SessionKey, + } + adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, adaptCfg) + + // 7. Create Fantasy agent with tools and configuration + agentOpts := []fantasy.AgentOption{ + fantasy.WithTools(adaptedTools...), + fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)), + } + if systemPrompt != "" { + agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt)) + } + agent := fantasy.NewAgent(al.languageModel, agentOpts...) + + logger.DebugCF("agent", "Fantasy agent created", + map[string]interface{}{ + "model": al.model, + "tools_count": len(adaptedTools), + "history_count": len(historyMsgs), + "max_iterations": al.maxIterations, + "memory_enabled": al.memoryStore != nil, + }) + + // 8. Call Fantasy agent.Generate() + result, err := agent.Generate(ctx, fantasy.AgentCall{ + Prompt: userPrompt, + Messages: fantasyHistory, + }) + if err != nil { + logger.ErrorCF("agent", "Fantasy Generate failed", + map[string]interface{}{ + "error": err.Error(), + }) + return "", fmt.Errorf("agent Generate failed: %w", err) + } + + // 9. Save all step messages to session + stepCount := len(result.Steps) + for _, step := range result.Steps { + stepMsgs := picofantasy.StepToMessages(step) + for _, m := range stepMsgs { + al.sessions.AddFullMessage(opts.SessionKey, m) + } + } + + // 10. Extract final text + finalContent := result.Response.Content.Text() + + // 11. Handle empty response if finalContent == "" { finalContent = opts.DefaultResponse } - // 6. Save final assistant message to session - al.sessions.AddMessage(opts.SessionKey, "assistant", finalContent) + // 12. Save session al.sessions.Save(opts.SessionKey) - // 7. Optional: summarization + // 13. Optional: summarization if opts.EnableSummary { al.maybeSummarize(opts.SessionKey) } - // 8. Optional: send response via bus + // 14. Optional: send response via bus if opts.SendResponse { al.bus.PublishOutbound(bus.OutboundMessage{ Channel: opts.Channel, @@ -394,179 +537,165 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str }) } - // 9. Log response + // 15. Log response responsePreview := utils.Truncate(finalContent, 120) logger.InfoCF("agent", fmt.Sprintf("Response: %s", responsePreview), map[string]interface{}{ "session_key": opts.SessionKey, - "iterations": iteration, + "steps": stepCount, "final_length": len(finalContent), }) return finalContent, nil } -// runLLMIteration executes the LLM call loop with tool handling. -// Returns the final content, iteration count, and any error. -func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.Message, opts processOptions) (string, int, error) { - iteration := 0 - var finalContent string - - for iteration < al.maxIterations { - iteration++ - - logger.DebugCF("agent", "LLM iteration", - map[string]interface{}{ - "iteration": iteration, - "max": al.maxIterations, - }) - - // Build tool definitions - providerToolDefs := al.tools.ToProviderDefs() - - // Log LLM request details - logger.DebugCF("agent", "LLM request", - map[string]interface{}{ - "iteration": iteration, - "model": al.model, - "messages_count": len(messages), - "tools_count": len(providerToolDefs), - "max_tokens": 8192, - "temperature": 0.7, - "system_prompt_len": len(messages[0].Content), - }) - - // Log full messages (detailed) - logger.DebugCF("agent", "Full LLM request", - map[string]interface{}{ - "iteration": iteration, - "messages_json": formatMessagesForLog(messages), - "tools_json": formatToolsForLog(providerToolDefs), - }) - - // Call LLM - response, err := al.provider.Chat(ctx, messages, providerToolDefs, al.model, map[string]interface{}{ - "max_tokens": 8192, - "temperature": 0.7, - }) - - if err != nil { - logger.ErrorCF("agent", "LLM call failed", - map[string]interface{}{ - "iteration": iteration, - "error": err.Error(), - }) - return "", iteration, fmt.Errorf("LLM call failed: %w", err) - } - - // Check if no tool calls - we're done - if len(response.ToolCalls) == 0 { - finalContent = response.Content - logger.InfoCF("agent", "LLM response without tool calls (direct answer)", - map[string]interface{}{ - "iteration": iteration, - "content_chars": len(finalContent), - }) - break - } - - // Log tool calls - toolNames := make([]string, 0, len(response.ToolCalls)) - for _, tc := range response.ToolCalls { - toolNames = append(toolNames, tc.Name) - } - logger.InfoCF("agent", "LLM requested tool calls", - map[string]interface{}{ - "tools": toolNames, - "count": len(response.ToolCalls), - "iteration": iteration, - }) - - // Build assistant message with tool calls - assistantMsg := providers.Message{ - Role: "assistant", - Content: response.Content, - } - for _, tc := range response.ToolCalls { - argumentsJSON, _ := json.Marshal(tc.Arguments) - assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, providers.ToolCall{ - ID: tc.ID, - Type: "function", - Function: &providers.FunctionCall{ - Name: tc.Name, - Arguments: string(argumentsJSON), - }, - }) - } - messages = append(messages, assistantMsg) - - // Save assistant message with tool calls to session - al.sessions.AddFullMessage(opts.SessionKey, assistantMsg) - - // Execute tool calls - for _, tc := range response.ToolCalls { - // Log tool call with arguments preview - argsJSON, _ := json.Marshal(tc.Arguments) - argsPreview := utils.Truncate(string(argsJSON), 200) - logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), - map[string]interface{}{ - "tool": tc.Name, - "iteration": iteration, - }) - - // Create async callback for tools that implement AsyncTool - // NOTE: Following openclaw's design, async tools do NOT send results directly to users. - // Instead, they notify the agent via PublishInbound, and the agent decides - // whether to forward the result to the user (in processSystemMessage). - asyncCallback := func(callbackCtx context.Context, result *tools.ToolResult) { - // Log the async completion but don't send directly to user - // The agent will handle user notification via processSystemMessage - if !result.Silent && result.ForUser != "" { - logger.InfoCF("agent", "Async tool completed, agent will handle notification", - map[string]interface{}{ - "tool": tc.Name, - "content_len": len(result.ForUser), - }) - } +// runAgentLoopStreaming uses Fantasy's agent.Stream() to stream token deltas +// to the bus in real time. Structure mirrors runAgentLoop but uses AgentStreamCall +// with OnTextDelta, OnStepFinish, and OnToolCall callbacks. +func (al *AgentLoop) runAgentLoopStreaming(ctx context.Context, opts processOptions) (string, error) { + // 0. Record last channel + if opts.Channel != "" && opts.ChatID != "" { + if !constants.IsInternalChannel(opts.Channel) { + channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID) + if err := al.RecordLastChannel(channelKey); err != nil { + logger.WarnCF("agent", "Failed to record last channel: %v", map[string]interface{}{"error": err.Error()}) } - - toolResult := al.tools.ExecuteWithContext(ctx, tc.Name, tc.Arguments, opts.Channel, opts.ChatID, asyncCallback) - - // Send ForUser content to user immediately if not Silent - if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse { - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: toolResult.ForUser, - }) - logger.DebugCF("agent", "Sent tool result to user", - map[string]interface{}{ - "tool": tc.Name, - "content_len": len(toolResult.ForUser), - }) - } - - // Determine content for LLM based on tool result - contentForLLM := toolResult.ForLLM - if contentForLLM == "" && toolResult.Err != nil { - contentForLLM = toolResult.Err.Error() - } - - toolResultMsg := providers.Message{ - Role: "tool", - Content: contentForLLM, - ToolCallID: tc.ID, - } - messages = append(messages, toolResultMsg) - - // Save tool result message to session - al.sessions.AddFullMessage(opts.SessionKey, toolResultMsg) } } - return finalContent, iteration, nil + // 1. Update tool contexts + al.updateToolContexts(opts.Channel, opts.ChatID) + + // 2. Build messages + var history []messages.Message + var summary string + if !opts.NoHistory { + history = al.sessions.GetHistory(opts.SessionKey) + summary = al.sessions.GetSummary(opts.SessionKey) + } + builtMsgs := al.contextBuilder.BuildMessages(history, summary, opts.UserMessage, nil, opts.Channel, opts.ChatID) + + // 3. Save user message + al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage) + + // 4. Split into system/history/user + systemPrompt := "" + var historyMsgs []messages.Message + userPrompt := opts.UserMessage + + if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" { + systemPrompt = builtMsgs[0].Content + if len(builtMsgs) > 2 { + historyMsgs = builtMsgs[1 : len(builtMsgs)-1] + } + } + + // 5. Convert history + fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs) + + // 6. Build adapted tools (with optional offloading) + streamAdaptCfg := picofantasy.AdaptedToolsConfig{ + MemStore: al.memoryStore, + AgentID: "picoclaw", + SessionKey: opts.SessionKey, + } + adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, streamAdaptCfg) + + // 7. Create Fantasy agent + agentOpts := []fantasy.AgentOption{ + fantasy.WithTools(adaptedTools...), + fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)), + } + if systemPrompt != "" { + agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt)) + } + fantasyAgent := fantasy.NewAgent(al.languageModel, agentOpts...) + + logger.DebugCF("agent", "Fantasy streaming agent created", + map[string]interface{}{ + "model": al.model, + "tools_count": len(adaptedTools), + "history_count": len(historyMsgs), + "max_iterations": al.maxIterations, + "memory_enabled": al.memoryStore != nil, + }) + + // 8. Build streaming call with callbacks + streamCall := fantasy.AgentStreamCall{ + Prompt: userPrompt, + Messages: fantasyHistory, + + // Stream text deltas to bus in real time + OnTextDelta: func(id, text string) error { + if opts.Channel != "" && opts.ChatID != "" { + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: text, + StreamDelta: true, + }) + } + return nil + }, + + // Save each step's messages to session as they complete + OnStepFinish: func(step fantasy.StepResult) error { + stepMsgs := picofantasy.StepToMessages(step) + for _, m := range stepMsgs { + al.sessions.AddFullMessage(opts.SessionKey, m) + } + return nil + }, + + // Log tool calls as they happen + OnToolCall: func(tc fantasy.ToolCallContent) error { + logger.DebugCF("agent", "Streaming tool call", + map[string]interface{}{ + "tool": tc.ToolName, + "id": tc.ToolCallID, + }) + return nil + }, + } + + // 9. Call Fantasy agent.Stream() + result, err := fantasyAgent.Stream(ctx, streamCall) + if err != nil { + logger.ErrorCF("agent", "Fantasy Stream failed", + map[string]interface{}{"error": err.Error()}) + return "", fmt.Errorf("agent Stream failed: %w", err) + } + + // 10. Extract final text + finalContent := result.Response.Content.Text() + if finalContent == "" { + finalContent = opts.DefaultResponse + } + + // 11. Save session + al.sessions.Save(opts.SessionKey) + + // 12. Summarization + if opts.EnableSummary { + al.maybeSummarize(opts.SessionKey) + } + + // 13. Log response + stepCount := len(result.Steps) + responsePreview := utils.Truncate(finalContent, 120) + logger.InfoCF("agent", fmt.Sprintf("Streaming response: %s", responsePreview), + map[string]interface{}{ + "session_key": opts.SessionKey, + "steps": stepCount, + "final_length": len(finalContent), + "total_tokens": result.TotalUsage.TotalTokens, + }) + + return finalContent, nil } +// runLLMIteration ā€” DELETED. Replaced by Fantasy's internal agent loop. + // updateToolContexts updates the context for tools that need channel/chatID info. func (al *AgentLoop) updateToolContexts(channel, chatID string) { // Use ContextualTool interface instead of type assertions @@ -621,16 +750,16 @@ func (al *AgentLoop) GetStartupInfo() map[string]interface{} { } // formatMessagesForLog formats messages for logging -func formatMessagesForLog(messages []providers.Message) string { - if len(messages) == 0 { +func formatMessagesForLog(msgs []messages.Message) string { + if len(msgs) == 0 { return "[]" } var result string result += "[\n" - for i, msg := range messages { + for i, msg := range msgs { result += fmt.Sprintf(" [%d] Role: %s\n", i, msg.Role) - if msg.ToolCalls != nil && len(msg.ToolCalls) > 0 { + if len(msg.ToolCalls) > 0 { result += " ToolCalls:\n" for _, tc := range msg.ToolCalls { result += fmt.Sprintf(" - ID: %s, Type: %s, Name: %s\n", tc.ID, tc.Type, tc.Name) @@ -652,25 +781,6 @@ func formatMessagesForLog(messages []providers.Message) string { return result } -// formatToolsForLog formats tool definitions for logging -func formatToolsForLog(tools []providers.ToolDefinition) string { - if len(tools) == 0 { - return "[]" - } - - var result string - result += "[\n" - for i, tool := range tools { - result += fmt.Sprintf(" [%d] Type: %s, Name: %s\n", i, tool.Type, tool.Function.Name) - result += fmt.Sprintf(" Description: %s\n", tool.Function.Description) - if len(tool.Function.Parameters) > 0 { - result += fmt.Sprintf(" Parameters: %s\n", utils.Truncate(fmt.Sprintf("%v", tool.Function.Parameters), 200)) - } - } - result += "]" - return result -} - // summarizeSession summarizes the conversation history for a session. func (al *AgentLoop) summarizeSession(sessionKey string) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) @@ -689,14 +799,13 @@ func (al *AgentLoop) summarizeSession(sessionKey string) { // Oversized Message Guard // Skip messages larger than 50% of context window to prevent summarizer overflow maxMessageTokens := al.contextWindow / 2 - validMessages := make([]providers.Message, 0) + validMessages := make([]messages.Message, 0) omitted := false for _, m := range toSummarize { if m.Role != "user" && m.Role != "assistant" { continue } - // Estimate tokens for this message msgTokens := len(m.Content) / 4 if msgTokens > maxMessageTokens { omitted = true @@ -710,7 +819,6 @@ func (al *AgentLoop) summarizeSession(sessionKey string) { } // Multi-Part Summarization - // Split into two parts if history is significant var finalSummary string if len(validMessages) > 10 { mid := len(validMessages) / 2 @@ -722,12 +830,9 @@ func (al *AgentLoop) summarizeSession(sessionKey string) { // Merge them mergePrompt := fmt.Sprintf("Merge these two conversation summaries into one cohesive summary:\n\n1: %s\n\n2: %s", s1, s2) - resp, err := al.provider.Chat(ctx, []providers.Message{{Role: "user", Content: mergePrompt}}, nil, al.model, map[string]interface{}{ - "max_tokens": 1024, - "temperature": 0.3, - }) + resp, err := al.callModel(ctx, mergePrompt) if err == nil { - finalSummary = resp.Content + finalSummary = resp } else { finalSummary = s1 + " " + s2 } @@ -743,11 +848,33 @@ func (al *AgentLoop) summarizeSession(sessionKey string) { al.sessions.SetSummary(sessionKey, finalSummary) al.sessions.TruncateHistory(sessionKey, 4) al.sessions.Save(sessionKey) + al.summarizeFailures.Delete(sessionKey) + } else { + var count int + if v, ok := al.summarizeFailures.Load(sessionKey); ok { + count = v.(int) + } + count++ + al.summarizeFailures.Store(sessionKey, count) + + const maxSummarizeFailures = 3 + const emergencyKeep = 10 + if count >= maxSummarizeFailures { + logger.ErrorCF("agent", "Summarization failed repeatedly, force-truncating session", + map[string]interface{}{ + "session": sessionKey, + "consecutive_failures": count, + "keep": emergencyKeep, + }) + al.sessions.TruncateHistory(sessionKey, emergencyKeep) + al.sessions.Save(sessionKey) + al.summarizeFailures.Delete(sessionKey) + } } } -// summarizeBatch summarizes a batch of messages. -func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []providers.Message, existingSummary string) (string, error) { +// summarizeBatch summarizes a batch of messages using the Fantasy LanguageModel directly. +func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []messages.Message, existingSummary string) (string, error) { prompt := "Provide a concise summary of this conversation segment, preserving core context and key points.\n" if existingSummary != "" { prompt += "Existing context: " + existingSummary + "\n" @@ -757,20 +884,32 @@ func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []providers.Messa prompt += fmt.Sprintf("%s: %s\n", m.Role, m.Content) } - response, err := al.provider.Chat(ctx, []providers.Message{{Role: "user", Content: prompt}}, nil, al.model, map[string]interface{}{ - "max_tokens": 1024, - "temperature": 0.3, + return al.callModel(ctx, prompt) +} + +// callModel makes a direct call to the Fantasy LanguageModel (no tools, no agent loop). +// Used for summarization and other simple generation tasks. +func (al *AgentLoop) callModel(ctx context.Context, prompt string) (string, error) { + temp := 0.3 + maxTokens := int64(1024) + + resp, err := al.languageModel.Generate(ctx, fantasy.Call{ + Prompt: fantasy.Prompt{ + fantasy.NewUserMessage(prompt), + }, + Temperature: &temp, + MaxOutputTokens: &maxTokens, }) if err != nil { return "", err } - return response.Content, nil + return resp.Content.Text(), nil } // estimateTokens estimates the number of tokens in a message list. -func (al *AgentLoop) estimateTokens(messages []providers.Message) int { +func (al *AgentLoop) estimateTokens(msgs []messages.Message) int { total := 0 - for _, m := range messages { + for _, m := range msgs { total += len(m.Content) / 4 // Simple heuristic: 4 chars per token } return total diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index c18220258..4b3a6b415 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -2,31 +2,55 @@ package agent import ( "context" + "fmt" "os" "path/filepath" "testing" "time" + fantasy "charm.land/fantasy" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/tools" ) -// mockProvider is a simple mock LLM provider for testing -type mockProvider struct{} +// mockLanguageModel is a simple mock fantasy.LanguageModel for testing +type mockLanguageModel struct { + response string +} -func (m *mockProvider) Chat(ctx context.Context, messages []providers.Message, tools []providers.ToolDefinition, model string, opts map[string]interface{}) (*providers.LLMResponse, error) { - return &providers.LLMResponse{ - Content: "Mock response", - ToolCalls: []providers.ToolCall{}, +func newMockLanguageModel(response string) *mockLanguageModel { + if response == "" { + response = "Mock response" + } + return &mockLanguageModel{response: response} +} + +func (m *mockLanguageModel) Generate(_ context.Context, call fantasy.Call) (*fantasy.Response, error) { + return &fantasy.Response{ + Content: fantasy.ResponseContent{fantasy.TextContent{Text: m.response}}, + FinishReason: fantasy.FinishReasonStop, }, nil } -func (m *mockProvider) GetDefaultModel() string { - return "mock-model" +func (m *mockLanguageModel) Stream(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) { + return func(yield func(fantasy.StreamPart) bool) { + yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, Delta: m.response}) + yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}) + }, nil } +func (m *mockLanguageModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *mockLanguageModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) { + return nil, fmt.Errorf("not implemented") +} + +func (m *mockLanguageModel) Provider() string { return "mock" } +func (m *mockLanguageModel) Model() string { return "mock-model" } + func TestRecordLastChannel(t *testing.T) { // Create temp workspace tmpDir, err := os.MkdirTemp("", "agent-test-*") @@ -49,8 +73,8 @@ func TestRecordLastChannel(t *testing.T) { // Create agent loop msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Test RecordLastChannel testChannel := "test-channel" @@ -66,7 +90,7 @@ func TestRecordLastChannel(t *testing.T) { } // Verify persistence by creating a new agent loop - al2 := NewAgentLoop(cfg, msgBus, provider) + al2 := NewAgentLoop(cfg, msgBus, model) if al2.state.GetLastChannel() != testChannel { t.Errorf("Expected persistent channel '%s', got '%s'", testChannel, al2.state.GetLastChannel()) } @@ -94,8 +118,8 @@ func TestRecordLastChatID(t *testing.T) { // Create agent loop msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Test RecordLastChatID testChatID := "test-chat-id-123" @@ -111,7 +135,7 @@ func TestRecordLastChatID(t *testing.T) { } // Verify persistence by creating a new agent loop - al2 := NewAgentLoop(cfg, msgBus, provider) + al2 := NewAgentLoop(cfg, msgBus, model) if al2.state.GetLastChatID() != testChatID { t.Errorf("Expected persistent chat ID '%s', got '%s'", testChatID, al2.state.GetLastChatID()) } @@ -139,8 +163,8 @@ func TestNewAgentLoop_StateInitialized(t *testing.T) { // Create agent loop msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Verify state manager is initialized if al.state == nil { @@ -174,8 +198,8 @@ func TestToolRegistry_ToolRegistration(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Register a custom tool customTool := &mockCustomTool{} @@ -220,8 +244,8 @@ func TestToolContext_Updates(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &simpleMockProvider{response: "OK"} - _ = NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("OK") + _ = NewAgentLoop(cfg, msgBus, model) // Verify that ContextualTool interface is defined and can be implemented // This test validates the interface contract exists @@ -251,8 +275,8 @@ func TestToolRegistry_GetDefinitions(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Register a test tool and verify it shows up in startup info testTool := &mockCustomTool{} @@ -295,8 +319,8 @@ func TestAgentLoop_GetStartupInfo(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) info := al.GetStartupInfo() @@ -342,8 +366,8 @@ func TestAgentLoop_Stop(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &mockProvider{} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("") + al := NewAgentLoop(cfg, msgBus, model) // Note: running is only set to true when Run() is called // We can't test that without starting the event loop @@ -358,21 +382,6 @@ func TestAgentLoop_Stop(t *testing.T) { // Mock implementations for testing -type simpleMockProvider struct { - response string -} - -func (m *simpleMockProvider) Chat(ctx context.Context, messages []providers.Message, tools []providers.ToolDefinition, model string, opts map[string]interface{}) (*providers.LLMResponse, error) { - return &providers.LLMResponse{ - Content: m.response, - ToolCalls: []providers.ToolCall{}, - }, nil -} - -func (m *simpleMockProvider) GetDefaultModel() string { - return "mock-model" -} - // mockCustomTool is a simple mock tool for registration testing type mockCustomTool struct{} @@ -464,8 +473,8 @@ func TestToolResult_SilentToolDoesNotSendUserMessage(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &simpleMockProvider{response: "File operation complete"} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("File operation complete") + al := NewAgentLoop(cfg, msgBus, model) helper := testHelper{al: al} // ReadFileTool returns SilentResult, which should not send user message @@ -506,8 +515,8 @@ func TestToolResult_UserFacingToolDoesSendMessage(t *testing.T) { } msgBus := bus.NewMessageBus() - provider := &simpleMockProvider{response: "Command output: hello world"} - al := NewAgentLoop(cfg, msgBus, provider) + model := newMockLanguageModel("Command output: hello world") + al := NewAgentLoop(cfg, msgBus, model) helper := testHelper{al: al} // ExecTool returns UserResult, which should send user message diff --git a/pkg/agent/memgpt_tool.go b/pkg/agent/memgpt_tool.go new file mode 100644 index 000000000..c4b97200c --- /dev/null +++ b/pkg/agent/memgpt_tool.go @@ -0,0 +1,106 @@ +// PicoClaw - Ultra-lightweight personal AI agent +// License: MIT +// +// Copyright (c) 2026 PicoClaw contributors + +package agent + +import ( + "context" + "encoding/json" + + memstore "github.com/sipeed/picoclaw/pkg/memory/store" + "github.com/sipeed/picoclaw/pkg/tools" +) + +// MemGPTTool wraps store.MemoryTool as a PicoClaw tools.Tool so it can be +// registered in the ToolRegistry and executed by the Fantasy agent loop. +type MemGPTTool struct { + inner *memstore.MemoryTool +} + +var _ tools.Tool = (*MemGPTTool)(nil) + +// NewMemGPTTool creates a PicoClaw tool wrapper around a MemoryTool. +func NewMemGPTTool(store *memstore.MemoryStore, agentID, session string) *MemGPTTool { + return &MemGPTTool{ + inner: memstore.NewMemoryTool(store, agentID, session), + } +} + +func (t *MemGPTTool) Name() string { return "memory" } + +func (t *MemGPTTool) Description() string { + return "Manage the agent's 3-tier memory system. Actions: search (hybrid keyword+vector), read (by ID), write (to recall or archival), update, delete, status (context pressure). All memory persists across sessions." +} + +func (t *MemGPTTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "action": map[string]interface{}{ + "type": "string", + "description": "The memory operation to perform.", + "enum": []string{"search", "read", "write", "update", "delete", "status"}, + }, + "query": map[string]interface{}{ + "type": "string", + "description": "Search query (for action=search).", + }, + "id": map[string]interface{}{ + "type": "string", + "description": "Memory ID (for action=read/update/delete).", + }, + "content": map[string]interface{}{ + "type": "string", + "description": "Content to store or update (for action=write/update).", + }, + "source": map[string]interface{}{ + "type": "string", + "description": "Source label for archival writes.", + }, + "sector": map[string]interface{}{ + "type": "string", + "description": "Memory sector: episodic, semantic, procedural, reflective.", + "enum": []string{"episodic", "semantic", "procedural", "reflective"}, + }, + "tags": map[string]interface{}{ + "type": "string", + "description": "Comma-separated tags for the memory entry.", + }, + "tier": map[string]interface{}{ + "type": "string", + "description": "Storage tier: recall (warm, default) or archival (cold, chunked+embedded).", + "enum": []string{"recall", "archival"}, + }, + "limit": map[string]interface{}{ + "type": "integer", + "description": "Max results for search. Default: 5.", + }, + }, + "required": []string{"action"}, + } +} + +func (t *MemGPTTool) Execute(ctx context.Context, args map[string]interface{}) *tools.ToolResult { + // Marshal the args back to JSON for the inner MemoryTool.Execute() + input, err := json.Marshal(args) + if err != nil { + return tools.ErrorResult("invalid arguments: " + err.Error()) + } + + result, err := t.inner.Execute(ctx, string(input)) + if err != nil { + return tools.ErrorResult("memory tool error: " + err.Error()) + } + + return &tools.ToolResult{ + ForLLM: result, + } +} + +// UpdateSession rebinds the inner MemoryTool to a new session. +// Called when the agent switches sessions. +func (t *MemGPTTool) UpdateSession(store *memstore.MemoryStore, agentID, session string) { + t.inner = memstore.NewMemoryTool(store, agentID, session) +} diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go index 6283251a4..1fcb0f130 100644 --- a/pkg/bus/bus.go +++ b/pkg/bus/bus.go @@ -3,6 +3,8 @@ package bus import ( "context" "sync" + + "github.com/sipeed/picoclaw/pkg/logger" ) type MessageBus struct { @@ -24,6 +26,22 @@ func (mb *MessageBus) PublishInbound(msg InboundMessage) { mb.inbound <- msg } +// TryPublishInbound attempts a non-blocking send. Returns false if the buffer +// is full (message dropped). Callers should log and/or notify the user. +func (mb *MessageBus) TryPublishInbound(msg InboundMessage) bool { + select { + case mb.inbound <- msg: + return true + default: + logger.InfoCF("bus", "Inbound buffer full, message dropped", + map[string]interface{}{ + "channel": msg.Channel, + "sender_id": msg.SenderID, + }) + return false + } +} + func (mb *MessageBus) ConsumeInbound(ctx context.Context) (InboundMessage, bool) { select { case msg := <-mb.inbound: diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 44f9181a5..d2f010322 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -11,9 +11,10 @@ type InboundMessage struct { } type OutboundMessage struct { - Channel string `json:"channel"` - ChatID string `json:"chat_id"` - Content string `json:"content"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + StreamDelta bool `json:"stream_delta,omitempty"` // true = partial token (not a complete message) } type MessageHandler func(InboundMessage) error diff --git a/pkg/channels/base.go b/pkg/channels/base.go index 8d2d9a65b..427550f63 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -4,8 +4,10 @@ import ( "context" "fmt" "strings" + "sync/atomic" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/logger" ) type Channel interface { @@ -20,19 +22,20 @@ type Channel interface { type BaseChannel struct { config interface{} bus *bus.MessageBus - running bool + running atomic.Bool name string allowList []string } func NewBaseChannel(name string, config interface{}, bus *bus.MessageBus, allowList []string) *BaseChannel { - return &BaseChannel{ + bc := &BaseChannel{ config: config, bus: bus, name: name, allowList: allowList, - running: false, } + bc.running.Store(false) + return bc } func (c *BaseChannel) Name() string { @@ -40,7 +43,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 { @@ -100,9 +103,15 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st Metadata: metadata, } - c.bus.PublishInbound(msg) + if !c.bus.TryPublishInbound(msg) { + logger.InfoCF(c.name, "Message dropped (bus full)", + map[string]interface{}{ + "sender_id": msg.SenderID, + "channel": msg.Channel, + }) + } } func (c *BaseChannel) setRunning(running bool) { - c.running = running + c.running.Store(running) } diff --git a/pkg/channels/maixcam.go b/pkg/channels/maixcam.go index 5fc19adbe..27185c0c6 100644 --- a/pkg/channels/maixcam.go +++ b/pkg/channels/maixcam.go @@ -72,7 +72,7 @@ func (c *MaixCamChannel) acceptConnections(ctx context.Context) { default: conn, err := c.listener.Accept() if err != nil { - if c.running { + if c.IsRunning() { logger.ErrorCF("maixcam", "Failed to accept connection", map[string]interface{}{ "error": err.Error(), }) diff --git a/pkg/config/config.go b/pkg/config/config.go index bbfa2e40b..8f9ba1411 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -175,6 +175,7 @@ type ProviderConfig struct { APIBase string `json:"api_base" env:"PICOCLAW_PROVIDERS_{{.Name}}_API_BASE"` Proxy string `json:"proxy,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_PROXY"` AuthMethod string `json:"auth_method,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_AUTH_METHOD"` + Timeout int `json:"timeout,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_TIMEOUT"` // seconds, 0 = default (120s) } type GatewayConfig struct { @@ -199,7 +200,8 @@ type WebToolsConfig struct { } type ToolsConfig struct { - Web WebToolsConfig `json:"web"` + Web WebToolsConfig `json:"web"` + ProgressiveDisclosure bool `json:"progressive_disclosure" env:"PICOCLAW_TOOLS_PROGRESSIVE_DISCLOSURE"` } func DefaultConfig() *Config { @@ -332,9 +334,48 @@ func LoadConfig(path string) (*Config, error) { return nil, err } + if warnings := cfg.Validate(); len(warnings) > 0 { + for _, w := range warnings { + fmt.Fprintf(os.Stderr, "config warning: %s\n", w) + } + } + return cfg, nil } +// Validate checks configuration for common issues. Returns a list of +// warning strings. These are warnings rather than hard errors to maintain +// backward compatibility, but they indicate values that will likely cause +// problems at runtime. +func (c *Config) Validate() []string { + c.mu.RLock() + defer c.mu.RUnlock() + + var warnings []string + + if c.Agents.Defaults.Model == "" { + warnings = append(warnings, "agents.defaults.model is empty: no default LLM model configured") + } + + if c.Agents.Defaults.MaxTokens <= 0 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.max_tokens=%d: should be > 0", c.Agents.Defaults.MaxTokens)) + } + + if c.Agents.Defaults.MaxToolIterations <= 0 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.max_tool_iterations=%d: should be > 0", c.Agents.Defaults.MaxToolIterations)) + } + + if c.Gateway.Port < 0 || c.Gateway.Port > 65535 { + warnings = append(warnings, fmt.Sprintf("gateway.port=%d: must be in range 1-65535", c.Gateway.Port)) + } + + if c.Heartbeat.Interval < 0 { + warnings = append(warnings, fmt.Sprintf("heartbeat.interval=%d: must be >= 0", c.Heartbeat.Interval)) + } + + return warnings +} + func SaveConfig(path string, cfg *Config) error { cfg.mu.RLock() defer cfg.mu.RUnlock() diff --git a/pkg/heartbeat/service.go b/pkg/heartbeat/service.go index dfdaef58b..45d970045 100644 --- a/pkg/heartbeat/service.go +++ b/pkg/heartbeat/service.go @@ -146,14 +146,11 @@ func (hs *HeartbeatService) runLoop(stopChan chan struct{}) { func (hs *HeartbeatService) executeHeartbeat() { hs.mu.RLock() enabled := hs.enabled + stopped := hs.stopChan == nil handler := hs.handler - if !hs.enabled || hs.stopChan == nil { - hs.mu.RUnlock() - return - } hs.mu.RUnlock() - if !enabled { + if !enabled || stopped { return }