Streaming Responses (SSE & Socket.IO)

curl -s -N -X POST "https://pria.praxislxp.com/api/ai/personal-stream/qanda-stream" \
  -H "Content-Type: application/json" \
  -H "x-access-token: your_api_access_token" \
  -d '{
    "inputs": ["What is the capital of Germany?"],
    "requestArgs": {
      "institutionPublicId": "e455529a-4f51-479e-94fc-bbebb41d19a1",
      "assistantId": "6970576f40219f56e41ec00a",
      "selectedCourse": {
        "course_id": 1769040847660,
        "course_name": "free Wednesday, Jan 21, 2026"
      },
      "ragOnly": false
    }
  }'

// sseClient.js - Complete Node.js SSE streaming client
const https = require('https');

/**
 * Stream a chat request using HTTP SSE
 * @param {string} token - JWT authentication token
 * @param {string} prompt - User's question or prompt
 * @param {object} options - Optional configuration
 * @param {string} [options.institutionPublicId] - Digital twin UUID
 * @param {string} [options.assistantId] - Assistant ObjectId
 * @param {object} [options.selectedCourse] - Conversation context
 * @param {boolean} [options.ragOnly] - RAG-only mode
 * @returns {Promise<object>} - Final response data
 */
async function streamChat(token, prompt, options = {}) {
  const baseUrl = 'https://pria.praxislxp.com';

  const requestBody = JSON.stringify({
    inputs: [prompt],
    requestArgs: {
      institutionPublicId: options.institutionPublicId,
      assistantId: options.assistantId,
      selectedCourse: options.selectedCourse || {},
      ragOnly: options.ragOnly || false,
    }
  });

  return new Promise((resolve, reject) => {
    const req = https.request(
      `${baseUrl}/api/ai/personal-stream/qanda-stream`,
      {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'x-access-token': token
        }
      },
      (res) => {
        if (res.statusCode !== 200) {
          reject(new Error(`HTTP ${res.statusCode}: ${res.statusMessage}`));
          return;
        }

        let buffer = '';
        let finalResponse = null;

        res.on('data', (chunk) => {
          buffer += chunk.toString();

          // Process complete SSE messages
          const lines = buffer.split('\n\n');
          buffer = lines.pop(); // Keep incomplete message in buffer

          for (const line of lines) {
            if (line.startsWith('data: ')) {
              try {
                const data = JSON.parse(line.slice(6));

                switch (data.type) {
                  case 'connected':
                    console.log('Stream connected');
                    break;

                  case 'stream':
                    // Use delta for incremental rendering
                    process.stdout.write(data.delta || '');
                    break;

                  case 'thinking':
                    // Model reasoning — keyed by `id` (round-0, round-1, ...).
                    // Render in a collapsible UI block separate from the answer.
                    if (data.done) {
                      console.log(`\n[Thinking ${data.id} complete: ${(data.text || '').length} chars]`);
                    } else if (data.delta) {
                      // Incremental thinking — accumulate per `data.id`
                      // process.stdout.write(`💡${data.delta}`);
                    }
                    break;

                  case 'tool_call':
                    console.log(`\n[Tool: ${data.name}]`);
                    break;

                  case 'tool_result':
                    console.log(`[Tool ${data.name} completed in ${data.responseDurationMs}ms]`);
                    break;

                  case 'complete':
                    console.log(`\n\nComplete - Tokens: ${data.usage}, Model: ${data.model}`);
                    finalResponse = data;
                    break;

                  case 'error':
                    console.error('\nError:', data.error);
                    reject(new Error(data.error?.message || 'Stream error'));
                    return;

                  case 'done':
                    resolve(finalResponse);
                    return;
                }
              } catch (e) {
                // Skip malformed JSON
              }
            }
          }
        });

        res.on('end', () => {
          resolve(finalResponse);
        });

        res.on('error', reject);
      }
    );

    req.on('error', reject);
    req.write(requestBody);
    req.end();
  });
}

// Example usage
(async () => {
  const token = 'your_jwt_token_here';

  try {
    const result = await streamChat(token, 'What is machine learning?', {
      institutionPublicId: 'f831501f-b645-481a-9cbb-331509aaf8c1',
      assistantId: '6856fa89cbafcff8d98680f5',
      selectedCourse: {
        course_id: 1750532703472,
        course_name: 'AI Fundamentals'
      }
    });

    console.log('\nFinal result:', result);
  } catch (error) {
    console.error('Error:', error.message);
  }
})();

// Modern fetch-based SSE client
async function streamChatFetch(token, prompt, onChunk, requestArgs = {}) {
  const response = await fetch('https://pria.praxislxp.com/api/ai/personal-stream/qanda-stream', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'x-access-token': token
    },
    body: JSON.stringify({
      inputs: [prompt],
      requestArgs
    })
  });

  if (!response.ok) {
    throw new Error(`HTTP ${response.status}: ${response.statusText}`);
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n\n');
    buffer = lines.pop();

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6));

        // Call the chunk handler
        if (onChunk) onChunk(data);

        if (data.type === 'complete') {
          return data; // Return final response
        }
      }
    }
  }
}

// Usage with callback — use delta for incremental rendering
await streamChatFetch(
  token,
  'Explain quantum computing',
  (chunk) => {
    if (chunk.type === 'stream') {
      document.getElementById('output').textContent += chunk.delta;
    } else if (chunk.type === 'tool_call') {
      console.log(`Tool invoked: ${chunk.name}`);
    } else if (chunk.type === 'tool_result') {
      console.log(`Tool ${chunk.name} completed in ${chunk.responseDurationMs}ms`);
    }
  },
  {
    institutionPublicId: 'f831501f-b645-481a-9cbb-331509aaf8c1',
    assistantId: '6856fa89cbafcff8d98680f5',
    selectedCourse: { course_id: 1750532703472, course_name: 'AI Fundamentals' }
  }
);

npm i socket.io-client

// socket.ts (client)
import { io } from 'socket.io-client';

const URL = 'https://pria.praxislxp.com/socket.io'; // Middleware streaming endpoint
localStorage.debug = '*'; // Optional: enable Socket.IO debug logs

// Keep a simple local state holder
const state = {};
state.socket = io(URL, { autoConnect: false });

// Mocked user profile (replace with your authenticated session values)
const user = {
  userId: profile?._id,
  email: profile?.email,
  profileName: `${profile?.fname} ${profile?.lname}`,
  institutionId: profile?.institution?._id,
  institutionName: profile?.institution?.name,
};

// Open the connection and register your user
state.socket.on('connect', () => {
  state.socket.emit('REGISTER', user);
});

// Handle graceful disconnect/unsubscribe
state.socket.on('disconnect', (e) => {
  console.log('SocketContext', 'Disconnect', e);
  state.socket.emit('UNSUBSCRIBE', user);
});

// Optional: initiate connection if autoConnect is false
state.socket.connect();

// Stream chunks arrive while REST Q&A requests are processed
state.socket.on('RECEIVE_STREAM', (payload) => {
  console.log('SocketContext', 'RECEIVE_STREAM', payload);
});

// Error & connectivity diagnostics
state.socket.on('error', (e) => console.log('SocketContext', 'Error', e));
state.socket.on('connect_success', (e) => console.log('SocketContext', 'Connect Success', e));
state.socket.on('connect_timeout', (e) => console.log('SocketContext', 'Connect Timeout', e));
state.socket.on('connect_error', (e) => console.log('SocketContext', 'Connect Error', e));

// Example: POST Q&A request linking REST call to Socket.IO session
const payload = {
  inputs: [userMessage],
  requestArgs: {
    socketId: state.socket.id, // CRITICAL: links the streaming session
    assistantId: currentAssistantId,
    selectedCourse: {
      course_id: currentCourseId,
      course_name: currentCourseName,
    },
  },
};

const response = await fetch('https://pria.praxislxp.com/api/ai/personal/qanda', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    // If your deployment requires header-based auth for this endpoint, include your token here.
    // 'x-access-token': your_api_access_token
  },
  body: JSON.stringify(payload),
});

if (!response.ok) {
  throw new Error(`HTTP error! status: ${response.status}`);
}

const data = await response.json();

// Example wrapper usage
const { sessionId } = useSocketIO();

await priaApi.qanda(
  {
    inputs: [userMessage],
    requestArgs: {
      socketId: sessionId, // CRITICAL: Link to Socket.IO session
      assistantId: currentAssistantId,
      selectedCourse: {
        course_id: currentCourseId,
        course_name: currentCourseName,
      },
    },
  },
  onStreamCallback // handle incremental chunks
);