feat(data_service): per-cell loss trajectory + curve-derived shape (GetMetaData)

tests/components/test_checkpoint_workflow.py

@@ -1298,33 +1298,15 @@ def test_logger_queue_saved_with_weights(self):
         self.chkpt_manager.update_experiment_hash(force=False, dump_immediately=False)
         self.chkpt_manager.save_pending_changes(force=True)
 
-        snapshot_dir = Path(self.chkpt_manager.loggers_dir)
-        manifest_path = snapshot_dir / "loggers.manifest.json"
-        legacy_snapshot_path = snapshot_dir / "loggers.json"
-        self.assertTrue(
-            manifest_path.exists() or legacy_snapshot_path.exists(),
-            "Logger snapshot should be saved with checkpoint"
-        )
-
-        if manifest_path.exists():
-            with open(manifest_path, "r") as f:
-                manifest = json.load(f)
-            chunk_files = manifest.get("chunks", [])
-            self.assertGreaterEqual(len(chunk_files), 1, "Chunked logger snapshot should contain at least one chunk")
-            self.assertTrue(
-                all((snapshot_dir / chunk_name).exists() for chunk_name in chunk_files),
-                "All logger snapshot chunks referenced by manifest should exist"
-            )
-            self.assertTrue(self.chkpt_manager.load_logger_snapshot())
-            lg = ledgers.get_logger('main')
-            loggers = {'main': {'signal_history': lg.get_signal_history() if hasattr(lg, 'get_signal_history') else []}}
-        else:
-            with open(legacy_snapshot_path, "r") as f:
-                snapshot = json.load(f)
-            loggers = snapshot.get("loggers", {})
-
-        self.assertIn('main', loggers, "Logger entry should be present")
-        signals = loggers['main'].get("signal_history", [])
+        # Logger history is persisted to an on-disk DuckDB file alongside the
+        # checkpoint (the chunked-zstd snapshot was replaced by DuckDB in #257).
+        db_path = Path(self.chkpt_manager.get_logger_db_path())
+        self.assertTrue(db_path.exists(), "Logger DuckDB should be saved with checkpoint")
+
+        # The registered logger carries the persisted signal history.
+        lg = ledgers.get_logger('main')
+        self.assertIsNotNone(lg, "Logger entry should be present")
+        signals = lg.get_signal_history() if hasattr(lg, 'get_signal_history') else []
         if isinstance(signals, dict):
             total_signals = 0
             for experiments in signals.values():

weightslab/trainer/services/data_service.py

@@ -3453,6 +3453,48 @@ def _is_metadata_only_request(self, request) -> bool:
         except Exception:
             return False
 
+    @staticmethod
+    def _downsample_curve(vals, max_points=32):
+        """Uniformly downsample a series to <= max_points floats (endpoints kept)."""
+        if len(vals) <= max_points:
+            return [float(v) for v in vals]
+        idx = np.linspace(0, len(vals) - 1, max_points).round().astype(int)
+        return [float(vals[j]) for j in idx]
+
+    def _loss_trajectory_curves(self, sample_ids):
+        """Downsampled per-sample loss curve for the on-cell sparkline.
+
+        Returns ``{str(sample_id): [float, ...]}`` for samples with history; an
+        empty dict when there is no logger or no per-sample loss signal. Read-only —
+        shape classification lives on the write path (enable_loss_shape_signal),
+        not here. Signal name overridable via WL_LOSS_TRAJ_SIGNAL (default "loss").
+        """
+        from weightslab.backend import ledgers
+
+        logger_q = ledgers.get_logger()
+        if logger_q is None:
+            return {}
+
+        signal = os.environ.get("WL_LOSS_TRAJ_SIGNAL", "loss")
+        try:
+            rows = logger_q.query_per_sample(signal, sample_ids=sample_ids)
+        except Exception:
+            logger.debug("loss_trajectory query skipped", exc_info=True)
+            return {}
+
+        by_sid = {}
+        for sid, step, val, _ in rows:
+            by_sid.setdefault(str(sid), []).append((int(step), float(val)))
+
+        curves = {
+            sid: self._downsample_curve([v for _, v in sorted(pts)])
+            for sid, pts in by_sid.items()
+        }
+        if curves:
+            logger.info("[loss_traj] page_ids=%d with_history=%d signal=%s",
+                        len(sample_ids), len(curves), signal)
+        return curves
+
     def _build_metadata_only_response(self, df_slice: pd.DataFrame, requested_cols=None):
         """Build a DataSamplesResponse of metadata DataRecords from dataframe columns only.
 
@@ -3571,6 +3613,20 @@ def _build_metadata_only_response(self, df_slice: pd.DataFrame, requested_cols=N
                             _DataStat(name=col, type="string", shape=[1], value_string="1")
                         )
 
+        # -- Per-sample loss trajectory (the on-cell sparkline; read-only) -----
+        # Ship each visible sample's loss curve as an 'array' DataStat for the grid
+        # sparkline. The loss-SHAPE haze is NOT computed here: classification runs on
+        # the write/train path (enable_loss_shape_signal / write_loss_shapes) which
+        # maintains the tag:loss_shape column, so this path does no per-request work.
+        loss_curves = self._loss_trajectory_curves(sample_ids)
+        for i, sid in enumerate(sample_ids):
+            curve = loss_curves.get(str(sid))
+            if not curve:
+                continue
+            row_stats[i].append(
+                _DataStat(name="loss_trajectory", type="array",
+                          shape=[len(curve)], value=curve))
+
         # -- Build DataRecord list in one comprehension -----------------------
         _DataRecord = pb2.DataRecord
         data_records = [

weightslab/trainer/services/experiment_service.py

@@ -792,11 +792,11 @@ def ExperimentCommand(self, request, context):
                     checkpoint_manager = ledgers.get_checkpoint_manager()
                 except Exception:
                     checkpoint_manager = None
-            if checkpoint_manager is not None and hasattr(checkpoint_manager, "save_logger_snapshot"):
+            if checkpoint_manager is not None and hasattr(checkpoint_manager, "flush_logger_to_disk"):
                 try:
-                    checkpoint_manager.save_logger_snapshot()
+                    checkpoint_manager.flush_logger_to_disk()
                 except Exception:
-                    logger.debug("Could not persist logger snapshot after note update", exc_info=True)
+                    logger.debug("Could not persist logger history after note update", exc_info=True)
 
             # Log to audit trail
             self._log_audit(