From 52b4d4cef42763bd2ee7234bb0446f56ab9d8bb2 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Fri, 13 Mar 2026 21:39:55 +0000
Subject: [PATCH] feat(agent): implement concurrent message processing

- Added a sync.WaitGroup to `AgentLoop` to track active concurrent requests.
- Refactored `AgentLoop.Run` to launch a new goroutine for each incoming message instead of processing sequentially.
- Ensured graceful shutdown by waiting on the WaitGroup during context cancellation or loop stop.
- Marked "Concurrent Message Processing" as complete in `AGENT_LOOP_IMPROVEMENTS.md`.

Co-authored-by: hobbyistlabs-coder <267281733+hobbyistlabs-coder@users.noreply.github.com>
---
 AGENT_LOOP_IMPROVEMENTS.md | 2 +-
 pkg/agent/loop.go          | 1 +
 pkg/agent/loop_init.go     | 8 ++++++--
 3 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/AGENT_LOOP_IMPROVEMENTS.md b/AGENT_LOOP_IMPROVEMENTS.md
index 856422849..3ca571a7d 100644
--- a/AGENT_LOOP_IMPROVEMENTS.md
+++ b/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.
 
diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go
index fe0bbe77e..89a49155c 100644
--- a/pkg/agent/loop.go
+++ b/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
diff --git a/pkg/agent/loop_init.go b/pkg/agent/loop_init.go
index 5cdce0c6d..8cda5a790 100644
--- a/pkg/agent/loop_init.go
+++ b/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
 }