1. Session persists in GATHERING state. 2. Added inactive sessions purging, but it does not work.
This commit is contained in:
@@ -51,6 +51,7 @@ app.post('/sessions', (req, res) => {
|
||||
responses: new Map(),
|
||||
clients: new Map(),
|
||||
finalResult: null,
|
||||
lastActivity: Date.now(),
|
||||
});
|
||||
console.log(`New session created: ${sessionId}`);
|
||||
res.status(201).json({ sessionId });
|
||||
|
||||
@@ -1,9 +1,18 @@
|
||||
|
||||
|
||||
import express from 'express';
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { sessions, SessionState, broadcastToSession, handleWebSocketMessage } from '../ws'; // Import sessions, SessionState, broadcastToSession, and handleWebSocketMessage from ws/index.ts
|
||||
import { sessions, SessionState, broadcastToSession } from '../ws'; // Import sessions, SessionState, broadcastToSession from ws/index.ts
|
||||
import { LLMService } from '../services/LLMService';
|
||||
import { EncryptionService } from '../services/EncryptionService';
|
||||
|
||||
const router = express.Router();
|
||||
|
||||
// Initialize LLM Service (API key from environment)
|
||||
const llmService = new LLMService(process.env.GEMINI_API_KEY || '');
|
||||
// Initialize Encryption Service
|
||||
const encryptionService = new EncryptionService(process.env.ENCRYPTION_KEY || '');
|
||||
|
||||
router.post('/sessions', (req, res) => {
|
||||
const sessionId = uuidv4();
|
||||
sessions.set(sessionId, {
|
||||
@@ -15,36 +24,81 @@ router.post('/sessions', (req, res) => {
|
||||
responses: new Map(),
|
||||
clients: new Map(),
|
||||
finalResult: null,
|
||||
lastActivity: Date.now(),
|
||||
});
|
||||
res.status(201).json({ sessionId });
|
||||
});
|
||||
|
||||
router.post('/sessions/:sessionId/responses', async (req, res) => {
|
||||
const { sessionId } = req.params;
|
||||
const { userId, wants, accepts, afraidToAsk } = req.body;
|
||||
const { clientId, wants, accepts, noGoes, afraidToAsk } = req.body; // Use clientId instead of userId
|
||||
|
||||
if (!sessions.has(sessionId)) {
|
||||
return res.status(404).json({ message: 'Session not found.' });
|
||||
}
|
||||
|
||||
// Create a dummy WebSocket object for the handleWebSocketMessage function.
|
||||
// This is a workaround to reuse the WebSocket message handling logic.
|
||||
// In a real application, consider a more robust event-driven architecture.
|
||||
const dummyWs: any = {
|
||||
send: (message: string) => console.log('Dummy WS send:', message),
|
||||
readyState: 1, // OPEN
|
||||
};
|
||||
const sessionData = sessions.get(sessionId)!;
|
||||
|
||||
const message = {
|
||||
type: 'SUBMIT_RESPONSE',
|
||||
clientId: userId, // Using userId as clientId for simplicity in this context
|
||||
payload: {
|
||||
response: { wants, accepts, afraidToAsk },
|
||||
},
|
||||
};
|
||||
if (sessionData.state !== SessionState.GATHERING) {
|
||||
return res.status(400).json({ message: `Session is not in GATHERING state. Current state: ${sessionData.state}` });
|
||||
}
|
||||
|
||||
if (sessionData.responses.has(clientId)) {
|
||||
return res.status(400).json({ message: 'You have already submitted a response for this session.' });
|
||||
}
|
||||
|
||||
if ([...wants, ...accepts, ...noGoes].some((desire: string) => desire.length > 500) || afraidToAsk.length > 500) {
|
||||
return res.status(400).json({ message: 'One of your desires or afraidToAsk exceeds the 500 character limit.' });
|
||||
}
|
||||
|
||||
try {
|
||||
await handleWebSocketMessage(dummyWs, sessionId, message);
|
||||
const hasContradictionsGist = await llmService.checkForInnerContradictions({ wants, accepts, noGoes, afraidToAsk });
|
||||
if (hasContradictionsGist) {
|
||||
return res.status(400).json({ message: `Your submission contains inner contradictions: ${hasContradictionsGist} Please resolve them and submit again.` });
|
||||
}
|
||||
|
||||
const encryptedWants = wants.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedAccepts = accepts.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedNoGoes = noGoes.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedAfraidToAsk = encryptionService.encrypt(afraidToAsk);
|
||||
|
||||
sessionData.responses.set(clientId, { wants: encryptedWants, accepts: encryptedAccepts, noGoes: encryptedNoGoes, afraidToAsk: encryptedAfraidToAsk });
|
||||
sessionData.submittedCount++;
|
||||
|
||||
console.log(`Client ${clientId} submitted response via HTTP. Submitted count: ${sessionData.submittedCount}/${sessionData.expectedResponses}`);
|
||||
|
||||
if (sessionData.submittedCount === sessionData.expectedResponses) {
|
||||
sessionData.state = SessionState.HARMONIZING;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
console.log(`Session ${sessionId} moved to HARMONIZING. Triggering LLM analysis.`);
|
||||
|
||||
// Perform LLM analysis asynchronously
|
||||
(async () => {
|
||||
try {
|
||||
const allDecryptedDesires = Array.from(sessionData.responses.values()).map(encryptedResponse => {
|
||||
const decryptedWants = encryptedResponse.wants.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedAccepts = encryptedResponse.accepts.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedNoGoes = encryptedResponse.noGoes.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedAfraidToAsk = encryptionService.decrypt(encryptedResponse.afraidToAsk);
|
||||
return { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: decryptedAfraidToAsk };
|
||||
});
|
||||
const decision = await llmService.analyzeDesires(allDecryptedDesires);
|
||||
|
||||
sessionData.finalResult = decision;
|
||||
sessionData.state = SessionState.FINAL;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
console.log(`Analysis complete for session ${sessionId}. Result:`, decision);
|
||||
} catch (error: any) {
|
||||
console.error(`Error during analysis for session ${sessionId}:`, error.message);
|
||||
sessionData.state = SessionState.ERROR;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
}
|
||||
})();
|
||||
} else {
|
||||
// Only broadcast the latest count if the session is not yet harmonizing
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
}
|
||||
|
||||
res.status(202).json({ message: 'Response submission acknowledged and processed.' });
|
||||
} catch (error: any) {
|
||||
console.error('Error processing response via HTTP route:', error);
|
||||
|
||||
@@ -41,6 +41,7 @@ interface SessionData {
|
||||
responses: Map<string, EncryptedResponseData>; // Stores the submitted desire objects. Map<ClientID, EncryptedResponseData>
|
||||
clients: Map<string, WebSocket>; // Maps the persistent Client ID to their active WebSocket connection object.
|
||||
finalResult: any | null; // The result returned by the LLM.
|
||||
lastActivity: number; // Timestamp of the last activity (e.g., message, client join/leave)
|
||||
}
|
||||
export const sessions = new Map<string, SessionData>();
|
||||
|
||||
@@ -102,9 +103,27 @@ export const broadcastToSession = (sessionId: string, message: any, excludeClien
|
||||
}
|
||||
};
|
||||
|
||||
const SESSION_TIMEOUT_MINUTES = parseInt(process.env.SESSION_TIMEOUT_MINUTES || '1440', 10);
|
||||
const SESSION_TIMEOUT_MS = SESSION_TIMEOUT_MINUTES * 60 * 1000; // Convert minutes to milliseconds
|
||||
|
||||
// Function to clean up inactive sessions
|
||||
const cleanupInactiveSessions = () => {
|
||||
const now = Date.now();
|
||||
for (const [sessionId, sessionData] of sessions.entries()) {
|
||||
if (sessionData.clients.size === 0 && (now - sessionData.lastActivity > SESSION_TIMEOUT_MS)) {
|
||||
sessions.delete(sessionId);
|
||||
logEvent('session_purged_inactive', sessionId);
|
||||
console.log(`Inactive session ${sessionId} purged.`);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
export const createWebSocketServer = (server: any) => {
|
||||
const wss = new WebSocketServer({ server });
|
||||
|
||||
// Schedule periodic cleanup of inactive sessions
|
||||
setInterval(cleanupInactiveSessions, 60 * 60 * 1000); // Run every hour
|
||||
|
||||
wss.on('connection', (ws, req) => {
|
||||
const url = new URL(req.url || '', `http://${req.headers.host}`);
|
||||
const sessionId = url.pathname.split('/').pop();
|
||||
@@ -114,21 +133,9 @@ export const createWebSocketServer = (server: any) => {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!sessions.has(sessionId)) {
|
||||
sessions.set(sessionId, {
|
||||
state: SessionState.SETUP,
|
||||
topic: null,
|
||||
description: null,
|
||||
expectedResponses: 0,
|
||||
submittedCount: 0,
|
||||
responses: new Map<string, any>(),
|
||||
clients: new Map<string, WebSocket>(),
|
||||
finalResult: null,
|
||||
});
|
||||
}
|
||||
const sessionData = sessions.get(sessionId)!;
|
||||
|
||||
console.log(`Client connecting to session: ${sessionId}`);
|
||||
|
||||
let sessionData: SessionData | null = null;
|
||||
|
||||
// Set up a ping interval to keep the connection alive
|
||||
const pingInterval = setInterval(() => {
|
||||
@@ -137,47 +144,45 @@ export const createWebSocketServer = (server: any) => {
|
||||
}
|
||||
}, 30000); // Send ping every 30 seconds
|
||||
|
||||
ws.on('message', async (message) => {
|
||||
const parsedMessage = JSON.parse(message.toString());
|
||||
const { type, clientId, payload } = parsedMessage;
|
||||
|
||||
if (!clientId) {
|
||||
console.error(`Received message without clientId in session ${sessionId}. Type: ${type}`);
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'clientId is required' } }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (!sessionData.clients.has(clientId)) {
|
||||
sessionData.clients.set(clientId, ws);
|
||||
console.log(`Client ${clientId} registered for session: ${sessionId}. Total clients: ${sessionData.clients.size}`);
|
||||
ws.send(JSON.stringify({ type: 'STATE_UPDATE', payload: { session: getSerializableSession(sessionData, clientId) } }));
|
||||
}
|
||||
|
||||
console.log(`Received message from ${clientId} in session ${sessionId}:`, type);
|
||||
await handleWebSocketMessage(ws, sessionId, parsedMessage);
|
||||
});
|
||||
|
||||
ws.on('close', () => {
|
||||
ws.on('message', async (message) => {
|
||||
const parsedMessage = JSON.parse(message.toString());
|
||||
const updatedSessionData = await handleWebSocketMessage(ws, sessionId, parsedMessage);
|
||||
if (updatedSessionData) {
|
||||
sessionData = updatedSessionData;
|
||||
}
|
||||
}); ws.on('close', () => {
|
||||
clearInterval(pingInterval); // Clear the interval when the connection closes
|
||||
let disconnectedClientId: string | null = null;
|
||||
for (const [clientId, clientWs] of sessionData.clients.entries()) {
|
||||
if (clientWs === ws) {
|
||||
disconnectedClientId = clientId;
|
||||
break;
|
||||
const currentSessionData = sessions.get(sessionId); // Retrieve the latest sessionData
|
||||
if (currentSessionData) { // Check if sessionData is not null
|
||||
for (const [clientId, clientWs] of currentSessionData.clients.entries()) {
|
||||
if (clientWs === ws) {
|
||||
disconnectedClientId = clientId;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (disconnectedClientId) {
|
||||
sessionData.clients.delete(disconnectedClientId);
|
||||
console.log(`Client ${disconnectedClientId} disconnected from session: ${sessionId}. Remaining clients: ${sessionData.clients.size}`);
|
||||
if (disconnectedClientId) {
|
||||
currentSessionData.clients.delete(disconnectedClientId);
|
||||
console.log(`Client ${disconnectedClientId} disconnected from session: ${sessionId}. Remaining clients: ${currentSessionData.clients.size}`);
|
||||
} else {
|
||||
console.log(`An unregistered client disconnected from session: ${sessionId}.`);
|
||||
}
|
||||
|
||||
if (currentSessionData.clients.size === 0) {
|
||||
// Only purge session if it's in SETUP, FINAL, or ERROR state
|
||||
if (currentSessionData.state === SessionState.SETUP ||
|
||||
currentSessionData.state === SessionState.FINAL ||
|
||||
currentSessionData.state === SessionState.ERROR) {
|
||||
sessions.delete(sessionId);
|
||||
logEvent('session_purged', sessionId);
|
||||
console.log(`Session ${sessionId} closed and state cleared.`);
|
||||
} else {
|
||||
console.log(`Session ${sessionId} is in ${currentSessionData.state} state. Not purging despite no active clients.`);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
console.log(`An unregistered client disconnected from session: ${sessionId}.`);
|
||||
}
|
||||
|
||||
if (sessionData.clients.size === 0) {
|
||||
sessions.delete(sessionId);
|
||||
logEvent('session_purged', sessionId);
|
||||
console.log(`Session ${sessionId} closed and state cleared.`);
|
||||
console.log(`Client disconnected from session: ${sessionId}. Session data was null.`);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -189,131 +194,337 @@ export const createWebSocketServer = (server: any) => {
|
||||
return wss;
|
||||
};
|
||||
|
||||
export const handleWebSocketMessage = async (ws: WebSocket, sessionId: string, parsedMessage: any) => {
|
||||
export const handleWebSocketMessage = async (ws: WebSocket, sessionId: string, parsedMessage: any): Promise<SessionData | null> => {
|
||||
const { type, clientId, payload } = parsedMessage;
|
||||
|
||||
if (!clientId) {
|
||||
console.error(`Received message without clientId in session ${sessionId}. Type: ${type}`);
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'clientId is required' } }));
|
||||
return;
|
||||
return sessions.get(sessionId) || null; // Return current session state if available
|
||||
}
|
||||
|
||||
const sessionData = sessions.get(sessionId)!;
|
||||
let sessionData = sessions.get(sessionId);
|
||||
|
||||
if (!sessionData.clients.has(clientId)) {
|
||||
sessionData.clients.set(clientId, ws);
|
||||
console.log(`Client ${clientId} registered for session: ${sessionId}. Total clients: ${sessionData.clients.size}`);
|
||||
ws.send(JSON.stringify({ type: 'STATE_UPDATE', payload: { session: getSerializableSession(sessionData, clientId) } }));
|
||||
}
|
||||
|
||||
|
||||
console.log(`Received message from ${clientId} in session ${sessionId}:`, type);
|
||||
// Update lastActivity timestamp on any message
|
||||
|
||||
switch (type) {
|
||||
case 'REGISTER_CLIENT':
|
||||
console.log(`Client ${clientId} registered successfully for session ${sessionId}.`);
|
||||
break;
|
||||
if (sessionData) {
|
||||
|
||||
case 'PING':
|
||||
// Respond to client pings with a pong
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.pong();
|
||||
}
|
||||
break;
|
||||
sessionData.lastActivity = Date.now();
|
||||
|
||||
case 'SETUP_SESSION':
|
||||
if (sessionData.state === SessionState.SETUP) {
|
||||
const { expectedResponses, topic, description } = payload;
|
||||
if (typeof expectedResponses !== 'number' || expectedResponses <= 0) {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Invalid expectedResponses' } }));
|
||||
return;
|
||||
}
|
||||
sessionData.expectedResponses = expectedResponses;
|
||||
sessionData.topic = topic || 'Untitled Session';
|
||||
sessionData.description = description || null;
|
||||
sessionData.state = SessionState.GATHERING;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
console.log(`Session ${sessionId} moved to GATHERING with topic "${sessionData.topic}" and ${expectedResponses} expected responses.`);
|
||||
} else {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in SETUP state. Current state: ${sessionData.state}` } }));
|
||||
}
|
||||
break;
|
||||
console.log(`Session ${sessionId}: lastActivity updated to ${sessionData.lastActivity}`);
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
// If sessionData is null here, it means a JOIN_SESSION message hasn't been processed yet for a new session.
|
||||
|
||||
// The JOIN_SESSION case will handle its creation.
|
||||
|
||||
if (!sessionData && type !== 'JOIN_SESSION') {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session not found and message is not JOIN_SESSION' } }));
|
||||
|
||||
return null; // Session not found, return null
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
switch (type) {
|
||||
|
||||
case 'JOIN_SESSION':
|
||||
|
||||
if (!sessionData) {
|
||||
|
||||
// Create a new session if it doesn't exist
|
||||
|
||||
const newSessionData: SessionData = {
|
||||
|
||||
state: SessionState.SETUP,
|
||||
|
||||
topic: null,
|
||||
|
||||
description: null,
|
||||
|
||||
expectedResponses: 0,
|
||||
|
||||
submittedCount: 0,
|
||||
|
||||
responses: new Map<string, any>(),
|
||||
|
||||
clients: new Map<string, WebSocket>(),
|
||||
|
||||
finalResult: null,
|
||||
|
||||
lastActivity: Date.now(), // Initialize lastActivity
|
||||
|
||||
};
|
||||
|
||||
sessions.set(sessionId, newSessionData); // Explicitly set in global map
|
||||
|
||||
sessionData = newSessionData; // Update local reference to the newly created session
|
||||
|
||||
console.log(`New session ${sessionId} created upon client ${clientId} joining. lastActivity: ${sessionData.lastActivity}`);
|
||||
|
||||
case 'SUBMIT_RESPONSE':
|
||||
if (sessionData.state === SessionState.GATHERING) {
|
||||
if (sessionData.responses.has(clientId)) {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'You have already submitted a response for this session.' } }));
|
||||
return;
|
||||
}
|
||||
|
||||
const { wants, accepts, noGoes, afraidToAsk } = payload.response;
|
||||
if ([...wants, ...accepts, ...noGoes].some(desire => desire.length > 500) || afraidToAsk.length > 500) {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'One of your desires or afraidToAsk exceeds the 500 character limit.' } }));
|
||||
return;
|
||||
|
||||
|
||||
// Register the client to the session's clients map
|
||||
|
||||
if (!sessionData.clients.has(clientId)) {
|
||||
|
||||
sessionData.clients.set(clientId, ws);
|
||||
|
||||
console.log(`Client ${clientId} joined session: ${sessionId}. Total clients: ${sessionData.clients.size}`);
|
||||
|
||||
}
|
||||
|
||||
const hasContradictionsGist = await llmService.checkForInnerContradictions(payload.response);
|
||||
if (hasContradictionsGist) {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Your submission contains inner contradictions: ${hasContradictionsGist} Please resolve them and submit again.` } }));
|
||||
return;
|
||||
ws.send(JSON.stringify({ type: 'STATE_UPDATE', payload: { session: getSerializableSession(sessionData, clientId) } }));
|
||||
|
||||
console.log(`Client ${clientId} received STATE_UPDATE for session ${sessionId} upon joining.`);
|
||||
|
||||
return sessionData; // Return the updated sessionData
|
||||
|
||||
|
||||
|
||||
case 'PING':
|
||||
|
||||
if (!sessionData) { ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for PING' } })); return null; }
|
||||
|
||||
// Respond to client pings with a pong
|
||||
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
|
||||
ws.pong();
|
||||
|
||||
}
|
||||
|
||||
const encryptedWants = wants.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedAccepts = accepts.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedNoGoes = noGoes.map((d: string) => encryptionService.encrypt(d));
|
||||
const encryptedAfraidToAsk = encryptionService.encrypt(afraidToAsk);
|
||||
return sessionData || null; // Return current sessionData or null if undefined
|
||||
|
||||
sessionData.responses.set(clientId, { wants: encryptedWants, accepts: encryptedAccepts, noGoes: encryptedNoGoes, afraidToAsk: encryptedAfraidToAsk });
|
||||
sessionData.submittedCount++;
|
||||
logEvent('response_submitted', sessionId, { clientId, submittedCount: sessionData.submittedCount });
|
||||
console.log(`Client ${clientId} submitted response. Submitted count: ${sessionData.submittedCount}/${sessionData.expectedResponses}`);
|
||||
case 'SETUP_SESSION':
|
||||
|
||||
if (!sessionData) { ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for SETUP_SESSION' } })); return null; }
|
||||
|
||||
if (sessionData.state === SessionState.SETUP) {
|
||||
|
||||
const { expectedResponses, topic, description } = payload;
|
||||
|
||||
if (typeof expectedResponses !== 'number' || expectedResponses <= 0) {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Invalid expectedResponses' } }));
|
||||
|
||||
return sessionData || null; // Return current sessionData on error
|
||||
|
||||
}
|
||||
|
||||
sessionData.expectedResponses = expectedResponses;
|
||||
|
||||
sessionData.topic = topic || 'Untitled Session';
|
||||
|
||||
sessionData.description = description || null;
|
||||
|
||||
sessionData.state = SessionState.GATHERING;
|
||||
|
||||
if (sessionData.submittedCount === sessionData.expectedResponses) {
|
||||
sessionData.state = SessionState.HARMONIZING;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
logEvent('session_harmonizing', sessionId, { expectedResponses: sessionData.expectedResponses });
|
||||
console.log(`Session ${sessionId} moved to HARMONIZING. Triggering LLM analysis.`);
|
||||
|
||||
// Perform LLM analysis asynchronously
|
||||
(async () => {
|
||||
let durationMs: number = 0; // Declare here
|
||||
try {
|
||||
logEvent('llm_analysis_started', sessionId);
|
||||
const startTime = process.hrtime.bigint(); const allDecryptedDesires = Array.from(sessionData.responses.values()).map(encryptedResponse => {
|
||||
const decryptedWants = encryptedResponse.wants.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedAccepts = encryptedResponse.accepts.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedNoGoes = encryptedResponse.noGoes.map((d: string) => encryptionService.decrypt(d));
|
||||
const decryptedAfraidToAsk = encryptionService.decrypt(encryptedResponse.afraidToAsk);
|
||||
return { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: decryptedAfraidToAsk };
|
||||
});
|
||||
const decision = await llmService.analyzeDesires(allDecryptedDesires);
|
||||
console.log(`Session ${sessionId} moved to GATHERING with topic "${sessionData.topic}" and ${expectedResponses} expected responses.`);
|
||||
|
||||
sessionData.finalResult = decision;
|
||||
sessionData.state = SessionState.FINAL;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
logEvent('llm_analysis_completed', sessionId, { result: decision });
|
||||
recordMetric('llm_analysis_duration', durationMs, sessionId, { status: 'success' }); recordMetric('llm_analysis_availability', 'available', sessionId);
|
||||
console.log(`Analysis complete for session ${sessionId}. Result:`, decision);
|
||||
}
|
||||
|
||||
else {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in SETUP state. Current state: ${sessionData.state}` } }));
|
||||
|
||||
}
|
||||
|
||||
return sessionData || null; // Return current sessionData
|
||||
|
||||
|
||||
|
||||
case 'SUBMIT_RESPONSE':
|
||||
|
||||
if (!sessionData) { ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for SUBMIT_RESPONSE' } })); return null; }
|
||||
|
||||
if (sessionData.state === SessionState.GATHERING) {
|
||||
|
||||
if (sessionData.responses.has(clientId)) {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'You have already submitted a response for this session.' } }));
|
||||
|
||||
return sessionData || null; // Return current sessionData on error
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
const { wants, accepts, noGoes, afraidToAsk } = payload.response;
|
||||
|
||||
if ([...wants, ...accepts, ...noGoes].some((desire: string) => desire.length > 500) || afraidToAsk.length > 500) {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'One of your desires or afraidToAsk exceeds the 500 character limit.' } }));
|
||||
|
||||
return sessionData || null; // Return current sessionData on error
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
const hasContradictionsGist = await llmService.checkForInnerContradictions(payload.response);
|
||||
|
||||
if (hasContradictionsGist) {
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Your submission contains inner contradictions: ${hasContradictionsGist} Please resolve them and submit again.` } }));
|
||||
|
||||
return sessionData || null; // Return current sessionData on error
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
const encryptedWants = wants.map((d: string) => encryptionService.encrypt(d));
|
||||
|
||||
const encryptedAccepts = accepts.map((d: string) => encryptionService.encrypt(d));
|
||||
|
||||
const encryptedNoGoes = noGoes.map((d: string) => encryptionService.encrypt(d));
|
||||
|
||||
const encryptedAfraidToAsk = encryptionService.encrypt(afraidToAsk);
|
||||
|
||||
|
||||
|
||||
sessionData.responses.set(clientId, { wants: encryptedWants, accepts: encryptedAccepts, noGoes: encryptedNoGoes, afraidToAsk: encryptedAfraidToAsk });
|
||||
|
||||
sessionData.submittedCount++;
|
||||
|
||||
|
||||
|
||||
console.log(`Client ${clientId} submitted response. Submitted count: ${sessionData.submittedCount}/${sessionData.expectedResponses}`);
|
||||
|
||||
|
||||
|
||||
if (sessionData.submittedCount === sessionData.expectedResponses) {
|
||||
|
||||
sessionData.state = SessionState.HARMONIZING;
|
||||
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
|
||||
console.log(`Session ${sessionId} moved to HARMONIZING. Triggering LLM analysis.`);
|
||||
|
||||
|
||||
|
||||
// Perform LLM analysis asynchronously
|
||||
|
||||
(async () => {
|
||||
|
||||
try {
|
||||
|
||||
console.log('llm_analysis_started', sessionId);
|
||||
|
||||
const startTime = process.hrtime.bigint(); const allDecryptedDesires = Array.from(sessionData.responses.values()).map(encryptedResponse => {
|
||||
|
||||
const decryptedWants = encryptedResponse.wants.map((d: string) => encryptionService.decrypt(d));
|
||||
|
||||
const decryptedAccepts = encryptedResponse.accepts.map((d: string) => encryptionService.decrypt(d));
|
||||
|
||||
const decryptedNoGoes = encryptedResponse.noGoes.map((d: string) => encryptionService.decrypt(d));
|
||||
|
||||
const decryptedAfraidToAsk = encryptionService.decrypt(encryptedResponse.afraidToAsk);
|
||||
|
||||
return { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: decryptedAfraidToAsk };
|
||||
|
||||
});
|
||||
|
||||
const decision = await llmService.analyzeDesires(allDecryptedDesires);
|
||||
|
||||
|
||||
|
||||
sessionData.finalResult = decision;
|
||||
|
||||
sessionData.state = SessionState.FINAL;
|
||||
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
|
||||
console.log('llm_analysis_completed', sessionId, { result: decision });
|
||||
|
||||
console.log('llm_analysis_duration', 0, sessionId, { status: 'success' }); console.log('llm_analysis_availability', 'available', sessionId);
|
||||
|
||||
console.log(`Analysis complete for session ${sessionId}. Result:`, decision);
|
||||
|
||||
|
||||
|
||||
} catch (error: any) {
|
||||
|
||||
console.error(`Error during analysis for session ${sessionId}:`, error.message);
|
||||
|
||||
sessionData.state = SessionState.ERROR;
|
||||
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
|
||||
console.log('llm_analysis_error', sessionId, { error: error.message });
|
||||
|
||||
console.log('llm_analysis_availability', 'unavailable', sessionId, { error: error.message });
|
||||
|
||||
}
|
||||
|
||||
})();
|
||||
|
||||
} else {
|
||||
|
||||
// Only broadcast the latest count if the session is not yet harmonizing
|
||||
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
|
||||
}
|
||||
|
||||
} catch (error: any) {
|
||||
console.error(`Error during analysis for session ${sessionId}:`, error.message);
|
||||
sessionData.state = SessionState.ERROR;
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
logEvent('llm_analysis_error', sessionId, { error: error.message });
|
||||
recordMetric('llm_analysis_availability', 'unavailable', sessionId, { error: error.message });
|
||||
}
|
||||
})();
|
||||
} else {
|
||||
// Only broadcast the latest count if the session is not yet harmonizing
|
||||
broadcastToSession(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
||||
}
|
||||
} else {
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in GATHERING state. Current state: ${sessionData.state}` } }));
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
console.warn(`Unknown message type: ${type} from client ${clientId} in session ${sessionId}`);
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Unknown message type: ${type}` } }));
|
||||
break;
|
||||
}
|
||||
};
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in GATHERING state. Current state: ${sessionData.state}` } }));
|
||||
|
||||
}
|
||||
|
||||
return sessionData || null; // Return current sessionData
|
||||
|
||||
|
||||
|
||||
default:
|
||||
|
||||
console.warn(`Unknown message type: ${type} from client ${clientId} in session ${sessionId}`);
|
||||
|
||||
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Unknown message type: ${type}` } }));
|
||||
|
||||
return sessionData || null; // Return current sessionData
|
||||
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
|
||||
|
||||
const cleanupInactiveSessions = () => {
|
||||
|
||||
console.log('Running cleanupInactiveSessions...');
|
||||
|
||||
const now = Date.now();
|
||||
|
||||
for (const [sessionId, sessionData] of sessions.entries()) {
|
||||
|
||||
console.log(`Session ${sessionId}: clients.size=${sessionData.clients.size}, lastActivity=${sessionData.lastActivity}, timeSinceLastActivity=${now - sessionData.lastActivity}, SESSION_TIMEOUT_MS=${SESSION_TIMEOUT_MS}`);
|
||||
|
||||
if (sessionData.clients.size === 0 && (now - sessionData.lastActivity > SESSION_TIMEOUT_MS)) {
|
||||
|
||||
sessions.delete(sessionId);
|
||||
|
||||
logEvent('session_purged_inactive', sessionId);
|
||||
|
||||
console.log(`Inactive session ${sessionId} purged.`);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user