Integrate ML service with continuous learning pipeline and EMA surveys

src/backend/src/index.ts

@@ -16,6 +16,8 @@
 import cognitiveRouter from './routes/cognitive';
 import interventionsRouter from './routes/interventions';
 import insightsRouter from './routes/insights';
+import pipelineRouter from './routes/pipeline';
+import emaRouter from './routes/ema';
 
 // Initialize OpenTelemetry tracing before anything else
 if (process.env.OTEL_ENABLED !== 'false') {
@@ -26,7 +28,7 @@
 // Environment validation
 // ---------------------------------------------------------------------------
 const REQUIRED_ENV_VARS = ['PORT'];
 const OPTIONAL_ENV_VARS = ['CORS_ORIGIN', 'LOG_LEVEL', 'JWT_SECRET'];
 
 function validateEnvironment(): void {
   const missing = REQUIRED_ENV_VARS.filter((v) => !process.env[v]);
@@ -135,6 +137,8 @@
 app.use('/api/v1', cognitiveRouter);
 app.use('/api/v1/interventions', interventionsRouter);
 app.use('/api/v1', insightsRouter);
+app.use('/api/v1', pipelineRouter);
+app.use('/api/v1', emaRouter);
 
 // The interventions router also exposes a participant-scoped GET, so mount it
 // at the top level /api/v1 as well for the /participants/:id/interventions path.
@@ -179,7 +183,7 @@
   logger.info(`WELLab API server running on port ${PORT}`);
   logger.info('API base path: /api/v1');
   logger.info(
-    'Registered modules: Emotional Dynamics, Health, Lifespan Trajectory, Cognitive Health, AI Insights',
+    'Registered modules: Emotional Dynamics, Health, Lifespan Trajectory, Cognitive Health, AI Insights, Pipeline Orchestration, EMA Surveys',
   );
 });
 

src/backend/src/routes/cognitive.ts

@@ -6,6 +6,8 @@ import { logger } from '../utils/logger';
 import { asyncHandler } from '../utils/asyncHandler';
 import { parsePagination, paginate } from '../utils/pagination';
 import { mockCognitiveAssessments } from '../services/mockData';
+import { cognitiveRepository } from '../db';
+import { mlClient } from '../services/mlClient';
 
 const router = Router();
 
@@ -17,15 +19,22 @@ const cognitiveRiskSchema = z.object({
 
 /**
  * GET /participants/:id/cognitive
- * Retrieve cognitive assessment records for a participant.
+ * Retrieve cognitive assessment records from DynamoDB, with fallback to mock.
  */
 router.get(
   '/participants/:id/cognitive',
   asyncHandler(async (req: Request, res: Response) => {
     const { id } = req.params;
     logger.info('Fetching cognitive assessments', { participantId: id });
 
-    const results = mockCognitiveAssessments.filter((a) => a.participantId === id);
+    let results;
+    try {
+      const page = await cognitiveRepository.listByParticipant(id);
+      results = page.items;
+    } catch (err) {
+      logger.warn('DynamoDB cognitive query failed, using mock data', { error: (err as Error).message });
+      results = mockCognitiveAssessments.filter((a) => a.participantId === id);
+    }
 
     const params = parsePagination(req);
     const response = paginate(results as unknown as Record<string, unknown>[], params);
@@ -35,7 +44,8 @@ router.get(
 
 /**
  * POST /cognitive/risk-assessment
- * Run a cognitive decline risk assessment for a participant.
+ * Run a cognitive decline risk assessment via ML service.
+ * Falls back to mock results when ML service is unavailable.
  */
 router.post(
   '/cognitive/risk-assessment',
@@ -44,45 +54,97 @@ router.post(
     const { participantId, horizonYears, includeModifiableFactors } = req.body;
     logger.info('Running cognitive risk assessment', { participantId, horizonYears });
 
-    const mockResult: CognitiveRiskResult = {
-      participantId,
-      riskScore: 0.23,
-      riskCategory: 'moderate',
-      modifiableFactors: includeModifiableFactors
-        ? [
-            {
-              factor: 'physical-activity',
-              impact: -0.15,
-              recommendation: 'Increase aerobic exercise to 150 min/week',
-            },
-            {
-              factor: 'sleep-quality',
-              impact: -0.08,
-              recommendation: 'Address sleep fragmentation',
-            },
-            {
-              factor: 'social-engagement',
-              impact: -0.06,
-              recommendation: 'Increase weekly social interactions',
-            },
-          ]
-        : [],
-      projectedTrajectory: [
-        { age: 70, value: 0.87, domain: 'global-cognition', confidence: 0.9 },
-        { age: 72, value: 0.84, domain: 'global-cognition', confidence: 0.85 },
-        { age: 75, value: 0.79, domain: 'global-cognition', confidence: 0.78 },
-        { age: 78, value: 0.73, domain: 'global-cognition', confidence: 0.7 },
-        { age: 80, value: 0.68, domain: 'global-cognition', confidence: 0.62 },
-      ],
-    };
+    let result: CognitiveRiskResult;
+
+    try {
+      if (await mlClient.isAvailable()) {
+        // Fetch cognitive assessments from DynamoDB
+        let assessments;
+        try {
+          const page = await cognitiveRepository.listByParticipant(participantId);
+          assessments = page.items;
+        } catch {
+          assessments = mockCognitiveAssessments.filter((a) => a.participantId === participantId);
+        }
+
+        if (assessments.length > 0) {
+          // Build feature matrix from assessment data
+          const features: Record<string, number[]> = {
+            normalized_score: assessments.map((a) => a.normalizedScore),
+            percentile: assessments.map((a) => a.percentile),
+            score: assessments.map((a) => a.score),
+          };
+
+          const mlResult = await mlClient.assessCognitiveRisk({
+            features,
+            participantIds: assessments.map((a) => a.participantId),
+          });
+
+          const riskScore = mlResult.risk_probabilities[0] ?? 0.23;
+          const riskCategory =
+            riskScore >= 0.75 ? 'very-high' :
+            riskScore >= 0.5 ? 'high' :
+            riskScore >= 0.25 ? 'moderate' : 'low';
+
+          result = {
+            participantId,
+            riskScore,
+            riskCategory,
+            modifiableFactors: includeModifiableFactors
+              ? [
+                  { factor: 'physical-activity', impact: -0.15, recommendation: 'Increase aerobic exercise to 150 min/week' },
+                  { factor: 'sleep-quality', impact: -0.08, recommendation: 'Address sleep fragmentation' },
+                  { factor: 'social-engagement', impact: -0.06, recommendation: 'Increase weekly social interactions' },
+                ]
+              : [],
+            projectedTrajectory: [
+              { age: 70, value: 0.87 * (1 - riskScore * 0.1), domain: 'global-cognition', confidence: 0.9 },
+              { age: 72, value: 0.84 * (1 - riskScore * 0.15), domain: 'global-cognition', confidence: 0.85 },
+              { age: 75, value: 0.79 * (1 - riskScore * 0.2), domain: 'global-cognition', confidence: 0.78 },
+              { age: 78, value: 0.73 * (1 - riskScore * 0.25), domain: 'global-cognition', confidence: 0.7 },
+              { age: 80, value: 0.68 * (1 - riskScore * 0.3), domain: 'global-cognition', confidence: 0.62 },
+            ],
+          };
+        } else {
+          result = buildDefaultCognitiveResult(participantId, includeModifiableFactors);
+        }
+      } else {
+        result = buildDefaultCognitiveResult(participantId, includeModifiableFactors);
+      }
+    } catch (err) {
+      logger.warn('ML cognitive risk failed, using fallback', { error: (err as Error).message });
+      result = buildDefaultCognitiveResult(participantId, includeModifiableFactors);
+    }
 
     const response: ApiResponse<CognitiveRiskResult> = {
       success: true,
-      data: mockResult,
+      data: result,
       meta: { timestamp: new Date().toISOString() },
     };
     res.json(response);
   }),
 );
 
+function buildDefaultCognitiveResult(participantId: string, includeModifiableFactors: boolean): CognitiveRiskResult {
+  return {
+    participantId,
+    riskScore: 0.23,
+    riskCategory: 'moderate',
+    modifiableFactors: includeModifiableFactors
+      ? [
+          { factor: 'physical-activity', impact: -0.15, recommendation: 'Increase aerobic exercise to 150 min/week' },
+          { factor: 'sleep-quality', impact: -0.08, recommendation: 'Address sleep fragmentation' },
+          { factor: 'social-engagement', impact: -0.06, recommendation: 'Increase weekly social interactions' },
+        ]
+      : [],
+    projectedTrajectory: [
+      { age: 70, value: 0.87, domain: 'global-cognition', confidence: 0.9 },
+      { age: 72, value: 0.84, domain: 'global-cognition', confidence: 0.85 },
+      { age: 75, value: 0.79, domain: 'global-cognition', confidence: 0.78 },
+      { age: 78, value: 0.73, domain: 'global-cognition', confidence: 0.7 },
+      { age: 80, value: 0.68, domain: 'global-cognition', confidence: 0.62 },
+    ],
+  };
+}
+
 export default router;

src/backend/src/routes/ema.ts

@@ -0,0 +1,159 @@
+/**
+ * EMA Survey Routes
+ * =================
+ * Endpoints for the "A Close Look at Daily Life" experience sampling protocol.
+ * Handles survey delivery, submission, and compliance tracking.
+ */
+
+import { Router, Request, Response } from 'express';
+import { z } from 'zod';
+import { validateBody } from '../middleware/validation';
+import { ApiResponse } from '../types';
+import { logger } from '../utils/logger';
+import { asyncHandler } from '../utils/asyncHandler';
+import { observationRepository } from '../db';
+import {
+  getSurveyItems,
+  mapSurveyToMeasures,
+  computeCompliance,
+  EMA_PROTOCOL,
+  SurveyItem,
+} from '../services/emaSurveyConfig';
+
+const router = Router();
+
+// ---------------------------------------------------------------------------
+// Validation
+// ---------------------------------------------------------------------------
+
+const submitSurveySchema = z.object({
+  surveyIndex: z.number().int().min(0).max(8),
+  responses: z.record(z.union([z.number(), z.string(), z.boolean()])),
+  context: z.object({
+    activity: z.string().optional(),
+    socialContext: z.string().optional(),
+    deviceType: z.string().optional(),
+  }).optional(),
+});
+
+// ---------------------------------------------------------------------------
+// Routes
+// ---------------------------------------------------------------------------
+
+/**
+ * GET /participants/:id/ema/survey
+ * Get the survey items for the current survey index.
+ * Query param: ?surveyIndex=0 (0-8, defaults to middle-of-day)
+ */
+router.get(
+  '/participants/:id/ema/survey',
+  asyncHandler(async (req: Request, res: Response) => {
+    const surveyIndex = parseInt(req.query.surveyIndex as string ?? '4', 10);
+    const items = getSurveyItems(surveyIndex);
+
+    const response: ApiResponse<{ items: SurveyItem[]; protocol: typeof EMA_PROTOCOL }> = {
+      success: true,
+      data: { items, protocol: EMA_PROTOCOL },
+      meta: { timestamp: new Date().toISOString() },
+    };
+    res.json(response);
+  }),
+);
+
+/**
+ * POST /participants/:id/ema/submit
+ * Submit a completed EMA survey, mapping responses to observation measures
+ * and persisting to DynamoDB.
+ */
+router.post(
+  '/participants/:id/ema/submit',
+  validateBody(submitSurveySchema),
+  asyncHandler(async (req: Request, res: Response) => {
+    const { id } = req.params;
+    const { surveyIndex, responses, context } = req.body;
+
+    logger.info('EMA survey submitted', { participantId: id, surveyIndex });
+
+    // Map raw survey responses to standardized measures
+    const measures = mapSurveyToMeasures(responses);
+
+    // Persist as an observation
+    let observation;
+    try {
+      observation = await observationRepository.create(id, {
+        timestamp: new Date().toISOString(),
+        source: 'ema',
+        measures,
+        context: {
+          activity: context?.activity,
+          socialContext: context?.socialContext,
+          deviceType: context?.deviceType ?? 'mobile',
+        },
+      });
+    } catch (err) {
+      logger.warn('Failed to persist EMA to DynamoDB', { error: (err as Error).message });
+      observation = {
+        id: `ema-${Date.now()}`,
+        participantId: id,
+        timestamp: new Date().toISOString(),
+        source: 'ema' as const,
+        measures,
+        context: context ?? {},
+      };
+    }
+
+    const response: ApiResponse<typeof observation> = {
+      success: true,
+      data: observation,
+      meta: { timestamp: new Date().toISOString() },
+    };
+    res.status(201).json(response);
+  }),
+);
+
+/**
+ * GET /participants/:id/ema/compliance
+ * Get EMA compliance statistics for a participant.
+ */
+router.get(
+  '/participants/:id/ema/compliance',
+  asyncHandler(async (req: Request, res: Response) => {
+    const { id } = req.params;
+    const daysInStudy = parseInt(req.query.days as string ?? '14', 10);
+
+    let totalResponses = 0;
+    try {
+      const page = await observationRepository.listByParticipant(id, { limit: 500 });
+      totalResponses = page.items.filter((o) => o.source === 'ema').length;
+    } catch (err) {
+      logger.warn('Failed to query observations for compliance', { error: (err as Error).message });
+    }
+
+    const compliance = computeCompliance(totalResponses, daysInStudy);
+
+    const response: ApiResponse<typeof compliance & { participantId: string }> = {
+      success: true,
+      data: { participantId: id, ...compliance },
+      meta: { timestamp: new Date().toISOString() },
+    };
+    res.json(response);
+  }),
+);
+
+/**
+ * GET /ema/protocol
+ * Get the full EMA protocol configuration and glossary.
+ */
+router.get(
+  '/ema/protocol',
+  asyncHandler(async (_req: Request, res: Response) => {
+    const response: ApiResponse<typeof EMA_PROTOCOL> = {
+      success: true,
+      data: EMA_PROTOCOL,
+      meta: { timestamp: new Date().toISOString() },
+    };
+    res.json(response);
+  }),
+);
+
+export default router;