perf(viewer): share one mtime-keyed cache for the trends CSV and dir listing

viewer/main.py

@@ -15,7 +15,6 @@ class State:
     selected_directory: str = ""
     selected_tab: str = "Dashboard"
     conversation_index: int = 0
-    eval_summaries: str = ""
     eval_id_filter: str = ""
     product_filter: str = ""
     requester_filter: str = ""
@@ -134,6 +133,70 @@ def get_results_dir():
     return results_dir_candidates[1]  # Fallback to default
 
 
+# (mtime, value). Each is replaced as a whole so a concurrent reader never sees a
+# value that disagrees with the mtime it was built from.
+_TRENDS_DF_CACHE = (None, None)
+_RUN_DIRS_CACHE = (None, [])
+
+
+def _mtime(path):
+    try:
+        return os.path.getmtime(path)
+    except OSError:
+        return None
+
+
+def load_trends_df(results_dir):
+    """Parsed trends_cache.csv, reread only when the file changes.
+
+    Shared by every request in this worker, so callers must not mutate it.
+    """
+    global _TRENDS_DF_CACHE
+
+    cache_file = os.path.join(results_dir, "trends_cache.csv")
+    mtime = _mtime(cache_file)
+    if mtime is None:
+        logging.warning(f"Trends cache file not found at {cache_file}")
+        return None
+
+    cached_mtime, df = _TRENDS_DF_CACHE
+    if cached_mtime == mtime:
+        return df
+
+    try:
+        df = pd.read_csv(cache_file)
+    except Exception as e:
+        logging.error(f"Error reading trends cache: {e}")
+        return None
+
+    logging.info(f"Loaded {len(df)} rows from trends cache.")
+    _TRENDS_DF_CACHE = (mtime, df)
+    return df
+
+
+def list_run_dirs(results_dir):
+    """Run directory names, restatted only when results_dir gains or loses entries.
+
+    Shared by every request in this worker, so callers must not mutate it.
+    """
+    global _RUN_DIRS_CACHE
+
+    mtime = _mtime(results_dir)
+    if mtime is None:
+        return []
+
+    cached_mtime, dirs = _RUN_DIRS_CACHE
+    if cached_mtime == mtime:
+        return dirs
+
+    dirs = [
+        d for d in os.listdir(results_dir)
+        if os.path.isdir(os.path.join(results_dir, d))
+    ]
+    _RUN_DIRS_CACHE = (mtime, dirs)
+    return dirs
+
+
 def get_eval_details(results_dir, dir_name):
     details = {
         "product": "N/A",
@@ -229,14 +292,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_dirs(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 +315,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_dirs(results_dir)
+
     with me.box(
         style=me.Style(
             background="#ffffff",
@@ -306,10 +356,9 @@ def on_agent_tab_change(e):
 
         # Build summary data from precomputed trends cache
         data = []
-        cache_file = os.path.join(results_dir, "trends_cache.csv")
-        if os.path.exists(cache_file):
+        cache_df = load_trends_df(results_dir)
+        if cache_df is not None:
             try:
-                cache_df = pd.read_csv(cache_file)
                 for _, row in cache_df.iterrows():
                     data.append({
                         'AI Score': row['ai_score'] if 'ai_score' in row else None,
@@ -327,7 +376,6 @@ def on_agent_tab_change(e):
             except Exception as e:
                 logging.error(f"Error reading trends cache: {e}")
         else:
-            logging.warning(f"Trends cache file not found at {cache_file}")
             me.text("Trends cache file not found. Please run precompute.")
             return
 
@@ -537,6 +585,76 @@ def handler(e: me.ClickEvent):
             me.text("No evaluation data found in results directories.")
 
 
+# (mtime, rows). Replaced as a whole so a concurrent reader never sees rows that
+# disagree with the mtime they were built from.
+_SUMMARIES_CACHE = (None, [])
+
+
+def _pct(value):
+    return f"{value:.0f}%" if not pd.isna(value) else "N/A"
+
+
+def _build_summaries(cache_df):
+    import re
+
+    rows = []
+    for _, row in cache_df.iterrows():
+        score = row['ai_score'] if 'ai_score' in row else 0.0
+        if (score == 0.0 or pd.isna(score)) and 'ai_summary' in row and not pd.isna(row['ai_summary']):
+            match = re.search(r"General Score:.*?(\d+(\.\d+)?)", row['ai_summary'])
+            if match:
+                score = float(match.group(1))
+
+        rows.append({
+            "id": str(row['job_id']),
+            "date": str(row['run_time']) if not pd.isna(row['run_time']) else "N/A",
+            "product": str(row['product']) if not pd.isna(row['product']) else "N/A",
+            "requester": str(row['requester']) if not pd.isna(row['requester']) else "N/A",
+            "dataset": str(row['dataset']) if 'dataset' in row and not pd.isna(row['dataset']) else "N/A",
+            "model_config.generator": str(row['model_config.generator']) if 'model_config.generator' in row and not pd.isna(row['model_config.generator']) else "unknown",
+            "ai_score": f"{score:.0f}%" if not pd.isna(score) and score != 0.0 else "N/A",
+            "exact_match": _pct(row['exact_match']),
+            "llmrater": _pct(row['llmrater']),
+            "trajectory_matcher": _pct(row['trajectory']),
+            "goal_completion": _pct(row['goal_completion']) if 'goal_completion' in row else "N/A",
+            "turn_count": f"{row['turn_count']:.1f}" if not pd.isna(row['turn_count']) else "N/A",
+            "executable": _pct(row['executable']),
+            "token_consumption": f"{row['tokens']:.0f}" if not pd.isna(row['tokens']) else "N/A",
+            "end_to_end_latency": f"{row['latency'] / 60000.0:.2f}m" if not pd.isna(row['latency']) else "N/A",
+        })
+    return rows
+
+
+def load_summaries(results_dir):
+    """Row dicts for the List table, reparsed only when the trends cache changes.
+
+    The returned list is shared by every request in this worker, so callers must
+    treat it as read-only.
+    """
+    global _SUMMARIES_CACHE
+
+    mtime = _mtime(os.path.join(results_dir, "trends_cache.csv"))
+    if mtime is None:
+        return []
+
+    cached_mtime, rows = _SUMMARIES_CACHE
+    if cached_mtime == mtime:
+        return rows
+
+    cache_df = load_trends_df(results_dir)
+    if cache_df is None:
+        return []
+
+    try:
+        rows = _build_summaries(cache_df)
+    except Exception as e:
+        logging.error(f"Error building list rows from trends cache: {e}")
+        return []
+
+    _SUMMARIES_CACHE = (mtime, rows)
+    return rows
+
+
 def list_view_component(directories, results_dir):
     state = me.state(State)
     try:
@@ -597,53 +715,8 @@ def on_list_agent_tab_change(e):
         )
         me.box(style=me.Style(height="16px"))
         if directories:
-            # Compute summaries if empty
-            s = me.state(State)
-            summaries = []
-            if s.eval_summaries:
-                try:
-                    summaries = json.loads(s.eval_summaries)
-                except Exception:
-                    summaries = []
-
-            if not summaries:
-                cache_file = os.path.join(results_dir, "trends_cache.csv")
-                logging.info(f"trends_cache.csv exists: {os.path.exists(cache_file)}")
-                if os.path.exists(cache_file):
-                    try:
-                        cache_df = pd.read_csv(cache_file)
-                        logging.info(f"Loaded {len(cache_df)} rows from trends cache.")
-                        for _, row in cache_df.iterrows():
-                            score = row['ai_score'] if 'ai_score' in row else 0.0
-                            if (score == 0.0 or pd.isna(score)) and 'ai_summary' in row and not pd.isna(row['ai_summary']):
-                                import re
-                                match = re.search(r"General Score:.*?(\d+(\.\d+)?)", row['ai_summary'])
-                                if match:
-                                    score = float(match.group(1))
-
-                            summaries.append({
-                                "id": str(row['job_id']),
-                                "date": str(row['run_time']) if not pd.isna(row['run_time']) else "N/A",
-                                "product": str(row['product']) if not pd.isna(row['product']) else "N/A",
-                                "requester": str(row['requester']) if not pd.isna(row['requester']) else "N/A",
-                                "dataset": str(row['dataset']) if 'dataset' in row and not pd.isna(row['dataset']) else "N/A",
-                                "model_config.generator": str(row['model_config.generator']) if 'model_config.generator' in row and not pd.isna(row['model_config.generator']) else "unknown",
-                                "ai_score": f"{score:.0f}%" if not pd.isna(score) and score != 0.0 else "N/A",
-                                "exact_match": f"{row['exact_match']:.0f}%" if not pd.isna(row['exact_match']) else "N/A",
-                                "llmrater": f"{row['llmrater']:.0f}%" if not pd.isna(row['llmrater']) else "N/A",
-                                "trajectory_matcher": f"{row['trajectory']:.0f}%" if not pd.isna(row['trajectory']) else "N/A",
-                                "goal_completion": f"{row['goal_completion']:.0f}%" if 'goal_completion' in row and not pd.isna(row['goal_completion']) else "N/A",
-                                "turn_count": f"{row['turn_count']:.1f}" if not pd.isna(row['turn_count']) else "N/A",
-                                "executable": f"{row['executable']:.0f}%" if not pd.isna(row['executable']) else "N/A",
-                                "token_consumption": f"{row['tokens']:.0f}" if not pd.isna(row['tokens']) else "N/A",
-                                "end_to_end_latency": f"{row['latency'] / 60000.0:.2f}m" if not pd.isna(row['latency']) else "N/A"
-                            })
-                        s.eval_summaries = json.dumps(summaries)
-                    except Exception as e:
-                        logging.error(f"Error reading trends cache: {e}")
-                else:
-                    logging.warning(f"Trends cache file not found at {cache_file}")
-
+            summaries = load_summaries(results_dir)
+
             # Sort by selected column
             reverse = state.sort_descending
             col = state.sort_column
@@ -685,16 +758,9 @@ def get_sort_key(x):
                     return "" if reverse else "\xff\xff\xff\xff"
                 return str(val)
 
-            summaries.sort(key=get_sort_key, reverse=reverse)
-
-            # Extract unique values for filters from ALL summaries
-            all_summaries = []
-            if s.eval_summaries:
-                try:
-                    all_summaries = json.loads(s.eval_summaries)
-                except Exception:
-                    all_summaries = []
-
+            all_summaries = summaries
+            summaries = sorted(summaries, key=get_sort_key, reverse=reverse)
+
             filters_file = os.path.join(results_dir, "filters_cache.json")
             if os.path.exists(filters_file):
                 try:
@@ -2055,15 +2121,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_dirs(results_dir)
+
         def on_title_click(e: me.ClickEvent):
             state.selected_directory = ""
             state.conversation_index = 0
@@ -2289,10 +2348,9 @@ def get_val(cfg_name):
                     me.text("AI Summary", type="headline-5")
 
                     if not state.ai_summary and state.selected_directory:
-                        trends_cache_file = os.path.join(results_dir, "trends_cache.csv")
-                        if os.path.exists(trends_cache_file):
+                        cache_df = load_trends_df(results_dir)
+                        if cache_df is not None:
                             try:
-                                cache_df = pd.read_csv(trends_cache_file)
                                 run_data = cache_df[cache_df['job_id'] == state.selected_directory]
                                 if not run_data.empty and 'ai_summary' in run_data.columns:
                                     summary = run_data['ai_summary'].values[0]

viewer/trends.py

@@ -2,7 +2,7 @@
 import logging
 import mesop as me
 import pandas as pd
-from main import State
+from main import State, list_run_dirs, load_trends_df
 
 def get_results_dir():
     # Try to read from environment variable
@@ -24,8 +24,9 @@ def get_results_dir():
     return results_dir_candidates[1]  # Fallback to default
 
 def generate_d3_chart(df, x_col, y_col, hue_col, title, ylabel):
-    df_sorted = df.sort_values(by=x_col)
-
+    # job_id is unplotted but read by the tooltip in chart.js.
+    df_sorted = df[[x_col, y_col, hue_col, "job_id"]].sort_values(by=x_col)
+
     # Convert dataframe to records for JSON
     data_records = df_sorted.to_dict(orient='records')
     import json
@@ -89,26 +90,12 @@ def trends_component():
         me.text(f"Results directory not found at {results_dir}")
         return
 
-    cache_file = os.path.join(results_dir, "trends_cache.csv")
-
-    df = None
-
-    # Try to load from cache
-    if os.path.exists(cache_file):
-        try:
-            df = pd.read_csv(cache_file)
-            logging.info("Loaded trends data from cache.")
-        except Exception as e:
-            logging.error(f"Error reading cache file: {e}")
-
+    df = load_trends_df(results_dir)
+
     # Fallback to computing on the fly if cache is missing or failed
     if df is None:
-        directories = [
-            d
-            for d in os.listdir(results_dir)
-            if os.path.isdir(os.path.join(results_dir, d))
-        ]
-
+        directories = list_run_dirs(results_dir)
+
         data = []
 
         for d in directories:
@@ -332,7 +319,8 @@ def handler(e: me.ClickEvent):
         if state.trends_requester_filter:
             df = df[df['requester'] == state.trends_requester_filter]
 
-        df['product_dataset'] = df['product'] + " (" + df['dataset'] + ")"
+        # assign, not item-set: df may be the shared cached frame from load_trends_df.
+        df = df.assign(product_dataset=df['product'] + " (" + df['dataset'] + ")")
         df = df[df['product'].notna() & (df['product'] != 'unknown') & (df['product'].str.strip() != '')]
 
         if state.trends_agent_tab == "Gemini":