diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go
index 7bbcd549f..c2d04aba7 100644
--- a/pkg/agent/loop.go
+++ b/pkg/agent/loop.go
@@ -251,63 +251,65 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return nil
-		case msg := <-al.bus.InboundChan():
+		case msg, ok := <-al.bus.InboundChan():
+			if !ok {
+				return nil
+			}
 			// Process message
-			func() {
-				// 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() {
-				// 	if al.mediaStore != nil && msg.MediaScope != "" {
-				// 		if releaseErr := al.mediaStore.ReleaseAll(msg.MediaScope); releaseErr != nil {
-				// 			logger.WarnCF("agent", "Failed to release media", map[string]any{
-				// 				"scope": msg.MediaScope,
-				// 				"error": releaseErr.Error(),
-				// 			})
-				// 		}
-				// 	}
-				// }()
+			// 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() {
+			// 	if al.mediaStore != nil && msg.MediaScope != "" {
+			// 		if releaseErr := al.mediaStore.ReleaseAll(msg.MediaScope); releaseErr != nil {
+			// 			logger.WarnCF("agent", "Failed to release media", map[string]any{
+			// 				"scope": msg.MediaScope,
+			// 				"error": releaseErr.Error(),
+			// 			})
+			// 		}
+			// 	}
+			// }()
 
-				response, err := al.processMessage(ctx, msg)
-				if err != nil {
-					response = fmt.Sprintf("Error processing message: %v", err)
-				}
+			response, err := al.processMessage(ctx, msg)
+			if err != nil {
+				response = fmt.Sprintf("Error processing message: %v", err)
+			}
 
-				if response != "" {
-					// Check if the message tool already sent a response during this round.
-					// If so, skip publishing to avoid duplicate messages to the user.
-					// Use default agent's tools to check (message tool is shared).
-					alreadySent := false
-					defaultAgent := al.GetRegistry().GetDefaultAgent()
-					if defaultAgent != nil {
-						if tool, ok := defaultAgent.Tools.Get("message"); ok {
-							if mt, ok := tool.(*tools.MessageTool); ok {
-								alreadySent = mt.HasSentInRound()
-							}
+			if response != "" {
+				// Check if the message tool already sent a response during this round.
+				// If so, skip publishing to avoid duplicate messages to the user.
+				// Use default agent's tools to check (message tool is shared).
+				alreadySent := false
+				defaultAgent := al.GetRegistry().GetDefaultAgent()
+				if defaultAgent != nil {
+					if tool, ok := defaultAgent.Tools.Get("message"); ok {
+						if mt, ok := tool.(*tools.MessageTool); ok {
+							alreadySent = mt.HasSentInRound()
 						}
 					}
-
-					if !alreadySent {
-						al.bus.PublishOutbound(ctx, bus.OutboundMessage{
-							Channel: msg.Channel,
-							ChatID:  msg.ChatID,
-							Content: response,
-						})
-						logger.InfoCF("agent", "Published outbound response",
-							map[string]any{
-								"channel":     msg.Channel,
-								"chat_id":     msg.ChatID,
-								"content_len": len(response),
-							})
-					} else {
-						logger.DebugCF(
-							"agent",
-							"Skipped outbound (message tool already sent)",
-							map[string]any{"channel": msg.Channel},
-						)
-					}
 				}
-			}()
+
+				if !alreadySent {
+					al.bus.PublishOutbound(ctx, bus.OutboundMessage{
+						Channel: msg.Channel,
+						ChatID:  msg.ChatID,
+						Content: response,
+					})
+					logger.InfoCF("agent", "Published outbound response",
+						map[string]any{
+							"channel":     msg.Channel,
+							"chat_id":     msg.ChatID,
+							"content_len": len(response),
+						})
+				} else {
+					logger.DebugCF(
+						"agent",
+						"Skipped outbound (message tool already sent)",
+						map[string]any{"channel": msg.Channel},
+					)
+				}
+			}
 		default:
+			time.Sleep(time.Second * 5)
 		}
 	}
 
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index 8e71484a8..3d08bda4f 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -88,11 +88,13 @@ func (mb *MessageBus) OutboundMediaChan() <-chan OutboundMediaMessage {
 
 func (mb *MessageBus) Close() {
 	mb.closeOnce.Do(func() {
-		mb.closed.Store(true)
-
 		// notify all blocked publishers to exit
 		close(mb.done)
 
+		// because every publisher will check mb.closed before acquiring wg
+		// so we can be sure that new publishers will not be added new messages after this point
+		mb.closed.Store(true)
+
 		// wait for all ongoing Publish calls to finish, ensuring all messages have been sent to channels or exited
 		mb.wg.Wait()
 
diff --git a/pkg/channels/whatsapp_native/whatsapp_command_test.go b/pkg/channels/whatsapp_native/whatsapp_command_test.go
index cc2dcb619..e51bec392 100644
--- a/pkg/channels/whatsapp_native/whatsapp_command_test.go
+++ b/pkg/channels/whatsapp_native/whatsapp_command_test.go
@@ -43,14 +43,19 @@ func TestHandleIncoming_DoesNotConsumeGenericCommandsLocally(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 
-	inbound, ok := messageBus.ConsumeInbound(ctx)
-	if !ok {
-		t.Fatal("expected inbound message to be forwarded")
-	}
-	if inbound.Channel != "whatsapp_native" {
-		t.Fatalf("channel=%q", inbound.Channel)
-	}
-	if inbound.Content != "/new" {
-		t.Fatalf("content=%q", inbound.Content)
+	select {
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for message to be forwarded")
+		return
+	case inbound, ok := <-messageBus.InboundChan():
+		if !ok {
+			t.Fatal("expected inbound message to be forwarded")
+		}
+		if inbound.Channel != "whatsapp_native" {
+			t.Fatalf("channel=%q", inbound.Channel)
+		}
+		if inbound.Content != "/new" {
+			t.Fatalf("content=%q", inbound.Content)
+		}
 	}
 }