Merge pull request #132 from bcgov/develop

Develop

Author: alex-struk

apps/backend-services/src/benchmark/benchmark-run.service.spec.ts

@@ -8,6 +8,7 @@
 import { Prisma } from "@generated/client";
 import { BadRequestException, NotFoundException } from "@nestjs/common";
 import { Test, TestingModule } from "@nestjs/testing";
+import { BLOB_STORAGE } from "@/blob-storage/blob-storage.interface";
 import { PrismaService } from "@/database/prisma.service";
 import { AuditLogService } from "./audit-log.service";
 import { BenchmarkRunService } from "./benchmark-run.service";
@@ -180,6 +181,17 @@ describe("BenchmarkRunService", () => {
             logBaselinePromoted: jest.fn().mockResolvedValue({}),
           },
         },
+        {
+          provide: BLOB_STORAGE,
+          useValue: {
+            read: jest.fn().mockResolvedValue(Buffer.from("{}", "utf-8")),
+            write: jest.fn().mockResolvedValue(undefined),
+            exists: jest.fn().mockResolvedValue(false),
+            delete: jest.fn().mockResolvedValue(undefined),
+            list: jest.fn().mockResolvedValue([]),
+            deleteByPrefix: jest.fn().mockResolvedValue(undefined),
+          },
+        },
       ],
     }).compile();
 

apps/backend-services/src/benchmark/benchmark-run.service.ts

@@ -13,13 +13,19 @@ import { getErrorMessage } from "@ai-di/shared-logging";
 
 import { execSync } from "node:child_process";
 import { createHash } from "node:crypto";
+import { validateBlobFilePath } from "@ai-di/blob-storage-paths";
 import { Prisma } from "@generated/client";
 import {
   BadRequestException,
+  Inject,
   Injectable,
   Logger,
   NotFoundException,
 } from "@nestjs/common";
+import {
+  BLOB_STORAGE,
+  type BlobStorageInterface,
+} from "@/blob-storage/blob-storage.interface";
 import { computeConfigHash } from "@/workflow/config-hash";
 import type { GraphWorkflowConfig } from "@/workflow/graph-workflow-types";
 import { AuditLogService } from "./audit-log.service";
@@ -53,8 +59,47 @@ export class BenchmarkRunService {
     private benchmarkTemporal: BenchmarkTemporalService,
     private datasetService: DatasetService,
     private readonly auditLogService: AuditLogService,
+    @Inject(BLOB_STORAGE)
+    private readonly blobStorage: BlobStorageInterface,
   ) {}
 
+  /**
+   * Read per-sample evaluation details (groundTruth / prediction /
+   * evaluationDetails) from blob storage. The Temporal workflow writes one
+   * JSON file per sample at `{groupId}/benchmark/runs/{runId}/{sampleId}.json`
+   * via `benchmark.persistEvaluationDetails`; we fetch and inline them here so
+   * the per-sample drill-down API contract stays stable. Returns an empty
+   * shape when the blob is missing or unreadable so the UI degrades gracefully
+   * (older runs that pre-date the blob-storage scheme have no path).
+   */
+  private async readEvaluationDetailsBlob(
+    blobPath: string | undefined,
+  ): Promise<{
+    groundTruth?: unknown;
+    prediction?: unknown;
+    evaluationDetails?: unknown;
+  }> {
+    if (!blobPath) return {};
+    try {
+      const buf = await this.blobStorage.read(validateBlobFilePath(blobPath));
+      const parsed = JSON.parse(buf.toString("utf-8")) as {
+        groundTruth?: unknown;
+        prediction?: unknown;
+        evaluationDetails?: unknown;
+      };
+      return {
+        groundTruth: parsed.groundTruth ?? undefined,
+        prediction: parsed.prediction ?? undefined,
+        evaluationDetails: parsed.evaluationDetails ?? undefined,
+      };
+    } catch (err) {
+      this.logger.warn(
+        `Failed to read evaluation details blob "${blobPath}": ${getErrorMessage(err)}`,
+      );
+      return {};
+    }
+  }
+
   /**
    * Get the current Git SHA of the worker codebase
    */
@@ -938,6 +983,13 @@ export class BenchmarkRunService {
       groundTruth?: unknown;
       prediction?: unknown;
       evaluationDetails?: unknown;
+      /**
+       * Path to the per-sample evaluation details blob written by
+       * `benchmark.persistEvaluationDetails`. Present on runs created after
+       * the blob-storage scheme landed; missing on older runs (which had the
+       * heavy fields inlined directly).
+       */
+      evaluationBlobPath?: string;
     }>;
 
     // Collect available dimensions: metadata keys plus "pass" as a synthetic dimension
@@ -995,17 +1047,36 @@ export class BenchmarkRunService {
     const offset = (page - 1) * limit;
     const paginatedResults = filteredResults.slice(offset, offset + limit);
 
-    // Map to DTO
-    const results: PerSampleResultDto[] = paginatedResults.map((result) => ({
-      sampleId: result.sampleId,
-      metadata: result.metadata || {},
-      metrics: result.metrics || {},
-      pass: result.pass ?? false,
-      diagnostics: result.diagnostics,
-      groundTruth: result.groundTruth,
-      prediction: result.prediction,
-      evaluationDetails: result.evaluationDetails,
-    }));
+    // Map to DTO. Heavy fields (groundTruth / prediction / evaluationDetails)
+    // were stripped from `metrics.perSampleResults` to keep the activity
+    // payload under Temporal's 2 MB blob limit; the workflow persists them to
+    // blob storage instead. Read each blob in parallel and inline the parsed
+    // contents so the API contract for the drill-down view stays unchanged.
+    const inlinedDetails = await Promise.all(
+      paginatedResults.map((r) =>
+        // Backwards-compat: prefer inline fields if present (older runs).
+        r.evaluationBlobPath
+          ? this.readEvaluationDetailsBlob(r.evaluationBlobPath)
+          : Promise.resolve({
+              groundTruth: r.groundTruth,
+              prediction: r.prediction,
+              evaluationDetails: r.evaluationDetails,
+            }),
+      ),
+    );
+
+    const results: PerSampleResultDto[] = paginatedResults.map(
+      (result, idx) => ({
+        sampleId: result.sampleId,
+        metadata: result.metadata || {},
+        metrics: result.metrics || {},
+        pass: result.pass ?? false,
+        diagnostics: result.diagnostics,
+        groundTruth: inlinedDetails[idx].groundTruth,
+        prediction: inlinedDetails[idx].prediction,
+        evaluationDetails: inlinedDetails[idx].evaluationDetails,
+      }),
+    );
 
     return {
       runId: run.id,

apps/temporal/src/activities.ts

@@ -47,6 +47,11 @@ export {
   benchmarkLoadOcrCache,
   benchmarkPersistOcrCache,
 } from "./activities/benchmark-ocr-cache";
+export type {
+  BenchmarkPersistEvaluationDetailsInput,
+  BenchmarkPersistEvaluationDetailsOutput,
+} from "./activities/benchmark-persist-evaluation-details";
+export { benchmarkPersistEvaluationDetails } from "./activities/benchmark-persist-evaluation-details";
 export type { BenchmarkUpdateRunStatusInput } from "./activities/benchmark-update-run";
 export { benchmarkUpdateRunStatus } from "./activities/benchmark-update-run";
 export type {

apps/temporal/src/activities/benchmark-persist-evaluation-details.ts

@@ -0,0 +1,101 @@
+/**
+ * Persist per-sample evaluation details (groundTruth, prediction,
+ * evaluationDetails, diagnostics) to blob storage as a JSON file.
+ *
+ * The drill-down UI consumes these heavy fields, but they're too large to
+ * round-trip through Temporal payloads at scale (~30 KB per sample × ~100
+ * samples = ~3 MB, exceeding the 2 MB blob limit on activity inputs).
+ * Instead, the parent workflow calls this activity once per sample (each
+ * payload safely under the limit), and stores only the returned
+ * `evaluationBlobPath` on the BenchmarkRun row.
+ *
+ * Backend reads the blob on demand when a client opens drill-down for a
+ * specific sample.
+ */
+
+import {
+  buildBlobFilePath,
+  OperationCategory,
+} from "@ai-di/blob-storage-paths";
+import { getBlobStorageClient } from "../blob-storage/blob-storage-client";
+import type { Prisma } from "../generated";
+import { createActivityLogger } from "../logger";
+import { getPrismaClient } from "./database-client";
+
+export interface BenchmarkPersistEvaluationDetailsInput {
+  runId: string;
+  sampleId: string;
+  /** Heavy fields stripped from the in-memory EvaluationResult before aggregate. */
+  details: {
+    groundTruth?: unknown;
+    prediction?: unknown;
+    evaluationDetails?: unknown;
+    diagnostics?: unknown;
+  };
+}
+
+export interface BenchmarkPersistEvaluationDetailsOutput {
+  /** Blob storage key written. Stored on the run row's perSampleResults. */
+  evaluationBlobPath: string;
+}
+
+/**
+ * Resolve the dataset/project group_id for a benchmark run via project relation.
+ * The blob path scheme requires a CUID group prefix (see buildBlobFilePath).
+ */
+async function resolveGroupId(
+  runId: string,
+  prisma: Prisma.TransactionClient | ReturnType<typeof getPrismaClient>,
+): Promise<string> {
+  const row = await prisma.benchmarkRun.findUnique({
+    where: { id: runId },
+    select: { project: { select: { group_id: true } } },
+  });
+  if (!row?.project?.group_id) {
+    throw new Error(
+      `Cannot persist evaluation details: BenchmarkRun ${runId} or its project not found`,
+    );
+  }
+  return row.project.group_id;
+}
+
+export async function benchmarkPersistEvaluationDetails(
+  input: BenchmarkPersistEvaluationDetailsInput,
+): Promise<BenchmarkPersistEvaluationDetailsOutput> {
+  const log = createActivityLogger("benchmarkPersistEvaluationDetails", {
+    runId: input.runId,
+    sampleId: input.sampleId,
+  });
+
+  const prisma = getPrismaClient();
+  const groupId = await resolveGroupId(input.runId, prisma);
+
+  const blobPath = buildBlobFilePath(
+    groupId,
+    OperationCategory.BENCHMARK,
+    ["runs", input.runId],
+    `${input.sampleId}.json`,
+  );
+
+  const blobStorage = getBlobStorageClient();
+  const data = Buffer.from(
+    JSON.stringify({
+      sampleId: input.sampleId,
+      runId: input.runId,
+      groundTruth: input.details.groundTruth ?? null,
+      prediction: input.details.prediction ?? null,
+      evaluationDetails: input.details.evaluationDetails ?? null,
+      diagnostics: input.details.diagnostics ?? null,
+    }),
+    "utf-8",
+  );
+  await blobStorage.write(blobPath, data);
+
+  log.info("benchmark evaluation details persist", {
+    event: "persist",
+    blobPath,
+    bytes: data.byteLength,
+  });
+
+  return { evaluationBlobPath: blobPath };
+}

apps/temporal/src/activities/check-ocr-confidence.test.ts

@@ -365,7 +365,7 @@ describe("checkOcrConfidence activity", () => {
     expect(result.requiresReview).toBe(false);
   });
 
-  it("skips document update when documentId is benchmark synthetic id", async () => {
+  it("forces requiresReview=false and skips DB update for benchmark documents", async () => {
     const ocrResult: OCRResult = {
       success: true,
       status: "succeeded",
@@ -413,7 +413,10 @@ describe("checkOcrConfidence activity", () => {
       threshold: 0.95,
     });
 
-    expect(result.requiresReview).toBe(true);
+    // Confidence is well below threshold but the documentId starts with
+    // "benchmark-", so the activity must short-circuit requiresReview to
+    // false (avoiding the 24h humanGate park) and never touch the DB.
+    expect(result.requiresReview).toBe(false);
     expect(prismaMock.document.update).not.toHaveBeenCalled();
   });
 });

apps/temporal/src/activities/check-ocr-confidence.ts

@@ -56,7 +56,13 @@ export async function checkOcrConfidence(params: {
     const normalizedConfidence =
       averageConfidence > 1 ? averageConfidence / 100 : averageConfidence;
 
-    const requiresReview = normalizedConfidence < confidenceThreshold;
+    // Benchmark runs must never wait for HITL: the humanReview node would
+    // park the workflow on a 24h timer waiting for an approval signal that
+    // can't arrive in benchmark mode. Force requiresReview=false so the
+    // switch routes straight to storeResults.
+    const isBenchmark = documentId.startsWith("benchmark-");
+    const requiresReview =
+      !isBenchmark && normalizedConfidence < confidenceThreshold;
 
     log.info("Check OCR confidence complete", {
       event: "complete",
@@ -69,8 +75,7 @@ export async function checkOcrConfidence(params: {
     // Update document status if review is required (skip in benchmark: no DB row exists)
     // Note: We keep status as 'ongoing_ocr' since the workflow is still in progress
     // The workflow itself tracks the 'awaiting_review' state separately
-    const isBenchmark = documentId.startsWith("benchmark-");
-    if (requiresReview && !isBenchmark) {
+    if (requiresReview) {
       const prisma = getPrismaClient();
       await prisma.document.update({
         where: { id: documentId },
@@ -84,12 +89,6 @@ export async function checkOcrConfidence(params: {
         status: "ongoing_ocr",
         requiresReview: true,
       });
-    } else if (requiresReview && isBenchmark) {
-      log.debug("Skipping document status update for benchmark run", {
-        event: "status_update_skipped",
-        documentId,
-        reason: "benchmark",
-      });
     }
 
     return {

apps/temporal/src/activity-registry.test.ts

@@ -32,6 +32,7 @@ const EXPECTED_ACTIVITY_TYPES = [
   "benchmark.loadDatasetManifest",
   "benchmark.loadOcrCache",
   "benchmark.persistOcrCache",
+  "benchmark.persistEvaluationDetails",
   "ocr.spellcheck",
   "ocr.characterConfusion",
   "ocr.normalizeFields",

apps/temporal/src/activity-registry.ts

@@ -14,6 +14,7 @@ import {
   benchmarkCompareAgainstBaseline,
   benchmarkEvaluate,
   benchmarkLoadOcrCache,
+  benchmarkPersistEvaluationDetails,
   benchmarkPersistOcrCache,
   benchmarkUpdateRunStatus,
   benchmarkWritePrediction,
@@ -342,6 +343,17 @@ register({
   description: "Persist Azure OCR poll JSON for a benchmark sample",
 });
 
+register({
+  activityType: "benchmark.persistEvaluationDetails",
+  activityFn: benchmarkPersistEvaluationDetails as (
+    ...args: unknown[]
+  ) => Promise<unknown>,
+  defaultTimeout: "30s",
+  defaultRetry: { maximumAttempts: 3 },
+  description:
+    "Persist per-sample evaluation details (groundTruth/prediction/evaluationDetails) to blob storage",
+});
+
 // -- Azure Classifier activities -------------------------------------------
 
 register({

apps/temporal/src/activity-types.ts

@@ -34,6 +34,7 @@ export const REGISTERED_ACTIVITY_TYPES = [
   "benchmark.loadDatasetManifest",
   "benchmark.loadOcrCache",
   "benchmark.persistOcrCache",
+  "benchmark.persistEvaluationDetails",
   "ocr.spellcheck",
   "ocr.characterConfusion",
   "ocr.normalizeFields",