FAC-130.2 refactor coverage stats handling for pipeline status queries (#328) (#330)

y4nder · claude · web-flow · commit 76e4b8f30b42 · 2026-04-14T03:20:54.000+08:00
Coverage stats (submissionCount, totalEnrolled, commentCount, responseRate) were computed once in CreatePipeline and cached on the AnalysisPipeline entity. GetPipelineStatus read them from the entity and never refreshed, so pipelines created early in data collection reported stale numbers forever — appearing as a hard "limit" to users when more submissions arrived afterwards. Now, while a pipeline is still in AWAITING_CONFIRMATION, GetPipelineStatus recomputes coverage live and persists the fresh values (including warnings) back to the entity so what the user sees matches what will be locked in at confirmation time. After confirmation, the stored snapshot is preserved untouched — it represents the corpus that was actually analyzed and must not drift. Refactors: - Extract BuildScopeFromPipeline helper (entity -> ScopeFilter) - Extract BuildCoverageWarnings helper (reused by CreatePipeline and GetPipelineStatus) Tests: 2 new cases cover the fresh-recompute path and the snapshot-preserved path for confirmed pipelines. https://claude.ai/code/session_01AsGM2DbyriRMyLWHr7Hwdw Co-authored-by: Claude <noreply@anthropic.com>
diff --git a/src/modules/analysis/services/pipeline-orchestrator.service.spec.ts b/src/modules/analysis/services/pipeline-orchestrator.service.spec.ts
@@ -677,6 +677,80 @@ describe('PipelineOrchestratorService', () => {
       expect(status.stages.embeddings.status).toBe('failed');
     });
 
+    it('should recompute coverage stats for AWAITING_CONFIRMATION pipelines', async () => {
+      // Pipeline was created with a stale snapshot (200 submissions, 205 enrolled).
+      // By the time GetPipelineStatus is called, more submissions have arrived.
+      const pipeline = {
+        ...basePipeline,
+        status: PipelineStatus.AWAITING_CONFIRMATION,
+        totalEnrolled: 205,
+        submissionCount: 200,
+        commentCount: 150,
+        responseRate: 200 / 205,
+        warnings: ['Only 200 submissions (minimum recommended: 30).'],
+      };
+
+      mockFork.findOne
+        .mockResolvedValueOnce(pipeline) // pipeline lookup
+        .mockResolvedValueOnce(null) // sentiment run
+        .mockResolvedValueOnce(null) // topic model run
+        .mockResolvedValueOnce(null) // recommendation run
+        .mockResolvedValueOnce({ updatedAt: new Date() }); // latest enrollment in ComputeCoverageStats
+
+      // ComputeCoverageStats flow: count submissions, count comments, find
+      // scoped submissions for course ids, count enrollments.
+      mockFork.count
+        .mockResolvedValueOnce(520) // fresh submissionCount
+        .mockResolvedValueOnce(410) // fresh commentCount
+        .mockResolvedValueOnce(600); // fresh totalEnrolled
+      mockFork.find.mockResolvedValueOnce([{ course: { id: 'c1' } }]);
+
+      const status = await service.GetPipelineStatus('p1');
+
+      // Fresh values, not the stale snapshot
+      expect(status.coverage.submissionCount).toBe(520);
+      expect(status.coverage.totalEnrolled).toBe(600);
+      expect(status.coverage.commentCount).toBe(410);
+      expect(status.coverage.responseRate).toBeCloseTo(520 / 600, 5);
+
+      // Pipeline entity was mutated with fresh values and flushed
+      expect(pipeline.submissionCount).toBe(520);
+      expect(pipeline.totalEnrolled).toBe(600);
+      expect(mockFork.flush).toHaveBeenCalled();
+
+      // Stale "Only 200 submissions" warning is replaced with fresh warnings
+      expect(
+        status.warnings.some((w) => w.includes('Only 200 submissions')),
+      ).toBe(false);
+    });
+
+    it('should use stored coverage snapshot for confirmed pipelines', async () => {
+      const pipeline = {
+        ...basePipeline,
+        status: PipelineStatus.SENTIMENT_ANALYSIS,
+        totalEnrolled: 205,
+        submissionCount: 200,
+        commentCount: 150,
+        responseRate: 200 / 205,
+      };
+
+      mockFork.findOne
+        .mockResolvedValueOnce(pipeline)
+        .mockResolvedValueOnce(null)
+        .mockResolvedValueOnce(null)
+        .mockResolvedValueOnce(null);
+
+      // No scoped submissions → no fresh recompute; stored snapshot wins
+      mockFork.find.mockResolvedValueOnce([]);
+
+      const status = await service.GetPipelineStatus('p1');
+
+      expect(status.coverage.submissionCount).toBe(200);
+      expect(status.coverage.totalEnrolled).toBe(205);
+      // ComputeCoverageStats should NOT have been invoked (no count calls)
+      expect(mockFork.count).not.toHaveBeenCalled();
+    });
+
     it('should return sentiment gate included/excluded with completed status', async () => {
       const pipeline = {
         ...basePipeline,
diff --git a/src/modules/analysis/services/pipeline-orchestrator.service.ts b/src/modules/analysis/services/pipeline-orchestrator.service.ts
@@ -128,34 +128,7 @@ export class PipelineOrchestratorService {
     if (input.courseId) scope.course = input.courseId;
 
     const coverage = await this.ComputeCoverageStats(fork, scope);
-
-    // Generate warnings
-    const warnings: string[] = [];
-    if (coverage.responseRate < COVERAGE_WARNINGS.MIN_RESPONSE_RATE) {
-      warnings.push(
-        `Response rate is ${(coverage.responseRate * 100).toFixed(1)}% (below ${COVERAGE_WARNINGS.MIN_RESPONSE_RATE * 100}% threshold).`,
-      );
-    }
-    if (coverage.submissionCount < COVERAGE_WARNINGS.MIN_SUBMISSIONS) {
-      warnings.push(
-        `Only ${coverage.submissionCount} submissions (minimum recommended: ${COVERAGE_WARNINGS.MIN_SUBMISSIONS}).`,
-      );
-    }
-    if (coverage.commentCount < COVERAGE_WARNINGS.MIN_COMMENTS) {
-      warnings.push(
-        `Only ${coverage.commentCount} qualitative comments (minimum recommended: ${COVERAGE_WARNINGS.MIN_COMMENTS}).`,
-      );
-    }
-    if (coverage.lastEnrollmentSyncAt) {
-      const hoursSinceSync =
-        (Date.now() - coverage.lastEnrollmentSyncAt.getTime()) / 3_600_000;
-      if (hoursSinceSync > COVERAGE_WARNINGS.STALE_SYNC_HOURS) {
-        const daysStale = Math.floor(hoursSinceSync / 24);
-        warnings.push(
-          `Enrollment data may be stale (last synced ${daysStale} day${daysStale !== 1 ? 's' : ''} ago).`,
-        );
-      }
-    }
+    const warnings = this.BuildCoverageWarnings(coverage);
 
     const pipeline = fork.create(AnalysisPipeline, {
       semester: fork.getReference(Semester, input.semesterId),
@@ -478,30 +451,63 @@ export class PipelineOrchestratorService {
       });
     }
 
-    // Compute lastEnrollmentSyncAt by scoping through courses in submission scope
-    const scope = buildSubmissionScope(pipeline);
+    // Coverage stats are cached on the pipeline entity at creation time. For
+    // pipelines still awaiting confirmation, recompute on every status fetch
+    // so the user sees the latest submission/enrollment counts before they
+    // lock in the snapshot. After confirmation, the stored values represent
+    // what was actually analyzed and must not drift.
+    const scope = this.BuildScopeFromPipeline(pipeline);
+    let totalEnrolled = pipeline.totalEnrolled;
+    let submissionCount = pipeline.submissionCount;
+    let commentCount = pipeline.commentCount;
+    let responseRate = Number(pipeline.responseRate);
+    let warnings = pipeline.warnings;
     let lastEnrollmentSyncAt: Date | null = null;
-    const scopedSubs = await fork.find(
-      QuestionnaireSubmission,
-      {
-        ...scope,
-        qualitativeComment: { $ne: null },
-      },
-      { fields: ['course'] },
-    );
-    const courseIds = [
-      ...new Set(
-        scopedSubs.map((s) => s.course?.id).filter((id): id is string => !!id),
-      ),
-    ];
-    if (courseIds.length > 0) {
-      const latestEnrollment = await fork.findOne(
-        Enrollment,
-        { isActive: true, course: { $in: courseIds } },
-        { orderBy: { updatedAt: 'DESC' } },
+
+    if (pipeline.status === PipelineStatus.AWAITING_CONFIRMATION) {
+      const freshCoverage = await this.ComputeCoverageStats(fork, scope);
+      totalEnrolled = freshCoverage.totalEnrolled;
+      submissionCount = freshCoverage.submissionCount;
+      commentCount = freshCoverage.commentCount;
+      responseRate = freshCoverage.responseRate;
+      lastEnrollmentSyncAt = freshCoverage.lastEnrollmentSyncAt;
+      warnings = this.BuildCoverageWarnings(freshCoverage);
+
+      // Persist refreshed snapshot so the values shown here match what will
+      // be locked in at confirmation time.
+      pipeline.totalEnrolled = freshCoverage.totalEnrolled;
+      pipeline.submissionCount = freshCoverage.submissionCount;
+      pipeline.commentCount = freshCoverage.commentCount;
+      pipeline.responseRate = freshCoverage.responseRate;
+      pipeline.warnings = warnings;
+      await fork.flush();
+    } else {
+      // For confirmed/terminal pipelines, derive lastEnrollmentSyncAt from
+      // courses in the original submission scope (snapshot view).
+      const scopedSubs = await fork.find(
+        QuestionnaireSubmission,
+        {
+          ...scope,
+          qualitativeComment: { $ne: null },
+        },
+        { fields: ['course'] },
       );
-      if (latestEnrollment) {
-        lastEnrollmentSyncAt = latestEnrollment.updatedAt;
+      const courseIds = [
+        ...new Set(
+          scopedSubs
+            .map((s) => s.course?.id)
+            .filter((id): id is string => !!id),
+        ),
+      ];
+      if (courseIds.length > 0) {
+        const latestEnrollment = await fork.findOne(
+          Enrollment,
+          { isActive: true, course: { $in: courseIds } },
+          { orderBy: { updatedAt: 'DESC' } },
+        );
+        if (latestEnrollment) {
+          lastEnrollmentSyncAt = latestEnrollment.updatedAt;
+        }
       }
     }
 
@@ -554,10 +560,10 @@ export class PipelineOrchestratorService {
         course: pipeline.course?.shortname || null,
       },
       coverage: {
-        totalEnrolled: pipeline.totalEnrolled,
-        submissionCount: pipeline.submissionCount,
-        commentCount: pipeline.commentCount,
-        responseRate: Number(pipeline.responseRate),
+        totalEnrolled,
+        submissionCount,
+        commentCount,
+        responseRate,
         lastEnrollmentSyncAt: lastEnrollmentSyncAt?.toISOString() || null,
       },
       stages: {
@@ -585,7 +591,7 @@ export class PipelineOrchestratorService {
           recommendationRun,
         ),
       },
-      warnings: pipeline.warnings,
+      warnings,
       errorMessage: pipeline.errorMessage ?? null,
       // Intent signal for future error categorization — currently equivalent to status === FAILED
       retryable: pipeline.status === PipelineStatus.FAILED,
@@ -636,6 +642,48 @@ export class PipelineOrchestratorService {
 
   // --- Private Helpers ---
 
+  private BuildScopeFromPipeline(pipeline: AnalysisPipeline): ScopeFilter {
+    const scope: ScopeFilter = { semester: pipeline.semester.id };
+    if (pipeline.faculty) scope.faculty = pipeline.faculty.id;
+    if (pipeline.questionnaireVersion)
+      scope.questionnaireVersion = pipeline.questionnaireVersion.id;
+    if (pipeline.department) scope.department = pipeline.department.id;
+    if (pipeline.program) scope.program = pipeline.program.id;
+    if (pipeline.campus) scope.campus = pipeline.campus.id;
+    if (pipeline.course) scope.course = pipeline.course.id;
+    return scope;
+  }
+
+  private BuildCoverageWarnings(coverage: CoverageStats): string[] {
+    const warnings: string[] = [];
+    if (coverage.responseRate < COVERAGE_WARNINGS.MIN_RESPONSE_RATE) {
+      warnings.push(
+        `Response rate is ${(coverage.responseRate * 100).toFixed(1)}% (below ${COVERAGE_WARNINGS.MIN_RESPONSE_RATE * 100}% threshold).`,
+      );
+    }
+    if (coverage.submissionCount < COVERAGE_WARNINGS.MIN_SUBMISSIONS) {
+      warnings.push(
+        `Only ${coverage.submissionCount} submissions (minimum recommended: ${COVERAGE_WARNINGS.MIN_SUBMISSIONS}).`,
+      );
+    }
+    if (coverage.commentCount < COVERAGE_WARNINGS.MIN_COMMENTS) {
+      warnings.push(
+        `Only ${coverage.commentCount} qualitative comments (minimum recommended: ${COVERAGE_WARNINGS.MIN_COMMENTS}).`,
+      );
+    }
+    if (coverage.lastEnrollmentSyncAt) {
+      const hoursSinceSync =
+        (Date.now() - coverage.lastEnrollmentSyncAt.getTime()) / 3_600_000;
+      if (hoursSinceSync > COVERAGE_WARNINGS.STALE_SYNC_HOURS) {
+        const daysStale = Math.floor(hoursSinceSync / 24);
+        warnings.push(
+          `Enrollment data may be stale (last synced ${daysStale} day${daysStale !== 1 ? 's' : ''} ago).`,
+        );
+      }
+    }
+    return warnings;
+  }
+
   private async ComputeCoverageStats(
     em: EntityManager,
     scope: ScopeFilter,
