Merge branch 'main' into ochafik/add-elicitation-tool

src/modules/mcp/services/redisTransport.integration.test.ts

@@ -0,0 +1,291 @@
+import { jest } from '@jest/globals';
+import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js';
+import { MockRedisClient, setRedisClient } from '../../shared/redis.js';
+import { 
+  ServerRedisTransport,
+  redisRelayToMcpServer,
+  shutdownSession
+} from './redisTransport.js';
+import { createMcpServer } from './mcp.js';
+import { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
+
+describe('Redis Transport Integration', () => {
+  let mockRedis: MockRedisClient;
+
+  beforeEach(() => {
+    mockRedis = new MockRedisClient();
+    setRedisClient(mockRedis);
+    jest.resetAllMocks();
+  });
+
+  afterEach(() => {
+    mockRedis.clear();
+  });
+
+  describe('MCP Initialization Flow', () => {
+    const sessionId = 'test-init-session';
+
+    it('should relay initialization request from client to server through Redis', async () => {
+      // 1. Start the server listening to Redis
+      const { server, cleanup: serverCleanup } = createMcpServer();
+      const serverTransport = new ServerRedisTransport(sessionId);
+      serverTransport.onclose = serverCleanup;
+      await server.connect(serverTransport);
+
+      // 2. Create a mock client transport (simulating the streamable HTTP client side)
+      const mockClientTransport: Transport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+
+      // 3. Set up the Redis relay (this is what happens in the HTTP handler)
+      const cleanup = await redisRelayToMcpServer(sessionId, mockClientTransport);
+
+      // Track messages received by server
+      const serverReceivedMessages: JSONRPCMessage[] = [];
+      const originalServerOnMessage = serverTransport.onmessage;
+      serverTransport.onmessage = (message, extra) => {
+        serverReceivedMessages.push(message);
+        originalServerOnMessage?.(message, extra);
+      };
+
+      // 4. Simulate client sending initialization request
+      const initMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'init-1',
+        method: 'initialize',
+        params: {
+          protocolVersion: '2024-11-05',
+          capabilities: {},
+          clientInfo: {
+            name: 'test-client',
+            version: '1.0.0'
+          }
+        }
+      };
+
+      // Trigger the client transport onmessage (simulates HTTP request)
+      mockClientTransport.onmessage?.(initMessage);
+
+      // Wait for message to be relayed through Redis
+      await new Promise(resolve => setTimeout(resolve, 50));
+
+      // 5. Verify server received the init message
+      expect(serverReceivedMessages).toHaveLength(1);
+      expect(serverReceivedMessages[0]).toMatchObject({
+        jsonrpc: '2.0',
+        id: 'init-1',
+        method: 'initialize'
+      });
+
+      // 6. Simulate server responding (this should get relayed back to client)
+      const initResponse: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'init-1',
+        result: {
+          protocolVersion: '2024-11-05',
+          capabilities: {
+            tools: {},
+            prompts: {},
+            resources: {}
+          },
+          serverInfo: {
+            name: 'example-server',
+            version: '1.0.0'
+          }
+        }
+      };
+
+      await serverTransport.send(initResponse, { relatedRequestId: 'init-1' });
+
+      // Wait for response to be relayed back
+      await new Promise(resolve => setTimeout(resolve, 50));
+
+      // 7. Verify client transport received the response
+      expect(mockClientTransport.send).toHaveBeenCalledWith(
+        expect.objectContaining({
+          jsonrpc: '2.0',
+          id: 'init-1',
+          result: expect.objectContaining({
+            protocolVersion: '2024-11-05',
+            serverInfo: expect.objectContaining({
+              name: 'example-server'
+            })
+          })
+        }),
+        { relatedRequestId: 'init-1' }
+      );
+
+      // Cleanup
+      await cleanup();
+      await shutdownSession(sessionId);
+      serverCleanup(); // Clean up MCP server intervals
+      
+      // Ensure server transport is closed
+      await serverTransport.close();
+      
+      await new Promise(resolve => setTimeout(resolve, 10));
+    });
+
+    it('should handle tools/list request through Redis relay', async () => {
+      // Set up server and mock client
+      const { server, cleanup: serverCleanup } = createMcpServer();
+      const serverTransport = new ServerRedisTransport(sessionId);
+      serverTransport.onclose = serverCleanup;
+      await server.connect(serverTransport);
+      
+      const mockClientTransport: Transport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+
+      const cleanup = await redisRelayToMcpServer(sessionId, mockClientTransport);
+
+      // Send tools/list request
+      const toolsListMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'tools-1',
+        method: 'tools/list',
+        params: {}
+      };
+
+      mockClientTransport.onmessage?.(toolsListMessage);
+
+      // Wait for processing and response
+      await new Promise(resolve => setTimeout(resolve, 100));
+
+      // Verify client received a response with tools
+      expect(mockClientTransport.send).toHaveBeenCalledWith(
+        expect.objectContaining({
+          jsonrpc: '2.0',
+          id: 'tools-1',
+          result: expect.objectContaining({
+            tools: expect.any(Array)
+          })
+        }),
+        undefined
+      );
+
+      // Cleanup
+      await cleanup();
+      await shutdownSession(sessionId);
+      serverCleanup(); // Clean up MCP server intervals
+      
+      // Ensure server transport is closed
+      await serverTransport.close();
+      
+      await new Promise(resolve => setTimeout(resolve, 10));
+    });
+
+    it('should handle notifications through Redis relay', async () => {
+      // Set up server and mock client
+      const { server, cleanup: serverCleanup } = createMcpServer();
+      const serverTransport = new ServerRedisTransport(sessionId);
+      serverTransport.onclose = serverCleanup;
+      await server.connect(serverTransport);
+      
+      const mockClientTransport: Transport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+
+      const cleanup = await redisRelayToMcpServer(sessionId, mockClientTransport);
+
+      // Set up notification subscription manually since notifications don't have an id
+      const notificationChannel = `mcp:shttp:toclient:${sessionId}:__GET_stream`;
+      const notificationCleanup = await mockRedis.createSubscription(notificationChannel, async (redisMessageJson) => {
+        const redisMessage = JSON.parse(redisMessageJson);
+        if (redisMessage.type === 'mcp') {
+          await mockClientTransport.send(redisMessage.message, redisMessage.options);
+        }
+      }, (error) => {
+        mockClientTransport.onerror?.(error);
+      });
+
+      // Send a notification from server (notifications don't have an id)
+      const notification: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        method: 'notifications/message',
+        params: {
+          level: 'info',
+          logger: 'test',
+          data: 'Test notification'
+        }
+      };
+
+      await serverTransport.send(notification);
+
+      // Wait for notification to be delivered
+      await new Promise(resolve => setTimeout(resolve, 50));
+
+      // Verify client received the notification
+      expect(mockClientTransport.send).toHaveBeenCalledWith(
+        expect.objectContaining({
+          jsonrpc: '2.0',
+          method: 'notifications/message',
+          params: expect.objectContaining({
+            level: 'info',
+            data: 'Test notification'
+          })
+        }),
+        undefined
+      );
+
+      // Cleanup notification subscription
+      await notificationCleanup();
+
+      // Cleanup
+      await cleanup();
+      await shutdownSession(sessionId);
+      serverCleanup(); // Clean up MCP server intervals
+      
+      // Ensure server transport is closed
+      await serverTransport.close();
+      
+      await new Promise(resolve => setTimeout(resolve, 10));
+    });
+
+    it('should not create response subscriptions for notifications', async () => {
+      const mockClientTransport: Transport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+
+      const cleanup = await redisRelayToMcpServer(sessionId, mockClientTransport);
+
+      // Send a notification (no id field)
+      const notification: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        method: 'notifications/initialized',
+        params: {}
+      };
+
+      mockClientTransport.onmessage?.(notification);
+
+      // Wait for processing
+      await new Promise(resolve => setTimeout(resolve, 50));
+
+      // Should not create any response channel subscriptions for notifications
+      // (we can't easily test this directly, but we can ensure no errors occur)
+      expect(mockClientTransport.send).not.toHaveBeenCalled();
+
+      await cleanup();
+    });
+  });
+});

src/modules/mcp/services/redisTransport.test.ts

@@ -0,0 +1,536 @@
+import { jest } from '@jest/globals';
+import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js';
+import { MockRedisClient, setRedisClient } from '../../shared/redis.js';
+import { 
+  ServerRedisTransport, 
+  redisRelayToMcpServer,
+  isLive,
+  shutdownSession,
+  setSessionOwner,
+  getSessionOwner,
+  validateSessionOwnership,
+  isSessionOwnedBy
+} from './redisTransport.js';
+import { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
+
+describe('Redis Transport', () => {
+  let mockRedis: MockRedisClient;
+
+  beforeEach(() => {
+    mockRedis = new MockRedisClient();
+    setRedisClient(mockRedis);
+    jest.resetAllMocks();
+  });
+
+  afterEach(() => {
+    // Clear all Redis data and subscriptions
+    mockRedis.clear();
+  });
+
+  describe('ServerRedisTransport', () => {
+    let transport: ServerRedisTransport;
+    const sessionId = 'test-session-123';
+
+    beforeEach(() => {
+      transport = new ServerRedisTransport(sessionId);
+    });
+
+    afterEach(async () => {
+      if (transport) {
+        await transport.close();
+      }
+    });
+
+    it('should create transport with session ID', () => {
+      expect(transport).toBeInstanceOf(ServerRedisTransport);
+    });
+
+    it('should send response messages to request-specific channels', async () => {
+      const responseMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 123,
+        result: { data: 'test response' }
+      };
+
+      const mockSubscriber = jest.fn();
+      await mockRedis.createSubscription(
+        `mcp:shttp:toclient:${sessionId}:123`, 
+        mockSubscriber, 
+        jest.fn()
+      );
+
+      await transport.send(responseMessage, { relatedRequestId: 123 });
+
+      expect(mockSubscriber).toHaveBeenCalledWith(
+        JSON.stringify({
+          type: 'mcp',
+          message: responseMessage,
+          options: { relatedRequestId: 123 }
+        })
+      );
+    });
+
+    it('should send notification messages to notification channel', async () => {
+      const notificationMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        method: 'notifications/message',
+        params: { message: 'test notification' }
+      };
+
+      const mockSubscriber = jest.fn();
+      await mockRedis.createSubscription(
+        `mcp:shttp:toclient:${sessionId}:__GET_stream`, 
+        mockSubscriber, 
+        jest.fn()
+      );
+
+      await transport.send(notificationMessage);
+
+      expect(mockSubscriber).toHaveBeenCalledWith(
+        JSON.stringify({
+          type: 'mcp',
+          message: notificationMessage,
+          options: undefined
+        })
+      );
+    });
+
+    it('should handle close gracefully', async () => {
+      const onCloseMock = jest.fn();
+      transport.onclose = onCloseMock;
+
+      await transport.close();
+
+      expect(onCloseMock).toHaveBeenCalled();
+    });
+
+    it('should respond to shutdown control messages', async () => {
+      await transport.start();
+      
+      const onCloseMock = jest.fn();
+      transport.onclose = onCloseMock;
+
+      // Send a shutdown control message
+      await shutdownSession(sessionId);
+
+      // Wait for async processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      expect(onCloseMock).toHaveBeenCalled();
+    });
+
+    it('should receive MCP messages from clients and call onmessage', async () => {
+      const onMessageMock = jest.fn();
+      transport.onmessage = onMessageMock;
+
+      await transport.start();
+
+      // Simulate client sending a message to server
+      const clientMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'test-req',
+        method: 'tools/list',
+        params: {}
+      };
+
+      await mockRedis.publish(
+        `mcp:shttp:toserver:${sessionId}`,
+        JSON.stringify({
+          type: 'mcp',
+          message: clientMessage,
+          extra: { authInfo: { token: 'test-token', clientId: 'test-client', scopes: [] } }
+        })
+      );
+
+      // Wait for async processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      expect(onMessageMock).toHaveBeenCalledWith(
+        clientMessage,
+        { authInfo: { token: 'test-token', clientId: 'test-client', scopes: [] } }
+      );
+
+      await transport.close();
+    });
+  });
+
+
+  describe('redisRelayToMcpServer', () => {
+    let mockTransport: Transport;
+    const sessionId = 'test-session-456';
+
+    beforeEach(() => {
+      mockTransport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+    });
+
+    it('should set up message relay from transport to server', async () => {
+      const cleanup = await redisRelayToMcpServer(sessionId, mockTransport);
+
+      // Simulate a message from the transport
+      const requestMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'req-123',
+        method: 'tools/list',
+        params: {}
+      };
+
+      // Trigger the onmessage handler
+      mockTransport.onmessage?.(requestMessage, { authInfo: { token: 'test-token', clientId: 'test-client', scopes: [] } });
+
+      // Wait a bit for async processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Check that message was published to server channel
+      const serverSubscriber = jest.fn();
+      await mockRedis.createSubscription(
+        `mcp:shttp:toserver:${sessionId}`,
+        serverSubscriber,
+        jest.fn()
+      );
+
+      // The message should have been published
+      expect(mockRedis.numsub(`mcp:shttp:toserver:${sessionId}`)).resolves.toBe(1);
+
+      await cleanup();
+    });
+
+    it('should subscribe to response channel for request messages', async () => {
+      const cleanup = await redisRelayToMcpServer(sessionId, mockTransport);
+
+      const requestMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'req-456',
+        method: 'tools/call',
+        params: { name: 'echo', arguments: { text: 'hello' } }
+      };
+
+      // Trigger the onmessage handler
+      mockTransport.onmessage?.(requestMessage, { authInfo: { token: 'test-token', clientId: 'test-client', scopes: [] } });
+
+      // Wait for subscription setup
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Now simulate a response from the server
+      const responseMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'req-456',
+        result: { content: [{ type: 'text', text: 'hello' }] }
+      };
+
+      await mockRedis.publish(
+        `mcp:shttp:toclient:${sessionId}:req-456`,
+        JSON.stringify({
+          type: 'mcp',
+          message: responseMessage,
+          options: undefined
+        })
+      );
+
+      // Check that the response was sent back to the transport
+      expect(mockTransport.send).toHaveBeenCalledWith(responseMessage, undefined);
+
+      await cleanup();
+    });
+
+    it('should not subscribe for notification messages (no id)', async () => {
+      const cleanup = await redisRelayToMcpServer(sessionId, mockTransport);
+
+      const notificationMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        method: 'notifications/message',
+        params: { message: 'test' }
+      };
+
+      // Trigger the onmessage handler
+      mockTransport.onmessage?.(notificationMessage);
+
+      // Wait a bit
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Should not create any response channel subscriptions
+      expect(await mockRedis.numsub(`mcp:shttp:toclient:${sessionId}:undefined`)).toBe(0);
+
+      await cleanup();
+    });
+  });
+
+  describe('isLive', () => {
+    const sessionId = 'test-session-789';
+
+    it('should return true when session has active subscribers', async () => {
+      // Create a subscription to the server channel
+      await mockRedis.createSubscription(
+        `mcp:shttp:toserver:${sessionId}`,
+        jest.fn(),
+        jest.fn()
+      );
+
+      expect(await isLive(sessionId)).toBe(true);
+    });
+
+    it('should return false when session has no subscribers', async () => {
+      expect(await isLive(sessionId)).toBe(false);
+    });
+  });
+
+  describe('Session Ownership', () => {
+    const sessionId = 'test-session-ownership';
+    const userId = 'test-user-123';
+
+    it('should set and get session owner', async () => {
+      await setSessionOwner(sessionId, userId);
+      const owner = await getSessionOwner(sessionId);
+      expect(owner).toBe(userId);
+    });
+
+    it('should validate session ownership correctly', async () => {
+      await setSessionOwner(sessionId, userId);
+      
+      expect(await validateSessionOwnership(sessionId, userId)).toBe(true);
+      expect(await validateSessionOwnership(sessionId, 'different-user')).toBe(false);
+    });
+
+    it('should check if session is owned by user including liveness', async () => {
+      // Session not live yet
+      expect(await isSessionOwnedBy(sessionId, userId)).toBe(false);
+      
+      // Make session live
+      await mockRedis.createSubscription(
+        `mcp:shttp:toserver:${sessionId}`,
+        jest.fn(),
+        jest.fn()
+      );
+      
+      // Still false because no owner set
+      expect(await isSessionOwnedBy(sessionId, userId)).toBe(false);
+      
+      // Set owner
+      await setSessionOwner(sessionId, userId);
+      
+      // Now should be true
+      expect(await isSessionOwnedBy(sessionId, userId)).toBe(true);
+      
+      // False for different user
+      expect(await isSessionOwnedBy(sessionId, 'different-user')).toBe(false);
+    });
+  });
+
+  describe('Integration: Redis message flow', () => {
+    const sessionId = 'integration-test-session';
+
+    it('should relay messages between client and server through Redis', async () => {
+      // Set up client-side transport simulation
+      const clientTransport: Transport = {
+        onmessage: undefined,
+        onclose: undefined,
+        onerror: undefined,
+        send: jest.fn(() => Promise.resolve()),
+        close: jest.fn(() => Promise.resolve()),
+        start: jest.fn(() => Promise.resolve())
+      };
+
+      const cleanup = await redisRelayToMcpServer(sessionId, clientTransport);
+
+      // Client sends a request
+      const listToolsRequest: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'integration-req-1',
+        method: 'tools/list',
+        params: {}
+      };
+
+      // Set up subscription to simulate server receiving the message
+      const serverSubscriber = jest.fn();
+      await mockRedis.createSubscription(
+        `mcp:shttp:toserver:${sessionId}`,
+        serverSubscriber,
+        jest.fn()
+      );
+
+      // Simulate client sending request
+      clientTransport.onmessage?.(listToolsRequest);
+
+      // Wait for async processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Verify the message was published to server channel
+      expect(serverSubscriber).toHaveBeenCalledWith(
+        JSON.stringify({
+          type: 'mcp',
+          message: listToolsRequest,
+          extra: undefined,
+          options: undefined
+        })
+      );
+
+      // Simulate server sending response
+      const serverResponse: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 'integration-req-1',
+        result: { tools: [{ name: 'echo', description: 'Echo tool' }] }
+      };
+
+      await mockRedis.publish(
+        `mcp:shttp:toclient:${sessionId}:integration-req-1`,
+        JSON.stringify({
+          type: 'mcp',
+          message: serverResponse,
+          options: undefined
+        })
+      );
+
+      // Wait for response processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Verify the response was sent back to client
+      expect(clientTransport.send).toHaveBeenCalledWith(serverResponse, undefined);
+
+      await cleanup();
+    });
+  });
+
+  describe('Control Messages', () => {
+    const sessionId = 'test-control-session';
+
+    it('should send shutdown control messages', async () => {
+      const controlSubscriber = jest.fn();
+      await mockRedis.createSubscription(
+        `mcp:control:${sessionId}`,
+        controlSubscriber,
+        jest.fn()
+      );
+
+      await shutdownSession(sessionId);
+
+      const callArgs = controlSubscriber.mock.calls[0][0] as string;
+      const message = JSON.parse(callArgs);
+      
+      expect(message.type).toBe('control');
+      expect(message.action).toBe('SHUTDOWN');
+      expect(typeof message.timestamp).toBe('number');
+    });
+
+    it('should properly shutdown server transport via control message', async () => {
+      const transport = new ServerRedisTransport(sessionId);
+      const onCloseMock = jest.fn();
+      transport.onclose = onCloseMock;
+
+      await transport.start();
+
+      // Send shutdown signal
+      await shutdownSession(sessionId);
+      
+      // Wait for async processing
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      expect(onCloseMock).toHaveBeenCalled();
+    });
+  });
+
+  describe('Inactivity Timeout', () => {
+    const sessionId = 'test-inactivity-session';
+
+    beforeEach(() => {
+      jest.useFakeTimers({ doNotFake: ['setImmediate', 'nextTick'] });
+    });
+
+    afterEach(() => {
+      jest.useRealTimers();
+    });
+
+    it('should shutdown session after 5 minutes of inactivity', async () => {
+      const transport = new ServerRedisTransport(sessionId);
+      const shutdownSpy = jest.spyOn(mockRedis, 'publish');
+      
+      await transport.start();
+
+      // Fast-forward time by 5 minutes
+      jest.advanceTimersByTime(5 * 60 * 1000);
+
+      // Should have published shutdown control message
+      expect(shutdownSpy).toHaveBeenCalledWith(
+        `mcp:control:${sessionId}`,
+        expect.stringContaining('"action":"SHUTDOWN"')
+      );
+
+      await transport.close();
+    });
+
+    it('should reset timeout when message is received', async () => {
+      const transport = new ServerRedisTransport(sessionId);
+      const onMessageMock = jest.fn();
+      transport.onmessage = onMessageMock;
+      
+      await transport.start();
+
+      // Fast-forward 4 minutes
+      jest.advanceTimersByTime(4 * 60 * 1000);
+
+      // Manually publish a message to trigger the subscription handler
+      const testMessage = { jsonrpc: '2.0', method: 'ping' };
+      await mockRedis.publish(
+        `mcp:shttp:toserver:${sessionId}`,
+        JSON.stringify({
+          type: 'mcp',
+          message: testMessage
+        })
+      );
+
+      // Wait for message to be processed
+      await new Promise(resolve => setImmediate(resolve));
+
+      // Verify message was received
+      expect(onMessageMock).toHaveBeenCalledWith(testMessage, undefined);
+
+      // Clear the publish spy to check only future calls
+      const shutdownSpy = jest.spyOn(mockRedis, 'publish');
+      shutdownSpy.mockClear();
+
+      // Fast-forward 4 more minutes (total 8, but only 4 since last message)
+      jest.advanceTimersByTime(4 * 60 * 1000);
+
+      // Should not have shutdown yet
+      expect(shutdownSpy).not.toHaveBeenCalledWith(
+        `mcp:control:${sessionId}`,
+        expect.stringContaining('"action":"SHUTDOWN"')
+      );
+
+      // Fast-forward 2 more minutes to exceed timeout
+      jest.advanceTimersByTime(2 * 60 * 1000);
+
+      // Now should have shutdown
+      expect(shutdownSpy).toHaveBeenCalledWith(
+        `mcp:control:${sessionId}`,
+        expect.stringContaining('"action":"SHUTDOWN"')
+      );
+
+      await transport.close();
+    }, 10000);
+
+    it('should clear timeout on close', async () => {
+      const transport = new ServerRedisTransport(sessionId);
+      const shutdownSpy = jest.spyOn(mockRedis, 'publish');
+      
+      await transport.start();
+
+      // Close transport before timeout
+      await transport.close();
+
+      // Fast-forward past timeout
+      jest.advanceTimersByTime(10 * 60 * 1000);
+
+      // Should not have triggered shutdown
+      expect(shutdownSpy).not.toHaveBeenCalledWith(
+        `mcp:control:${sessionId}`,
+        expect.stringContaining('"action":"SHUTDOWN"')
+      );
+    });
+  });
+});

src/modules/mcp/services/redisTransport.ts

@@ -0,0 +1,351 @@
+import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
+import { redisClient } from "../../shared/redis.js";
+import { Transport, TransportSendOptions } from "@modelcontextprotocol/sdk/shared/transport.js";
+import { AuthInfo } from "@modelcontextprotocol/sdk/server/auth/types.js";
+import { JSONRPCMessage, MessageExtraInfo } from "@modelcontextprotocol/sdk/types.js";
+import { logger } from "../../shared/logger.js";
+
+let redisTransportCounter = 0;
+const notificationStreamId = "__GET_stream";
+
+// Message types for Redis transport
+type RedisMessage = 
+  | {
+      type: 'mcp';
+      message: JSONRPCMessage;
+      extra?: MessageExtraInfo;
+      options?: TransportSendOptions;
+    }
+  | {
+      type: 'control';
+      action: 'SHUTDOWN' | 'PING' | 'STATUS';
+      timestamp?: number;
+    };
+
+function sendToMcpServer(sessionId: string, message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }, options?: TransportSendOptions): Promise<void> {
+  const toServerChannel = getToServerChannel(sessionId);
+  
+  logger.debug('Sending message to MCP server via Redis', {
+    sessionId,
+    channel: toServerChannel,
+    method: ('method' in message ? message.method : undefined),
+    id: ('id' in message ? message.id : undefined)
+  });
+  
+  const redisMessage: RedisMessage = { type: 'mcp', message, extra, options };
+  return redisClient.publish(toServerChannel, JSON.stringify(redisMessage));
+}
+
+function getToServerChannel(sessionId: string): string {
+  return `mcp:shttp:toserver:${sessionId}`;
+}
+
+function getToClientChannel(sessionId: string, relatedRequestId: string): string {
+  return `mcp:shttp:toclient:${sessionId}:${relatedRequestId}`;
+}
+
+function getControlChannel(sessionId: string): string {
+  return `mcp:control:${sessionId}`;
+}
+
+function sendControlMessage(sessionId: string, action: 'SHUTDOWN' | 'PING' | 'STATUS'): Promise<void> {
+  const controlChannel = getControlChannel(sessionId);
+  const redisMessage: RedisMessage = {
+    type: 'control',
+    action,
+    timestamp: Date.now()
+  };
+  return redisClient.publish(controlChannel, JSON.stringify(redisMessage));
+}
+
+export async function shutdownSession(sessionId: string): Promise<void> {
+  logger.info('Sending shutdown control message', { sessionId });
+  return sendControlMessage(sessionId, 'SHUTDOWN');
+}
+
+export async function isLive(sessionId: string): Promise<boolean> {
+  // Check if the session is live by checking if the key exists in Redis
+  const numSubs = await redisClient.numsub(getToServerChannel(sessionId));
+  return numSubs > 0;
+}
+
+export async function setSessionOwner(sessionId: string, userId: string): Promise<void> {
+  logger.debug('Setting session owner', { sessionId, userId });
+  await redisClient.set(`session:${sessionId}:owner`, userId);
+}
+
+export async function getSessionOwner(sessionId: string): Promise<string | null> {
+  return await redisClient.get(`session:${sessionId}:owner`);
+}
+
+export async function validateSessionOwnership(sessionId: string, userId: string): Promise<boolean> {
+  const owner = await getSessionOwner(sessionId);
+  return owner === userId;
+}
+
+export async function isSessionOwnedBy(sessionId: string, userId: string): Promise<boolean> {
+  const isLiveSession = await isLive(sessionId);
+  if (!isLiveSession) {
+    logger.debug('Session not live', { sessionId });
+    return false;
+  }
+  const isOwned = await validateSessionOwnership(sessionId, userId);
+  logger.debug('Session ownership check', { sessionId, userId, isOwned });
+  return isOwned;
+}
+
+
+export async function redisRelayToMcpServer(sessionId: string, transport: Transport, isGetRequest: boolean = false): Promise<() => Promise<void>> {
+  logger.debug('Setting up Redis relay to MCP server', {
+    sessionId,
+    isGetRequest
+  });
+  
+  let redisCleanup: (() => Promise<void>) | undefined = undefined;
+  const cleanup = async () => {
+    // TODO: solve race conditions where we call cleanup while the subscription is being created / before it is created
+    if (redisCleanup) {
+      logger.debug('Cleaning up Redis relay', { sessionId });
+      await redisCleanup();
+    }
+  }
+
+  const subscribe = async (requestId: string) => {
+    const toClientChannel = getToClientChannel(sessionId, requestId);
+    
+    logger.debug('Subscribing to client channel', {
+      sessionId,
+      requestId,
+      channel: toClientChannel
+    });
+
+    redisCleanup = await redisClient.createSubscription(toClientChannel, async (redisMessageJson) => {
+      const redisMessage = JSON.parse(redisMessageJson) as RedisMessage;
+      if (redisMessage.type === 'mcp') {
+        logger.debug('Relaying message from Redis to client', {
+          sessionId,
+          requestId,
+          method: ('method' in redisMessage.message ? redisMessage.message.method : undefined)
+        });
+        await transport.send(redisMessage.message, redisMessage.options);
+      }
+    }, (error) => {
+      logger.error('Error in Redis relay subscription', error, {
+        sessionId,
+        channel: toClientChannel
+      });
+      transport.onerror?.(error);
+    });
+  }
+
+  if (isGetRequest) {
+    await subscribe(notificationStreamId);
+  } else {
+    const messagePromise = new Promise<JSONRPCMessage>((resolve) => {
+      transport.onmessage = async (message, extra) => {
+        // First, set up response subscription if needed
+        if ("id" in message) {
+          logger.debug('Setting up response subscription', {
+            sessionId,
+            messageId: message.id,
+            method: ('method' in message ? message.method : undefined)
+          });
+          await subscribe(message.id.toString());
+        }
+        // Now send the message to the MCP server
+        await sendToMcpServer(sessionId, message, extra);
+        resolve(message);
+      }
+    });
+  
+    messagePromise.catch((error) => {
+      transport.onerror?.(error);
+      cleanup();
+    });   
+  }
+  return cleanup;
+}
+
+
+// New Redis transport for server->client messages using request-id based channels
+export class ServerRedisTransport implements Transport {
+  private counter: number;
+  private _sessionId: string;
+  private controlCleanup?: (() => Promise<void>);
+  private serverCleanup?: (() => Promise<void>);
+  private shouldShutdown = false;
+  private inactivityTimeout?: NodeJS.Timeout;
+  private readonly INACTIVITY_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes
+
+  onclose?: (() => void) | undefined;
+  onerror?: ((error: Error) => void) | undefined;
+  onmessage?: ((message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }) => void) | undefined;
+
+  constructor(sessionId: string) {
+    this.counter = redisTransportCounter++;
+    this._sessionId = sessionId;
+  }
+
+  private resetInactivityTimer(): void {
+    // Clear existing timeout if any
+    if (this.inactivityTimeout) {
+      clearTimeout(this.inactivityTimeout);
+    }
+
+    // Set new timeout
+    this.inactivityTimeout = setTimeout(() => {
+      logger.info('Session timed out due to inactivity', {
+        sessionId: this._sessionId,
+        timeoutMs: this.INACTIVITY_TIMEOUT_MS
+      });
+      void shutdownSession(this._sessionId);
+    }, this.INACTIVITY_TIMEOUT_MS);
+  }
+
+  private clearInactivityTimer(): void {
+    if (this.inactivityTimeout) {
+      clearTimeout(this.inactivityTimeout);
+      this.inactivityTimeout = undefined;
+    }
+  }
+
+  async start(): Promise<void> {
+    logger.info('Starting ServerRedisTransport', {
+      sessionId: this._sessionId,
+      inactivityTimeoutMs: this.INACTIVITY_TIMEOUT_MS
+    });
+    
+    // Start inactivity timer
+    this.resetInactivityTimer();
+    
+    // Subscribe to MCP messages from clients
+    const serverChannel = getToServerChannel(this._sessionId);
+    logger.debug('Subscribing to server channel', {
+      sessionId: this._sessionId,
+      channel: serverChannel
+    });
+    
+    this.serverCleanup = await redisClient.createSubscription(
+      serverChannel,
+      (messageJson) => {
+        const redisMessage = JSON.parse(messageJson) as RedisMessage;
+        if (redisMessage.type === 'mcp') {
+          // Reset inactivity timer on each message from client
+          this.resetInactivityTimer();
+          
+          logger.debug('Received MCP message from client', {
+            sessionId: this._sessionId,
+            method: ('method' in redisMessage.message ? redisMessage.message.method : undefined),
+            id: ('id' in redisMessage.message ? redisMessage.message.id : undefined)
+          });
+          
+          this.onmessage?.(redisMessage.message, redisMessage.extra);
+        }
+      },
+      (error) => {
+        logger.error('Error in server channel subscription', error, {
+          sessionId: this._sessionId,
+          channel: serverChannel
+        });
+        this.onerror?.(error);
+      }
+    );
+    
+    // Subscribe to control messages for shutdown
+    const controlChannel = getControlChannel(this._sessionId);
+    logger.debug('Subscribing to control channel', {
+      sessionId: this._sessionId,
+      channel: controlChannel
+    });
+    
+    this.controlCleanup = await redisClient.createSubscription(
+      controlChannel,
+      (messageJson) => {
+        const redisMessage = JSON.parse(messageJson) as RedisMessage;
+        if (redisMessage.type === 'control') {
+          logger.info('Received control message', {
+            sessionId: this._sessionId,
+            action: redisMessage.action
+          });
+          
+          if (redisMessage.action === 'SHUTDOWN') {
+            logger.info('Shutting down transport due to control message', {
+              sessionId: this._sessionId
+            });
+            this.shouldShutdown = true;
+            this.close();
+          }
+        }
+      },
+      (error) => {
+        logger.error('Error in control channel subscription', error, {
+          sessionId: this._sessionId,
+          channel: controlChannel
+        });
+        this.onerror?.(error);
+      }
+    );
+    
+  }
+
+  async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise<void> {
+    const relatedRequestId = options?.relatedRequestId?.toString() ?? ("id" in message ? message.id?.toString() : notificationStreamId);
+    const channel = getToClientChannel(this._sessionId, relatedRequestId)
+
+    logger.debug('Sending message to client', {
+      sessionId: this._sessionId,
+      channel,
+      method: ('method' in message ? message.method : undefined),
+      id: ('id' in message ? message.id : undefined),
+      relatedRequestId
+    });
+
+    const redisMessage: RedisMessage = { type: 'mcp', message, options };
+    const messageStr = JSON.stringify(redisMessage);
+    await redisClient.publish(channel, messageStr);
+  }
+
+  async close(): Promise<void> {
+    logger.info('Closing ServerRedisTransport', {
+      sessionId: this._sessionId,
+      wasShutdown: this.shouldShutdown
+    });
+    
+    // Clear inactivity timer
+    this.clearInactivityTimer();
+    
+    // Clean up server message subscription
+    if (this.serverCleanup) {
+      await this.serverCleanup();
+      this.serverCleanup = undefined;
+    }
+    
+    // Clean up control message subscription
+    if (this.controlCleanup) {
+      await this.controlCleanup();
+      this.controlCleanup = undefined;
+    }
+    
+    this.onclose?.();
+  }
+}
+
+export async function getShttpTransport(sessionId: string, onsessionclosed: (sessionId: string) => void | Promise<void>, isGetRequest: boolean = false): Promise<StreamableHTTPServerTransport> {
+  logger.debug('Getting StreamableHTTPServerTransport for existing session', {
+    sessionId,
+    isGetRequest
+  });
+  
+  // Giving undefined here and setting the sessionId means the 
+  // transport wont try to create a new session.
+  const shttpTransport = new StreamableHTTPServerTransport({
+    sessionIdGenerator: undefined,
+    onsessionclosed,
+  })
+  shttpTransport.sessionId = sessionId;
+
+  // Use the new request-id based relay approach
+  const cleanup = await redisRelayToMcpServer(sessionId, shttpTransport, isGetRequest);
+  shttpTransport.onclose = cleanup; 
+  return shttpTransport;
+}