Merge pull request #24 from hobbyistlabs-coder/feature/background-summarization-10072916128906298897

⚡ Bolt: Offload summarization to background worker

AGENT_LOOP_IMPROVEMENTS.md

@@ -21,7 +21,7 @@ This document outlines a series of tasks to improve the main loop of the agentic
 ## Phase 3: Performance & Latency
 
 * [x] **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 an asynchronous worker. Instead of blocking the main thread, the worker outputs its execution trace and compressed context to a structured `/logs/{session}/{subagent}/` directory to provide a persistent audit trail while keeping the loop responsive.
+* [x] **Background Summarization:** Offload `maybeSummarize` and context compression to an asynchronous worker. Instead of blocking the main thread, the worker outputs its execution trace and compressed context to a structured `/logs/{session}/{subagent}/` directory to provide a persistent audit trail while keeping the loop responsive.
 * [ ] **Streaming Responses:** Implement streaming LLM token generation directly to the `bus.PublishOutbound` instead of waiting for full generation.
 
 ## Phase 4: Features

pkg/agent/loop.go

@@ -27,6 +27,7 @@ type AgentLoop struct {
 	state          *state.Manager
 	running        atomic.Bool
 	summarizing    sync.Map
+	summaryJobs    chan summaryJob
 	wg             sync.WaitGroup
 	fallback       *providers.FallbackChain
 	channelManager *channels.Manager
@@ -37,6 +38,11 @@ type AgentLoop struct {
 }
 
 // processOptions configures how a message is processed
+type summaryJob struct {
+	agent      *AgentInstance
+	sessionKey string
+}
+
 type processOptions struct {
 	SessionKey      string   // Session identifier for history/context
 	Channel         string   // Target channel for tool execution

pkg/agent/loop_init.go

@@ -52,6 +52,7 @@ func NewAgentLoop(
 		registry:    registry,
 		state:       stateManager,
 		summarizing: sync.Map{},
+		summaryJobs: make(chan summaryJob, 100),
 		fallback:    fallbackChain,
 		cmdRegistry: commands.NewRegistry(commands.BuiltinDefinitions()),
 	}
@@ -195,6 +196,25 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 		return err
 	}
 
+	// Start background summarization worker
+	al.wg.Add(1)
+	go func() {
+		defer al.wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case job, ok := <-al.summaryJobs:
+				if !ok {
+					return
+				}
+				logger.Debug("Memory threshold reached. Optimizing conversation history...")
+				al.summarizeSession(job.agent, job.sessionKey)
+				al.summarizing.Delete(job.agent.ID + ":" + job.sessionKey)
+			}
+		}
+	}()
+
 	for al.running.Load() {
 		select {
 		case <-ctx.Done():

pkg/agent/loop_summary.go

@@ -8,7 +8,10 @@ package agent
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"os"
+	"path/filepath"
 	"strings"
 	"time"
 	"unicode/utf8"
@@ -26,11 +29,15 @@ func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, c
 	if len(newHistory) > agent.SummarizeMessageThreshold || tokenEstimate > threshold {
 		summarizeKey := agent.ID + ":" + sessionKey
 		if _, loading := al.summarizing.LoadOrStore(summarizeKey, true); !loading {
-			go func() {
-				defer al.summarizing.Delete(summarizeKey)
-				logger.Debug("Memory threshold reached. Optimizing conversation history...")
-				al.summarizeSession(agent, sessionKey)
-			}()
+			// Send to background worker queue
+			select {
+			case al.summaryJobs <- summaryJob{agent: agent, sessionKey: sessionKey}:
+				// job accepted
+			default:
+				// Worker queue is full, delete from map so it can be retried later
+				logger.WarnCF("agent", "Summarization worker queue is full, skipping summarization", map[string]any{"session_key": sessionKey})
+				al.summarizing.Delete(summarizeKey)
+			}
 		}
 	}
 }
@@ -43,6 +50,23 @@ func (al *AgentLoop) forceCompression(agent *AgentInstance, sessionKey string) {
 		return
 	}
 
+	// Create logging directory: /logs/{session_id}/summarizer/
+	logDir := filepath.Join("logs", sessionKey, "summarizer")
+	if err := os.MkdirAll(logDir, 0755); err != nil {
+		logger.WarnCF("agent", "Failed to create summarizer log directory", map[string]any{"error": err.Error()})
+	}
+
+	// Dump input context
+	if b, err := json.MarshalIndent(history, "", "  "); err == nil {
+		_ = os.WriteFile(filepath.Join(logDir, "input_context.json"), b, 0644)
+	}
+
+	traceFile, _ := os.OpenFile(filepath.Join(logDir, "trace.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if traceFile != nil {
+		defer traceFile.Close()
+		fmt.Fprintf(traceFile, "[%s] Starting summarization for session: %s\n", time.Now().Format(time.RFC3339), sessionKey)
+	}
+
 	// Keep system prompt (usually [0]) and the very last message (user's trigger)
 	// We want to drop the oldest half of the *conversation*
 	// Assuming [0] is system, [1:] is conversation
@@ -101,6 +125,23 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) {
 		return
 	}
 
+	// Create logging directory: /logs/{session_id}/summarizer/
+	logDir := filepath.Join("logs", sessionKey, "summarizer")
+	if err := os.MkdirAll(logDir, 0755); err != nil {
+		logger.WarnCF("agent", "Failed to create summarizer log directory", map[string]any{"error": err.Error()})
+	}
+
+	// Dump input context
+	if b, err := json.MarshalIndent(history, "", "  "); err == nil {
+		_ = os.WriteFile(filepath.Join(logDir, "input_context.json"), b, 0644)
+	}
+
+	traceFile, _ := os.OpenFile(filepath.Join(logDir, "trace.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if traceFile != nil {
+		defer traceFile.Close()
+		fmt.Fprintf(traceFile, "[%s] Starting summarization for session: %s\n", time.Now().Format(time.RFC3339), sessionKey)
+	}
+
 	toSummarize := history[:len(history)-4]
 
 	// Oversized Message Guard
@@ -115,12 +156,18 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) {
 		msgTokens := len(m.Content) / 2
 		if msgTokens > maxMessageTokens {
 			omitted = true
+			if traceFile != nil {
+				fmt.Fprintf(traceFile, "[%s] Omitting oversized message from %s (length: %d)\n", time.Now().Format(time.RFC3339), m.Role, len(m.Content))
+			}
 			continue
 		}
 		validMessages = append(validMessages, m)
 	}
 
 	if len(validMessages) == 0 {
+		if traceFile != nil {
+			fmt.Fprintf(traceFile, "[%s] No valid messages to summarize.\n", time.Now().Format(time.RFC3339))
+		}
 		return
 	}
 
@@ -150,13 +197,23 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) {
 			s2,
 		)
 
+		if traceFile != nil {
+			fmt.Fprintf(traceFile, "[%s] Merging multi-part summaries...\n", time.Now().Format(time.RFC3339))
+		}
+
 		resp, err := al.retryLLMCall(ctx, agent, mergePrompt, llmMaxRetries)
 		if err == nil && resp.Content != "" {
 			finalSummary = resp.Content
 		} else {
+			if traceFile != nil {
+				fmt.Fprintf(traceFile, "[%s] LLM merge failed: %v. Falling back to concatenation.\n", time.Now().Format(time.RFC3339), err)
+			}
 			finalSummary = s1 + " " + s2
 		}
 	} else {
+		if traceFile != nil {
+			fmt.Fprintf(traceFile, "[%s] Summarizing single batch...\n", time.Now().Format(time.RFC3339))
+		}
 		finalSummary, _ = al.summarizeBatch(ctx, agent, validMessages, summary)
 	}
 
@@ -165,9 +222,18 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) {
 	}
 
 	if finalSummary != "" {
+		if traceFile != nil {
+			fmt.Fprintf(traceFile, "[%s] Summarization complete. Saving session.\n", time.Now().Format(time.RFC3339))
+		}
+		_ = os.WriteFile(filepath.Join(logDir, "result_summary.md"), []byte(finalSummary), 0644)
+
 		agent.Sessions.SetSummary(sessionKey, finalSummary)
 		agent.Sessions.TruncateHistory(sessionKey, 4)
 		agent.Sessions.Save(sessionKey)
+	} else {
+		if traceFile != nil {
+			fmt.Fprintf(traceFile, "[%s] Final summary is empty, skipping save.\n", time.Now().Format(time.RFC3339))
+		}
 	}
 }