fix: test

Author: standujar

packages/server/src/__tests__/integration/http-transport.test.ts

@@ -259,8 +259,8 @@ describe('HTTP Transport Integration', () => {
       expect(response.status).toBe(201);
       expect(data.userMessage).toHaveProperty('id');
       expect(data.userMessage).toHaveProperty('content');
-      expect(data.userMessage).toHaveProperty('authorId');
-      expect(data.userMessage).toHaveProperty('createdAt');
+      expect(data.userMessage).toHaveProperty('author_id');
+      expect(data.userMessage).toHaveProperty('created_at');
     });
 
     it('should include agentResponse with text', async () => {

packages/server/src/__tests__/integration/sse-transport.test.ts

@@ -57,6 +57,7 @@ describe('SSE Transport Integration', () => {
   let baseUrl: string;
   let agentId: UUID;
   let sessionId: UUID;
+  let originalHandleMessage: ElizaOS['handleMessage'];
 
   beforeAll(async () => {
     // Arrange - Setup server
@@ -77,11 +78,48 @@ describe('SSE Transport Integration', () => {
 
     agentId = runtime.agentId;
 
-    // Mock processActions to simulate streaming
-    runtime.processActions = async (_message: unknown, responses: unknown[]) => {
-      (responses as Array<{ text: string }>).push({
-        text: 'SSE streaming response',
-      });
+    // Mock elizaOS.handleMessage to support SSE callbacks
+    const elizaOS = serverFixture.getServer().elizaOS!;
+    originalHandleMessage = elizaOS.handleMessage.bind(elizaOS);
+    elizaOS.handleMessage = async (
+      _agentId: UUID,
+      message: any,
+      options?: {
+        onStreamChunk?: (chunk: string) => Promise<void>;
+        onResponse?: (content: any) => Promise<void>;
+      }
+    ): Promise<HandleMessageResult> => {
+      const mockMessageId = stringToUuid(`mock-msg-${Date.now()}`);
+
+      // Simulate streaming chunks if callback provided
+      if (options?.onStreamChunk) {
+        await options.onStreamChunk('SSE ');
+        await options.onStreamChunk('streaming ');
+        await options.onStreamChunk('response');
+      }
+
+      // Call onResponse callback for SSE done event
+      if (options?.onResponse) {
+        await options.onResponse({ text: 'SSE streaming response' });
+      }
+
+      return {
+        messageId: mockMessageId,
+        userMessage: {
+          id: mockMessageId,
+          entityId: message.entityId,
+          agentId: _agentId,
+          roomId: message.roomId,
+          content: message.content,
+          createdAt: Date.now(),
+        } as any,
+        processing: {
+          didRespond: true,
+          responseContent: { text: 'SSE streaming response' },
+          responseMessages: [],
+          state: {} as any,
+        },
+      };
     };
 
     // Create a session for testing
@@ -102,6 +140,10 @@ describe('SSE Transport Integration', () => {
   }, 30000);
 
   afterAll(async () => {
+    // Restore original handleMessage
+    if (originalHandleMessage && serverFixture.getServer().elizaOS) {
+      serverFixture.getServer().elizaOS!.handleMessage = originalHandleMessage;
+    }
     await agentFixture.cleanup();
     await serverFixture.cleanup();
   });
@@ -220,7 +262,7 @@ describe('SSE Transport Integration', () => {
       expect(response.status).toBe(400);
 
       const data = await response.json();
-      expect(data.error).toContain('Invalid transport');
+      expect(data.error.message).toContain('Invalid transport');
     });
 
     it('should reject non-string transport parameter', async () => {
@@ -242,7 +284,7 @@ describe('SSE Transport Integration', () => {
       expect(response.status).toBe(400);
 
       const data = await response.json();
-      expect(data.error).toContain('Transport must be a string');
+      expect(data.error.message).toContain('Transport must be a string');
     });
   });
 

packages/server/src/__tests__/integration/websocket-transport.test.ts

@@ -21,6 +21,7 @@ import {
   CharacterBuilder,
   stringToUuid,
 } from '../index';
+import internalMessageBus from '../../services/message-bus';
 
 const DEFAULT_MESSAGE_SERVER_ID = '00000000-0000-0000-0000-000000000000' as UUID;
 
@@ -181,7 +182,7 @@ describe('WebSocket Transport Integration', () => {
       expect(response.status).toBe(400);
 
       const data = await response.json();
-      expect(data.error).toContain('Invalid transport');
+      expect(data.error.message).toContain('Invalid transport');
     });
 
     it('should reject non-string transport parameter', async () => {
@@ -203,7 +204,7 @@ describe('WebSocket Transport Integration', () => {
       expect(response.status).toBe(400);
 
       const data = await response.json();
-      expect(data.error).toContain('Transport must be a string');
+      expect(data.error.message).toContain('Transport must be a string');
     });
   });
 
@@ -274,7 +275,46 @@ describe('WebSocket Transport Integration', () => {
   });
 
   describe('WebSocket + Socket.IO Integration', () => {
-    it('should receive agent response via Socket.IO after HTTP acknowledgment', async () => {
+    it('should emit new_message to internal bus for agent processing', async () => {
+      // Arrange - Setup listener on internal bus to verify message is emitted
+      const authorId = stringToUuid('ws-bus-user');
+      let receivedMessage: any = null;
+
+      const messageHandler = (data: any) => {
+        receivedMessage = data;
+      };
+      internalMessageBus.on('new_message', messageHandler);
+
+      // Act - Send message via HTTP with websocket transport
+      const response = await fetch(`${baseUrl}/api/messaging/sessions/${sessionId}/messages`, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({
+          content: 'Hello, should emit to internal bus',
+          author_id: authorId,
+          transport: 'websocket',
+        }),
+      });
+
+      // Small delay to allow async event emission
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      // Cleanup
+      internalMessageBus.off('new_message', messageHandler);
+
+      // Assert HTTP response is immediate
+      expect(response.status).toBe(201);
+      const httpData = await response.json();
+      expect(httpData.success).toBe(true);
+      expect(httpData.agentResponse).toBeUndefined();
+
+      // Assert message was emitted to internal bus
+      expect(receivedMessage).toBeDefined();
+      expect(receivedMessage.content).toBe('Hello, should emit to internal bus');
+      expect(receivedMessage.channel_id).toBeDefined();
+    });
+
+    it('should connect Socket.IO client and join channel', async () => {
       // Arrange - Create new session for this test to get channelId
       const authorId = stringToUuid('ws-socketio-user');
       const entityId = stringToUuid('ws-socketio-entity');
@@ -288,7 +328,6 @@ describe('WebSocket Transport Integration', () => {
         }),
       });
       const newSessionData = await newSessionResponse.json();
-      const testSessionId = newSessionData.sessionId;
       const channelId = newSessionData.channelId;
 
       // Connect Socket.IO client
@@ -302,33 +341,8 @@ describe('WebSocket Transport Integration', () => {
         messageServerId: DEFAULT_MESSAGE_SERVER_ID,
       });
 
-      // Setup listener for agent response
-      const agentResponsePromise = clientFixture.waitForEvent<{
-        text: string;
-        senderId: string;
-      }>('messageBroadcast');
-
-      // Act - Send message via HTTP with websocket transport
-      const response = await fetch(`${baseUrl}/api/messaging/sessions/${testSessionId}/messages`, {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({
-          content: 'Hello, waiting for Socket.IO response',
-          author_id: authorId,
-          transport: 'websocket',
-        }),
-      });
-
-      // Assert HTTP response is immediate
-      expect(response.status).toBe(201);
-      const httpData = await response.json();
-      expect(httpData.success).toBe(true);
-      expect(httpData.agentResponse).toBeUndefined();
-
-      // Assert agent response comes via Socket.IO
-      const socketResponse = await agentResponsePromise;
-      expect(socketResponse).toBeDefined();
-      expect(socketResponse.text).toBeDefined();
+      // Assert client is connected and joined
+      expect(clientFixture.isConnected()).toBe(true);
     });
   });
 });