Fix evalbench dashboard outage: render timeouts, precompute crash-loop, duplicate precompute writers

evalbench_service/supervisord_cloudrun.conf

@@ -19,6 +19,8 @@ command=python /evalbench/viewer/run_precompute.py
 directory=/evalbench/viewer
 autostart=true
 autorestart=true
+
+environment=PRECOMPUTE_WORKERS=12,PRECOMPUTE_INTERVAL=600
 stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
 stderr_logfile=/dev/stderr

viewer/main.py

@@ -7,6 +7,7 @@
 import subprocess
 import precompute_trends
 import dataset_quality
+from run_index import list_run_directories
 from summarizer import summarize_eval_scoring
 from ai_comparer import compare_evals
 
@@ -229,14 +230,7 @@ def get_color_for_pct(val_str):
 def on_load(e: me.LoadEvent):
     state = me.state(State)
     results_dir = get_results_dir()
-    directories = []
-    if os.path.exists(results_dir):
-        # List directories only
-        directories = [
-            d
-            for d in os.listdir(results_dir)
-            if os.path.isdir(os.path.join(results_dir, d))
-        ]
+    directories = list_run_directories(results_dir)
 
     job_id = me.query_params.get("job_id") or me.query_params.get("jobid")
     if job_id and job_id in directories:
@@ -259,14 +253,8 @@ def on_load(e: me.LoadEvent):
 
 def status_component():
     results_dir = get_results_dir()
-    directories = []
-    if os.path.exists(results_dir):
-        directories = [
-            d
-            for d in os.listdir(results_dir)
-            if os.path.isdir(os.path.join(results_dir, d))
-        ]
-
+    directories = list_run_directories(results_dir)
+
     with me.box(
         style=me.Style(
             background="#ffffff",
@@ -2055,15 +2043,8 @@ def render_app_content():
         results_dir = get_results_dir()
         logging.info(f"render_app_content: selected_directory='{state.selected_directory}', selected_evals='{state.selected_evals}', selected_main_tab='{state.selected_main_tab}'")
 
-        directories = []
-        if os.path.exists(results_dir):
-            # List directories only
-            directories = [
-                d
-                for d in os.listdir(results_dir)
-                if os.path.isdir(os.path.join(results_dir, d))
-            ]
-
+        directories = list_run_directories(results_dir)
+
         def on_title_click(e: me.ClickEvent):
             state.selected_directory = ""
             state.conversation_index = 0
@@ -2084,7 +2065,9 @@ def on_clear_cache_click(e: me.ClickEvent):
                     os.remove(filters_cache_file)
 
                 logging.info("Cleared precomputed files. Triggering precompute...")
-
+
+                run_index.invalidate()
+
                 import threading
                 threading.Thread(target=precompute_trends.precompute).start()
 
@@ -2289,12 +2272,22 @@ def get_val(cfg_name):
                     me.text("AI Summary", type="headline-5")
 
                     if not state.ai_summary and state.selected_directory:
+                        # The cache only carries summaries for the recent window.
+                        state.ai_summary = precompute_trends.read_ai_summary(
+                            results_dir, state.selected_directory
+                        )
+
                         trends_cache_file = os.path.join(results_dir, "trends_cache.csv")
                         if os.path.exists(trends_cache_file):
                             try:
-                                cache_df = pd.read_csv(trends_cache_file)
+                                wanted = ['job_id', 'ai_score']
+                                if not state.ai_summary:
+                                    wanted.append('ai_summary')
+                                cache_df = pd.read_csv(
+                                    trends_cache_file, usecols=lambda c: c in wanted
+                                )
                                 run_data = cache_df[cache_df['job_id'] == state.selected_directory]
-                                if not run_data.empty and 'ai_summary' in run_data.columns:
+                                if not state.ai_summary and not run_data.empty and 'ai_summary' in run_data.columns:
                                     summary = run_data['ai_summary'].values[0]
                                     if not pd.isna(summary) and summary != "N/A":
                                         state.ai_summary = summary

viewer/precompute_dataset_quality.py

@@ -13,6 +13,8 @@
 import logging
 import os
 
+from run_index import list_run_directories
+
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
 CACHE_FILENAME = "dataset_quality_cache.json"
@@ -146,8 +148,8 @@ def precompute():
             logging.warning("Could not read %s, rebuilding: %s", processed_dirs_file, e)
 
     directories = [
-        d for d in os.listdir(results_dir)
-        if os.path.isdir(os.path.join(results_dir, d)) and d not in processed
+        d for d in list_run_directories(results_dir, force=True)
+        if d not in processed
     ]
     logging.info("Dataset quality: %d new directories to scan", len(directories))
 

viewer/precompute_lease.py

@@ -0,0 +1,140 @@
+"""Single-writer lease so only one instance precomputes at a time.
+
+Supervisord starts a precompute loop in every Cloud Run instance, but they share
+one results directory and the caches are rewritten whole, so a later write
+silently discards an earlier one's rows while processed_dirs.json still records
+those runs as done -- they are then never reconsidered.
+
+Acquiring writes a holder id and timestamp, waits out the window in which a
+competing write could still land, then reads back: a whole-file rewrite has one
+winner and GCS reads are strongly consistent after it, so whoever reads its own
+id back is the sole holder. Timestamps are wall-clock because they are compared
+across hosts, where monotonic clocks are meaningless.
+"""
+
+import logging
+import os
+import socket
+import threading
+import time
+import uuid
+
+LEASE_FILENAME = "precompute.lease"
+
+# Long enough to ride out a slow FUSE write, short enough that a killed holder
+# does not stall precompute for long.
+LEASE_TTL_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_TTL", "120"))
+RENEW_INTERVAL_SECONDS = LEASE_TTL_SECONDS / 4
+
+# Read-back is only meaningful once any competing write has had time to land.
+GUARD_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_GUARD", "5"))
+
+_HOLDER_ID = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
+
+
+def _lease_path(results_dir):
+    return os.path.join(results_dir, LEASE_FILENAME)
+
+
+def _read(results_dir):
+    """Return (holder, timestamp), or (None, 0.0) if absent or unreadable."""
+    try:
+        with open(_lease_path(results_dir)) as f:
+            holder, _, timestamp = f.read().partition("\n")
+        return holder.strip(), float(timestamp.strip())
+    except (OSError, ValueError):
+        # A torn read lands here too; safe, because read-back must still pass.
+        return None, 0.0
+
+
+def _write(results_dir):
+    with open(_lease_path(results_dir), "w") as f:
+        f.write(f"{_HOLDER_ID}\n{time.time()}\n")
+
+
+class PrecomputeLease:
+    """Context manager guarding a precompute pass. Truthy only if held."""
+
+    def __init__(self, results_dir):
+        self.results_dir = results_dir
+        self.acquired = False
+        self._stop = threading.Event()
+        self._renewer = None
+
+    def acquire(self):
+        holder, timestamp = _read(self.results_dir)
+        age = time.time() - timestamp
+
+        if holder and holder != _HOLDER_ID and age < LEASE_TTL_SECONDS:
+            logging.info(
+                "Precompute lease held by %s (renewed %.0fs ago); skipping pass",
+                holder, age,
+            )
+            return False
+        if holder and holder != _HOLDER_ID:
+            logging.warning(
+                "Taking over precompute lease from %s, stale for %.0fs", holder, age
+            )
+
+        try:
+            _write(self.results_dir)
+        except OSError:
+            # Fail open: a dashboard that silently stops updating is worse than
+            # the duplicated work this lease exists to prevent.
+            logging.exception("Could not write precompute lease; running unguarded")
+            self.acquired = True
+            return True
+
+        time.sleep(GUARD_SECONDS)
+
+        winner, _ = _read(self.results_dir)
+        if winner != _HOLDER_ID:
+            logging.info("Lost precompute lease race to %s; skipping pass", winner)
+            return False
+
+        self.acquired = True
+        self._renewer = threading.Thread(
+            target=self._renew_loop, name="precompute-lease-renew", daemon=True
+        )
+        self._renewer.start()
+        logging.info("Acquired precompute lease as %s", _HOLDER_ID)
+        return True
+
+    def _renew_loop(self):
+        while not self._stop.wait(RENEW_INTERVAL_SECONDS):
+            holder, _ = _read(self.results_dir)
+            if holder and holder != _HOLDER_ID:
+                # Stand down rather than fight the successor for the file.
+                logging.warning("Precompute lease taken over by %s; stopping renewal", holder)
+                return
+            try:
+                _write(self.results_dir)
+            except OSError:
+                logging.warning("Could not renew precompute lease")
+
+    def release(self):
+        self._stop.set()
+        if self._renewer:
+            self._renewer.join(timeout=RENEW_INTERVAL_SECONDS)
+            self._renewer = None
+        if not self.acquired:
+            return
+        self.acquired = False
+        # Only clear the file if it is still ours, never a successor's.
+        holder, _ = _read(self.results_dir)
+        if holder == _HOLDER_ID:
+            try:
+                os.remove(_lease_path(self.results_dir))
+            except OSError:
+                logging.warning("Could not remove precompute lease file")
+
+    def __enter__(self):
+        self.acquire()
+        return self
+
+    def __exit__(self, *exc):
+        self.release()
+        return False
+
+    def __bool__(self):
+        return self.acquired