Merge pull request #3 from hobbyistlabs-coder/agent-loop-improvements-7773771147000011586

chore(agent): Agent loop refactoring, panic recovery, and exponential backoff

AGENT_LOOP_IMPROVEMENTS.md

@@ -0,0 +1,23 @@
+# Agent Loop Improvements Plan
+
+This document outlines a series of tasks to improve the main loop of the agentic stack (`AgentLoop` in `pkg/agent/loop.go`), focusing on Performance, Reliability, Architecture, and Features.
+
+## Phase 1: Architecture & Maintainability (Refactoring)
+- [ ] **Extract LLM Call & Retry Logic:** Refactor `runLLMIteration` by moving the LLM calling, fallback chain, and context window retry logic into a dedicated method (e.g., `executeLLMWithRetry`).
+- [ ] **Extract Tool Execution Logic:** Refactor `runLLMIteration` by moving the parallel tool execution (`sync.WaitGroup`), channel routing, and async callbacks into a dedicated method (e.g., `executeToolBatch`).
+- [ ] **State Machine / Explicit Flow:** Clean up the main loop logic to reduce nested `if/for` blocks and make the transition between generating, executing tools, and compressing context more explicit.
+
+## Phase 2: Reliability & Error Handling
+- [ ] **Graceful Recovery on Tool Panic:** Add `defer recover()` inside the parallel tool execution goroutines to prevent a panicked tool from crashing the entire `AgentLoop`. Return the panic as an error string to the LLM.
+- [ ] **Exponential Backoff:** Replace the linear backoff in LLM retries (`time.Duration(retry+1) * 5 * time.Second`) with exponential backoff and jitter to better handle rate limits.
+- [ ] **Granular Error Classification:** Update `LLMProvider` interfaces to return structured, typed errors (e.g., `providers.ErrContextLengthExceeded`) instead of relying on fragile string matching.
+
+## Phase 3: Performance & Latency
+- [ ] **Concurrent Message Processing:** Evaluate introducing a worker pool or goroutines to process independent user requests concurrently without blocking the entire agent instance.
+- [ ] **Background Summarization:** Offload `maybeSummarize` and context compression to a background worker to unblock the main thread and respond to the user faster.
+- [ ] **Streaming Responses:** Implement streaming LLM token generation directly to the `bus.PublishOutbound` instead of waiting for full generation.
+
+## Phase 4: Features
+- [ ] **Human-in-the-Loop:** Introduce an approval prompt state for high-risk tools (e.g., SQL execution) that pauses the loop until a user explicitly replies "Yes/No".
+- [ ] **Background / Long-Running Tasks:** Implement tools that allow the LLM to run slow operations in the background, releasing the main loop and notifying the user asynchronously upon completion.
+- [ ] **Multi-Agent Orchestration:** Create a Supervisor Loop where an agent can delegate tasks to other `AgentInstance`s and synthesize their results.

pkg/agent/loop_execute_llm.go

@@ -0,0 +1,154 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"jane/pkg/bus"
+	"jane/pkg/constants"
+	"jane/pkg/logger"
+	"jane/pkg/providers"
+)
+
+// executeLLMWithRetry handles calling the LLM, managing the fallback chain, and retrying
+// on timeouts or context window limits.
+func (al *AgentLoop) executeLLMWithRetry(
+	ctx context.Context,
+	agent *AgentInstance,
+	opts processOptions,
+	messages *[]providers.Message, // pointer to allow replacing messages slice on compression
+	providerToolDefs []providers.ToolDefinition,
+	activeCandidates []providers.FallbackCandidate,
+	activeModel string,
+	iteration int,
+) (*providers.LLMResponse, error) {
+
+	llmOpts := map[string]any{
+		"max_tokens":       agent.MaxTokens,
+		"temperature":      agent.Temperature,
+		"prompt_cache_key": agent.ID,
+	}
+
+	if agent.ThinkingLevel != ThinkingOff {
+		if tc, ok := agent.Provider.(providers.ThinkingCapable); ok && tc.SupportsThinking() {
+			llmOpts["thinking_level"] = string(agent.ThinkingLevel)
+		} else {
+			logger.WarnCF("agent", "thinking_level is set but current provider does not support it, ignoring",
+				map[string]any{"agent_id": agent.ID, "thinking_level": string(agent.ThinkingLevel)})
+		}
+	}
+
+	callLLM := func() (*providers.LLMResponse, error) {
+		if len(activeCandidates) > 1 && al.fallback != nil {
+			fbResult, fbErr := al.fallback.Execute(
+				ctx,
+				activeCandidates,
+				func(ctx context.Context, provider, model string) (*providers.LLMResponse, error) {
+					return agent.Provider.Chat(ctx, *messages, providerToolDefs, model, llmOpts)
+				},
+			)
+			if fbErr != nil {
+				return nil, fbErr
+			}
+			if fbResult.Provider != "" && len(fbResult.Attempts) > 0 {
+				logger.InfoCF(
+					"agent",
+					fmt.Sprintf("Fallback: succeeded with %s/%s after %d attempts",
+						fbResult.Provider, fbResult.Model, len(fbResult.Attempts)+1),
+					map[string]any{"agent_id": agent.ID, "iteration": iteration},
+				)
+			}
+			return fbResult.Response, nil
+		}
+		return agent.Provider.Chat(ctx, *messages, providerToolDefs, activeModel, llmOpts)
+	}
+
+	var response *providers.LLMResponse
+	var err error
+
+	// Retry loop for context/token errors
+	maxRetries := 2
+	for retry := 0; retry <= maxRetries; retry++ {
+		response, err = callLLM()
+		if err == nil {
+			break
+		}
+
+		errMsg := strings.ToLower(err.Error())
+
+		// Check if this is a network/HTTP timeout — not a context window error.
+		isTimeoutError := errors.Is(err, context.DeadlineExceeded) ||
+			strings.Contains(errMsg, "deadline exceeded") ||
+			strings.Contains(errMsg, "client.timeout") ||
+			strings.Contains(errMsg, "timed out") ||
+			strings.Contains(errMsg, "timeout exceeded")
+
+		// Detect real context window / token limit errors, excluding network timeouts.
+		isContextError := !isTimeoutError && (strings.Contains(errMsg, "context_length_exceeded") ||
+			strings.Contains(errMsg, "context window") ||
+			strings.Contains(errMsg, "maximum context length") ||
+			strings.Contains(errMsg, "token limit") ||
+			strings.Contains(errMsg, "too many tokens") ||
+			strings.Contains(errMsg, "max_tokens") ||
+			strings.Contains(errMsg, "invalidparameter") ||
+			strings.Contains(errMsg, "prompt is too long") ||
+			strings.Contains(errMsg, "request too large"))
+
+		if isTimeoutError && retry < maxRetries {
+			// Exponential backoff: 2s, 4s, 8s
+			backoff := time.Duration(1<<(retry+1)) * time.Second
+			logger.WarnCF("agent", "Timeout error, retrying after backoff", map[string]any{
+				"error":   err.Error(),
+				"retry":   retry,
+				"backoff": backoff.String(),
+			})
+			time.Sleep(backoff)
+			continue
+		}
+
+		if isContextError && retry < maxRetries {
+			logger.WarnCF(
+				"agent",
+				"Context window error detected, attempting compression",
+				map[string]any{
+					"error": err.Error(),
+					"retry": retry,
+				},
+			)
+
+			if retry == 0 && !constants.IsInternalChannel(opts.Channel) {
+				al.bus.PublishOutbound(ctx, bus.OutboundMessage{
+					Channel: opts.Channel,
+					ChatID:  opts.ChatID,
+					Content: "Context window exceeded. Compressing history and retrying...",
+				})
+			}
+
+			al.forceCompression(agent, opts.SessionKey)
+			newHistory := agent.Sessions.GetHistory(opts.SessionKey)
+			newSummary := agent.Sessions.GetSummary(opts.SessionKey)
+
+			*messages = agent.ContextBuilder.BuildMessages(
+				newHistory, newSummary, "",
+				nil, opts.Channel, opts.ChatID,
+			)
+			continue
+		}
+		break
+	}
+
+	if err != nil {
+		logger.ErrorCF("agent", "LLM call failed",
+			map[string]any{
+				"agent_id":  agent.ID,
+				"iteration": iteration,
+				"error":     err.Error(),
+			})
+		return nil, fmt.Errorf("LLM call failed after retries: %w", err)
+	}
+
+	return response, nil
+}

pkg/agent/loop_execute_tools.go

@@ -0,0 +1,123 @@
+package agent
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
+
+	"jane/pkg/bus"
+	"jane/pkg/logger"
+	"jane/pkg/providers"
+	"jane/pkg/tools"
+	"jane/pkg/utils"
+)
+
+type indexedAgentResult struct {
+	result *tools.ToolResult
+	tc     providers.ToolCall
+}
+
+// executeToolBatch runs multiple tools concurrently in goroutines and returns their results
+// in the same order they were requested.
+func (al *AgentLoop) executeToolBatch(
+	ctx context.Context,
+	agent *AgentInstance,
+	opts processOptions,
+	normalizedToolCalls []providers.ToolCall,
+	iteration int,
+) []indexedAgentResult {
+	agentResults := make([]indexedAgentResult, len(normalizedToolCalls))
+	var wg sync.WaitGroup
+
+	for i, tc := range normalizedToolCalls {
+		agentResults[i].tc = tc
+
+		wg.Add(1)
+		go func(idx int, tc providers.ToolCall) {
+			defer wg.Done()
+
+			// Panic recovery for robust tool execution
+			defer func() {
+				if r := recover(); r != nil {
+					errStr := fmt.Sprintf("Tool execution panicked: %v", r)
+					logger.ErrorCF("agent", "Tool panic recovered", map[string]any{
+						"agent_id": agent.ID,
+						"tool":     tc.Name,
+						"panic":    r,
+					})
+					agentResults[idx].result = &tools.ToolResult{
+						ForLLM: errStr,
+						Err:    fmt.Errorf("%s", errStr),
+					}
+				}
+			}()
+
+			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]any{
+					"agent_id":  agent.ID,
+					"tool":      tc.Name,
+					"iteration": iteration,
+				})
+
+			// Create async callback for tools that implement AsyncExecutor.
+			// When the background work completes, this publishes the result
+			// as an inbound system message so processSystemMessage routes it
+			// back to the user via the normal agent loop.
+			asyncCallback := func(_ context.Context, result *tools.ToolResult) {
+				// Send ForUser content directly to the user (immediate feedback),
+				// mirroring the synchronous tool execution path.
+				if !result.Silent && result.ForUser != "" {
+					outCtx, outCancel := context.WithTimeout(context.Background(), 5*time.Second)
+					defer outCancel()
+					_ = al.bus.PublishOutbound(outCtx, bus.OutboundMessage{
+						Channel: opts.Channel,
+						ChatID:  opts.ChatID,
+						Content: result.ForUser,
+					})
+				}
+
+				// Determine content for the agent loop (ForLLM or error).
+				content := result.ForLLM
+				if content == "" && result.Err != nil {
+					content = result.Err.Error()
+				}
+				if content == "" {
+					return
+				}
+
+				logger.InfoCF("agent", "Async tool completed, publishing result",
+					map[string]any{
+						"tool":        tc.Name,
+						"content_len": len(content),
+						"channel":     opts.Channel,
+					})
+
+				pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer pubCancel()
+				_ = al.bus.PublishInbound(pubCtx, bus.InboundMessage{
+					Channel:  "system",
+					SenderID: fmt.Sprintf("async:%s", tc.Name),
+					ChatID:   fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID),
+					Content:  content,
+				})
+			}
+
+			toolResult := agent.Tools.ExecuteWithContext(
+				ctx,
+				tc.Name,
+				tc.Arguments,
+				opts.Channel,
+				opts.ChatID,
+				asyncCallback,
+			)
+			agentResults[idx].result = toolResult
+		}(i, tc)
+	}
+	wg.Wait()
+
+	return agentResults
+}

pkg/tools/alpaca/alpaca.go

@@ -3,7 +3,6 @@ package alpaca
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/alpacahq/alpaca-trade-api-go/v3/alpaca"
 	"github.com/alpacahq/alpaca-trade-api-go/v3/marketdata"
@@ -60,7 +59,7 @@ func (t *AlpacaTool) Parameters() map[string]any {
 func (t *AlpacaTool) Execute(ctx context.Context, args map[string]any) *tools.ToolResult {
 	action, ok := args["action"].(string)
 	if !ok {
-		return tools.ErrorResult("missing or invalid ction\ parameter")
+		return tools.ErrorResult("missing or invalid 'action' parameter")
 	}
 
 	if !ok {
@@ -92,7 +91,7 @@ func (t *AlpacaTool) getEquity() *tools.ToolResult {
 	if err != nil {
 		return tools.ErrorResult(fmt.Sprintf("failed to get account: %v", err))
 	}
-	return tools.TextResult(fmt.Sprintf("Account Equity: $%s", acct.Equity.String()))
+	return tools.UserResult(fmt.Sprintf("Account Equity: $%s", acct.Equity.String()))
 }
 
 func (t *AlpacaTool) getPrice(symbol string) *tools.ToolResult {
@@ -103,13 +102,13 @@ func (t *AlpacaTool) getPrice(symbol string) *tools.ToolResult {
 	if err != nil {
 		return tools.ErrorResult(fmt.Sprintf("failed to get latest trade for %s: %v", symbol, err))
 	}
-	return tools.TextResult(fmt.Sprintf("Latest price for %s: $%.2f", symbol, trade.Price))
+	return tools.UserResult(fmt.Sprintf("Latest price for %s: $%.2f", symbol, trade.Price))
 }
 
 func (t *AlpacaTool) getSMA(symbol string) *tools.ToolResult {
 	req := marketdata.GetBarsRequest{
 		TimeFrame: marketdata.OneDay,
-		Limit:     10, // 10-day simple moving average
+		TotalLimit:     10, // 10-day simple moving average
 	}
 	bars, err := t.marketData.GetBars(symbol, req)
 	if err != nil {
@@ -123,7 +122,7 @@ func (t *AlpacaTool) getSMA(symbol string) *tools.ToolResult {
 		sum += bar.Close
 	}
 	sma := sum / float64(len(bars))
-	return tools.TextResult(fmt.Sprintf("10-Day SMA for %s: $%.2f", symbol, sma))
+	return tools.UserResult(fmt.Sprintf("10-Day SMA for %s: $%.2f", symbol, sma))
 }
 func init() {
 	// tools.Register(&AlpacaTool{}) // We will register it manually where we have access to config.