Merge pull request #20 from hobbyistlabs-coder/feature/concurrent-message-processing-2066943644482552119

feat(agent): implement concurrent message processing

AGENT_LOOP_IMPROVEMENTS.md

@@ -13,7 +13,7 @@ This document outlines a series of tasks to improve the main loop of the agentic
 - [ ] **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.
+- [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 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.
 

pkg/agent/loop.go

@@ -27,6 +27,7 @@ type AgentLoop struct {
 	state          *state.Manager
 	running        atomic.Bool
 	summarizing    sync.Map
+	wg             sync.WaitGroup
 	fallback       *providers.FallbackChain
 	channelManager *channels.Manager
 	mediaStore     media.MediaStore

pkg/agent/loop_init.go

@@ -198,6 +198,7 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 	for al.running.Load() {
 		select {
 		case <-ctx.Done():
+			al.wg.Wait()
 			return nil
 		default:
 			msg, ok := al.bus.ConsumeInbound(ctx)
@@ -205,8 +206,10 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 				continue
 			}
 
+			al.wg.Add(1)
 			// Process message
-			func() {
+			go func(msg bus.InboundMessage) {
+				defer al.wg.Done()
 				// TODO: Re-enable media cleanup after inbound media is properly consumed by the agent.
 				// Currently disabled because files are deleted before the LLM can access their content.
 				// defer func() {
@@ -259,10 +262,11 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 						)
 					}
 				}
-			}()
+			}(msg)
 		}
 	}
 
+	al.wg.Wait()
 	return nil
 }