integ: opensearch-project#414 (goyamegh/run-resume-checkpoint)

Author: goyamegh

CHANGELOG.md

@@ -16,7 +16,7 @@ import React, { useState, useEffect, useCallback } from 'react';
 import { useParams, useNavigate } from 'react-router-dom';
 import {
   Loader2, CheckCircle2, XCircle, Clock, AlertTriangle,
-  ChevronDown, ChevronRight, ArrowLeft, Bookmark, RotateCcw,
+  ChevronDown, ChevronRight, ArrowLeft, Bookmark, RotateCcw, Play,
 } from 'lucide-react';
 import { Button } from '@/components/ui/button';
 import { Badge } from '@/components/ui/badge';
@@ -33,6 +33,7 @@ import {
   getEvaluationRun,
   cancelEvaluationRun,
   promoteEvaluationRun,
+  resumeEvaluationRun,
 } from '@/services/client/evaluationRunsApi';
 import { Breadcrumbs } from './Breadcrumbs';
 
@@ -68,6 +69,19 @@ function SourceBadge({ source }: { source: any }) {
   );
 }
 
+/**
+ * Client-side heuristic mirroring the server's liveness check: a 'running'
+ * run whose last liveness signal (heartbeat > resumed > created) is older
+ * than 10 minutes probably lost its server. The server re-validates against
+ * the authoritative EVALUATION_RUN_STALE_AFTER_MS on resume, so a false
+ * positive here just gets a clear 409.
+ */
+function runLooksOrphaned(run: EvaluationRun): boolean {
+  const last = new Date(run.heartbeatAt || run.resumedAt || run.createdAt || 0).getTime();
+  if (!Number.isFinite(last) || last <= 0) return true;
+  return Date.now() - last > 10 * 60 * 1000;
+}
+
 // ─── Main Component ──────────────────────────────────────────────────────────
 
 export const EvalRunDetailPage: React.FC = () => {
@@ -82,6 +96,7 @@ export const EvalRunDetailPage: React.FC = () => {
   const [promoteName, setPromoteName] = useState('');
   const [promoting, setPromoting] = useState(false);
   const [cancelling, setCancelling] = useState(false);
+  const [resuming, setResuming] = useState(false);
 
   const loadRun = useCallback(async () => {
     if (!runId) return;
@@ -119,6 +134,20 @@ export const EvalRunDetailPage: React.FC = () => {
     }
   };
 
+  // Resume: re-execute only the test cases without a persisted report.
+  // Fire-and-poll — the SSE stream runs in the background; the existing
+  // running-status poller picks up per-test-case progress.
+  const handleResume = async () => {
+    if (!runId) return;
+    setResuming(true);
+    setError(null);
+    resumeEvaluationRun(runId, () => {})
+      .catch((err: any) => setError(err.message))
+      .finally(() => loadRun());
+    // Give the server a moment to flip status to running, then refresh
+    setTimeout(() => { loadRun(); setResuming(false); }, 1000);
+  };
+
   const handlePromote = async () => {
     if (!runId || !promoteName.trim()) return;
     setPromoting(true);
@@ -211,6 +240,23 @@ export const EvalRunDetailPage: React.FC = () => {
 
             {/* Actions */}
             <div className="flex items-center gap-2">
+              {/* Resume: continue THIS run — re-executes only test cases without
+                  a persisted report (checkpoint-resume). Shown for interrupted
+                  runs (failed / cancelled) and for 'running' runs whose
+                  liveness heartbeat went silent (orphaned by a dead server —
+                  the server re-validates staleness and 409s if it's alive). */}
+              {(run.status !== 'running' || runLooksOrphaned(run)) && (run.testCaseSnapshots || []).some(s => !run.results?.[s.id]?.reportId) && (
+                <Button
+                  variant="default"
+                  size="sm"
+                  data-testid="resume-run-btn"
+                  onClick={handleResume}
+                  disabled={resuming}
+                >
+                  {resuming ? <Loader2 size={14} className="mr-1 animate-spin" /> : <Play size={14} className="mr-1" />}
+                  Resume ({(run.testCaseSnapshots || []).filter(s => !run.results?.[s.id]?.reportId).length} left)
+                </Button>
+              )}
               {/* Re-run: restart this run with its stored config (same agent,
                   test-case sources, evaluator, judge model) via the New-Run
                   flow, pre-filled. Available for any run that isn't currently

components/evals3/EvalRunDetailPage.tsx

@@ -827,6 +827,21 @@ export class FileSessionMetadataOperations implements ISessionMetadataOperations
 
 class FileEvaluationRunOperations implements IEvaluationRunOperations {
   private readonly dir: string;
+  /**
+   * Per-run write serialization. update()/updateResult() are
+   * read-modify-write with an `await` between read and write — with
+   * run.concurrency > 1 two in-flight updates could interleave at that
+   * boundary and the second write clobbers the first (lost per-test-case
+   * results). Chain writes per run id so they apply in order.
+   */
+  private writeQueues = new Map<string, Promise<unknown>>();
+
+  private serialized<T>(id: string, op: () => Promise<T>): Promise<T> {
+    const prev = this.writeQueues.get(id) ?? Promise.resolve();
+    const next = prev.then(op, op);
+    this.writeQueues.set(id, next.catch(() => {}));
+    return next;
+  }
 
   constructor(baseDir: string) {
     // Stored in same directory as benchmarks (same "index" concept)
@@ -857,11 +872,13 @@ class FileEvaluationRunOperations implements IEvaluationRunOperations {
   }
 
   async update(id: string, updates: Partial<EvaluationRun>): Promise<EvaluationRun> {
-    const existing = await this.getById(id);
-    if (!existing) throw new Error(`Evaluation run ${id} not found`);
-    const updated = { ...existing, ...updates } as EvaluationRun;
-    writeJsonFile(path.join(this.dir, `${id}.json`), updated);
-    return updated;
+    return this.serialized(id, async () => {
+      const existing = await this.getById(id);
+      if (!existing) throw new Error(`Evaluation run ${id} not found`);
+      const updated = { ...existing, ...updates } as EvaluationRun;
+      writeJsonFile(path.join(this.dir, `${id}.json`), updated);
+      return updated;
+    });
   }
 
   async delete(id: string): Promise<{ deleted: boolean }> {
@@ -912,11 +929,13 @@ class FileEvaluationRunOperations implements IEvaluationRunOperations {
     status: RunResultStatus;
     error?: string;
   }): Promise<boolean> {
-    const existing = await this.getById(runId);
-    if (!existing) return false;
-    existing.results[testCaseId] = result;
-    writeJsonFile(path.join(this.dir, `${runId}.json`), existing);
-    return true;
+    return this.serialized(runId, async () => {
+      const existing = await this.getById(runId);
+      if (!existing) return false;
+      existing.results[testCaseId] = result;
+      writeJsonFile(path.join(this.dir, `${runId}.json`), existing);
+      return true;
+    });
   }
 }
 

server/adapters/file/StorageModule.ts

@@ -1195,18 +1195,27 @@ class OpenSearchEvaluationRunOperations implements IEvaluationRunOperations {
 
   async update(id: string, updates: Partial<EvaluationRun>): Promise<EvaluationRun> {
     assertNotMigrating(this.index);
-    const existing = await this.getById(id);
-    if (!existing) throw new Error(`Evaluation run ${id} not found`);
-
-    const updated = { ...existing, ...updates };
-    await this.client.index({
-      index: this.index,
-      id,
-      body: updated,
-      refresh: 'wait_for',
-    });
+    // Partial doc-merge via the _update API instead of read-modify-write +
+    // full reindex: a stale read here would silently clobber concurrent
+    // per-test-case `updateResult` script updates (seen under concurrency>1
+    // and with the periodic run heartbeat). Only the provided fields are
+    // touched, and OpenSearch retries CAS conflicts server-side.
+    try {
+      await this.client.update({
+        index: this.index,
+        id,
+        retry_on_conflict: 10,
+        body: { doc: updates },
+        refresh: 'wait_for',
+      });
+    } catch (error: any) {
+      if (error.meta?.statusCode === 404) throw new Error(`Evaluation run ${id} not found`);
+      throw error;
+    }
 
-    return updated as EvaluationRun;
+    const updated = await this.getById(id);
+    if (!updated) throw new Error(`Evaluation run ${id} not found`);
+    return updated;
   }
 
   async delete(id: string): Promise<{ deleted: boolean }> {
@@ -1293,6 +1302,10 @@ class OpenSearchEvaluationRunOperations implements IEvaluationRunOperations {
       await this.client.update({
         index: this.index,
         id: runId,
+        // Concurrent test cases (run.concurrency > 1) update the same run
+        // doc — without CAS retries these script updates fail with
+        // version_conflict_engine_exception and the whole run aborts.
+        retry_on_conflict: 10,
         body: {
           script: {
             source: `ctx._source.results.put(params.testCaseId, params.result)`,

server/adapters/opensearch/StorageModule.ts

@@ -17,6 +17,7 @@ import { findObservioRoot, spawnObservioAgent, OBSERVIO_DEFAULT_PORT, resetObser
 import { validateAwsCredentials } from './services/tracesService.js';
 import { resumePendingTracePollsSafely } from './services/traceRecoveryOnBoot.js';
 import { recoverOrphanBenchmarkRunsSafely } from './services/benchmarkRunRecoveryOnBoot.js';
+import { recoverOrphanEvaluationRunsSafely } from './services/evaluationRunRecoveryOnBoot.js';
 import { getStorageModule } from './adapters/index.js';
 
 // Register server-side connectors (subprocess, claude-code)
@@ -170,6 +171,10 @@ async function startServer() {
             // with the runner long dead). Different bug class — see
             // server/services/benchmarkRunRecoveryOnBoot.ts.
             recoverOrphanBenchmarkRunsSafely(storage);
+            // Same failure class for run-first EvaluationRun docs — flip stale
+            // 'running' orphans to failed-with-note so they become resumable
+            // via POST /api/storage/evaluation-runs/:id/resume.
+            recoverOrphanEvaluationRunsSafely(storage);
           }
         } catch (err: any) {
           console.warn(`[bootRecovery] Could not start: ${err?.message || err}`);