From 999bf20144fac68f01eaa9dcf238d0e36e112be2 Mon Sep 17 00:00:00 2001 From: stevef Date: Fri, 27 Mar 2026 22:06:12 +0100 Subject: [PATCH] fixes --- pkg/agent/loop.go | 51 +++++++++++++++++++++++++++++++++++++++ pkg/agent/loop_test.go | 40 ++++++++++++++++++++++++++++++ pkg/channels/http/http.go | 45 ++++++++++++++++++++++++++++++++++ pkg/channels/manager.go | 3 +++ pkg/gateway/gateway.go | 6 +++-- 5 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 pkg/channels/http/http.go diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index e4f6abc64..b25203c37 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -111,6 +111,7 @@ type continuationTarget struct { const ( defaultResponse = "The model returned an empty response. This may indicate a provider error or token limit." toolLimitResponse = "I've reached `max_tool_iterations` without a final response. Increase `max_tool_iterations` in config.json if this task needs more tool steps." + toolRepeatLoopResponse = "Detected repeated tool calls without progress; stopping to avoid an infinite loop." handledToolResponseSummary = "Requested output delivered via tool attachment." sessionKeyAgentPrefix = "agent::" metadataKeyAccountID = "account_id" @@ -1914,6 +1915,9 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er } pendingMessages := append([]providers.Message(nil), ts.opts.InitialSteeringMessages...) var finalContent string + lastToolCallsFingerprint := "" + consecutiveRepeatedToolCalls := 0 + const maxConsecutiveRepeatedToolCalls = 3 turnLoop: for ts.currentIteration() < ts.agent.MaxIterations || len(pendingMessages) > 0 || func() bool { @@ -2411,6 +2415,53 @@ turnLoop: "iteration": iteration, }) + // Guardrail: if the model keeps requesting the exact same tool calls + // over and over (often due to missing/filtered tool results), stop + // early instead of running until max_tool_iterations. + type toolCallFP struct { + Name string `json:"name"` + Args json.RawMessage `json:"args"` + } + fpParts := make([]toolCallFP, 0, len(normalizedToolCalls)) + fingerprintBytes := make([]byte, 0) + for _, tc := range normalizedToolCalls { + argsJSON, err := json.Marshal(tc.Arguments) + if err != nil { + continue + } + fpParts = append(fpParts, toolCallFP{ + Name: tc.Name, + Args: json.RawMessage(argsJSON), + }) + } + if len(fpParts) > 0 { + if fp, err := json.Marshal(fpParts); err == nil { + fingerprintBytes = fp + } + } + if len(fingerprintBytes) > 0 { + toolCallsFingerprint := string(fingerprintBytes) + if toolCallsFingerprint == lastToolCallsFingerprint { + consecutiveRepeatedToolCalls++ + } else { + lastToolCallsFingerprint = toolCallsFingerprint + consecutiveRepeatedToolCalls = 1 + } + + if consecutiveRepeatedToolCalls >= maxConsecutiveRepeatedToolCalls { + turnStatus = TurnEndStatusError + finalContent = toolRepeatLoopResponse + logger.WarnCF("agent", "Stopping repeated tool call loop", + map[string]any{ + "agent_id": ts.agent.ID, + "fingerprint_repeats": consecutiveRepeatedToolCalls, + "tools": toolNames, + "iteration": iteration, + }) + break turnLoop + } + } + allResponsesHandled := len(normalizedToolCalls) > 0 assistantMsg := providers.Message{ Role: "assistant", diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index cc81f181c..7fc7dcb0b 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -2113,6 +2113,46 @@ func TestAgentLoop_ToolLimitUsesDedicatedFallback(t *testing.T) { } } +func TestAgentLoop_ToolRepeatLoopBreaksEarly(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + 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, + ModelName: "test-model", + MaxTokens: 4096, + // Keep this high so the loop-breaker (not the iteration limit) + // is what terminates the turn. + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &toolLimitOnlyProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + al.RegisterTool(&toolLimitTestTool{}) + + response, err := al.ProcessDirectWithChannel( + context.Background(), + "hello", + "tool-repeat-loop", + "test", + "direct", + ) + if err != nil { + t.Fatalf("ProcessDirectWithChannel failed: %v", err) + } + if response != toolRepeatLoopResponse { + t.Fatalf("response = %q, want %q", response, toolRepeatLoopResponse) + } +} + // TestProcessDirectWithChannel_TriggersMCPInitialization verifies that // ProcessDirectWithChannel triggers MCP initialization when MCP is enabled. // Note: Manager is only initialized when at least one MCP server is configured diff --git a/pkg/channels/http/http.go b/pkg/channels/http/http.go new file mode 100644 index 000000000..403e1ce23 --- /dev/null +++ b/pkg/channels/http/http.go @@ -0,0 +1,45 @@ +package http + +import ( + "context" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" +) + +func init() { + channels.RegisterFactory("http", NewHTTPChannel) +} + +type HTTPChannel struct { + *channels.BaseChannel +} + +func NewHTTPChannel(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + bc := channels.NewBaseChannel("http", nil, b, nil) + return &HTTPChannel{ + BaseChannel: bc, + }, nil +} + +func (c *HTTPChannel) Start(ctx context.Context) error { + c.SetRunning(true) + return nil +} + +func (c *HTTPChannel) Stop(ctx context.Context) error { + c.SetRunning(false) + return nil +} + +func (c *HTTPChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + logger.InfoCF("channels", "HTTP channel received outbound message", map[string]any{ + "chat_id": msg.ChatID, + "content": msg.Content, + }) + // For synchronous HTTP, the response is usually handled by the caller of ProcessDirectWithChannel. + // Asynchronous messages (e.g. from subagents) will just be logged here for now. + return nil +} diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 6d9f5eda8..29705e9bf 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -430,6 +430,9 @@ func (m *Manager) initChannels(channels *config.ChannelsConfig) error { m.initChannel("vk", "VK") } + // Always initialize HTTP channel as it is used for synchronous gateway chat + m.initChannel("http", "HTTP") + logger.InfoCF("channels", "Channel initialization completed", map[string]any{ "enabled_channels": len(m.channels), }) diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 397091d30..ea1997a43 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -21,6 +21,7 @@ import ( _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" _ "github.com/sipeed/picoclaw/pkg/channels/discord" _ "github.com/sipeed/picoclaw/pkg/channels/feishu" + _ "github.com/sipeed/picoclaw/pkg/channels/http" _ "github.com/sipeed/picoclaw/pkg/channels/irc" _ "github.com/sipeed/picoclaw/pkg/channels/line" _ "github.com/sipeed/picoclaw/pkg/channels/maixcam" @@ -227,10 +228,11 @@ func Run(debug bool, homePath, configPath string, allowEmptyStartup bool) error if cfg.Gateway.ChatEnabled { runningServices.HealthServer.SetChatFunc(func(ctx context.Context, message, sessionID, chatID string) (string, error) { if sessionID == "" { - sessionID = "http-chat" + sessionID = fmt.Sprintf("chat-%s", time.Now().Format("20060102-150405")) } if chatID == "" { - chatID = "chat" + // Default to sessionID to ensure isolation + chatID = sessionID } return agentLoop.ProcessDirectWithChannel(ctx, message, sessionID, "http", chatID) })