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>
This commit is contained in:
google-labs-jules[bot] 2026-03-13 21:39:55 +00:00
parent f77867d8e9
commit 52b4d4cef4
3 changed files with 8 additions and 3 deletions

View file

@ -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. - [ ] **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 ## 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. - [ ] **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. - [ ] **Streaming Responses:** Implement streaming LLM token generation directly to the `bus.PublishOutbound` instead of waiting for full generation.

View file

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

View file

@ -198,6 +198,7 @@ func (al *AgentLoop) Run(ctx context.Context) error {
for al.running.Load() { for al.running.Load() {
select { select {
case <-ctx.Done(): case <-ctx.Done():
al.wg.Wait()
return nil return nil
default: default:
msg, ok := al.bus.ConsumeInbound(ctx) msg, ok := al.bus.ConsumeInbound(ctx)
@ -205,8 +206,10 @@ func (al *AgentLoop) Run(ctx context.Context) error {
continue continue
} }
al.wg.Add(1)
// Process message // 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. // 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. // Currently disabled because files are deleted before the LLM can access their content.
// defer func() { // defer func() {
@ -259,10 +262,11 @@ func (al *AgentLoop) Run(ctx context.Context) error {
) )
} }
} }
}() }(msg)
} }
} }
al.wg.Wait()
return nil return nil
} }