From 09736f9e151940e7882a3db64464cf81b94bb6dc Mon Sep 17 00:00:00 2001
From: zaiun xu <xuzaikun@huaga3d.com>
Date: Fri, 10 Apr 2026 10:30:01 +0800
Subject: [PATCH] fix video label

---
 FishMeasure/fish_video_weight_evaluation.py | 133 +++++++-----
 FishMeasure/predict_weigth_from_svo2.py     |  15 +-
 README.md                                   |   5 +-
 fish_api/README.md                          |  17 +-
 fish_api/app/db.py                          | 226 ++++++++++++++------
 fish_api/app/prestart_fresh.py              |   2 +-
 fish_api/app/routers/ingest.py              |  42 ++--
 fish_api/app/services/action_watch.py       |  41 ++--
 fish_api/app/services/measure.py            | 167 ++++++++++++---
 fish_api/app/services/measure_watch.py      |  49 +++--
 fish_api/app/settings.py                    |   2 +
 fish_api/start_fresh.sh                     |  27 ++-
 fish_api/{start.sh => start_no_fresh.sh}    |  14 +-
 packaging/README.md                         |   2 +-
 packaging/bootstrap_fishserver.sh           |   2 +-
 scripts/biomass_poller.py                   |   4 +-
 scripts/run_fishserver.sh                   |  11 -
 scripts/start_fishapi_fresh.sh              |  10 -
 scripts/start_fresh.sh                      |   9 +
 scripts/start_no_fresh.sh                   |   8 +
 20 files changed, 518 insertions(+), 268 deletions(-)
 rename fish_api/{start.sh => start_no_fresh.sh} (60%)
 mode change 100644 => 100755
 delete mode 100755 scripts/run_fishserver.sh
 delete mode 100755 scripts/start_fishapi_fresh.sh
 create mode 100755 scripts/start_fresh.sh
 create mode 100755 scripts/start_no_fresh.sh

diff --git a/FishMeasure/fish_video_weight_evaluation.py b/FishMeasure/fish_video_weight_evaluation.py
index 87e033f..fd9f37b 100755
--- a/FishMeasure/fish_video_weight_evaluation.py
+++ b/FishMeasure/fish_video_weight_evaluation.py
@@ -478,8 +478,9 @@ def draw_fish_boxes_from_arrays(
     track_ids: Optional[np.ndarray],
     class_names: Dict[int, str],
     weights_by_track: Optional[Dict[int, float]] = None,
+    lengths_by_track_mm: Optional[Dict[int, float]] = None,
 ) -> np.ndarray:
-    """Draw boxes from numpy arrays; labels show DGCNN weight (g), not YOLO conf or depth."""
+    """Draw boxes from numpy arrays; labels show DGCNN weight (g) and length (mm), not YOLO conf or depth."""
     if boxes is None or len(boxes) == 0:
         return image
     for i, box in enumerate(boxes):
@@ -491,10 +492,19 @@ def draw_fish_boxes_from_arrays(
         wg: Optional[float] = None
         if weights_by_track is not None and tid >= 0 and tid in weights_by_track:
             wg = weights_by_track[tid]
+        ln_mm: Optional[float] = None
+        if lengths_by_track_mm is not None and tid >= 0 and tid in lengths_by_track_mm:
+            ln_mm = lengths_by_track_mm[tid]
         if wg is not None and np.isfinite(wg):
-            label = f"ID:{tid} {cname} weight: {wg:.0f} g"
+            extra = ""
+            if ln_mm is not None and np.isfinite(ln_mm):
+                extra = f" len:{ln_mm:.0f}mm"
+            label = f"ID:{tid} {cname} weight: {wg:.0f} g{extra}"
         else:
-            label = f"ID:{tid} {cname} weight: -- g"
+            extra = ""
+            if ln_mm is not None and np.isfinite(ln_mm):
+                extra = f" len:{ln_mm:.0f}mm"
+            label = f"ID:{tid} {cname} weight: -- g{extra}"
         (text_w, text_h), baseline = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.55, 1)
         cv2.rectangle(image, (x1, y1 - text_h - baseline - 5), (x1 + text_w, y1), (0, 255, 0), -1)
         cv2.putText(
@@ -508,6 +518,7 @@ def draw_fish_boxes_with_weight(
     results,
     class_names: Dict[int, str],
     weights_by_track: Optional[Dict[int, float]] = None,
+    lengths_by_track_mm: Optional[Dict[int, float]] = None,
 ) -> np.ndarray:
     """Draw YOLO boxes with fish weight (g) per track; no confidence, no depth."""
     if results is None or results.boxes is None:
@@ -521,11 +532,13 @@ def draw_fish_boxes_with_weight(
     tid = None
     if hasattr(results.boxes, "id") and results.boxes.id is not None:
         tid = results.boxes.id.cpu().numpy().astype(int)
-    return draw_fish_boxes_from_arrays(image, boxes, class_ids, tid, class_names, weights_by_track)
+    return draw_fish_boxes_from_arrays(
+        image, boxes, class_ids, tid, class_names, weights_by_track, lengths_by_track_mm
+    )
 
 
-def draw_overlay_header(image: np.ndarray, lines: List[str]) -> None:
-    y = 22
+def draw_overlay_header(image: np.ndarray, lines: List[str], start_y: int = 22) -> None:
+    y = start_y
     for line in lines:
         if not line:
             continue
@@ -552,9 +565,10 @@ def build_track_weights_minute_top5(
     *,
     fps: float,
     minute_interval_sec: float,
-) -> Tuple[Dict[int, float], Dict[int, float], List[float]]:
-    """track_id -> max weight g; minute_bucket -> mean weight in window; global top-5 weights (desc)."""
+) -> Tuple[Dict[int, float], Dict[int, float], Dict[int, float], List[float]]:
+    """track_id -> max weight g; track_id -> length_input (mm) at that max-weight PLY; minute_bucket -> mean g; top-5 weights (desc)."""
     tid_max: Dict[int, float] = {}
+    tid_len_mm: Dict[int, float] = {}
     bucket_vals: Dict[int, List[float]] = {}
     for it in per_file:
         ply = str(it.get("ply", ""))
@@ -563,7 +577,12 @@ def build_track_weights_minute_top5(
             continue
         tid = _parse_tid_from_ply_name(Path(ply).name)
         if tid is not None:
-            tid_max[tid] = max(tid_max.get(tid, float("-inf")), g)
+            prev = tid_max.get(tid)
+            if prev is None or g > prev:
+                tid_max[tid] = g
+                ln = float(it.get("length_input", float("nan")))
+                if np.isfinite(ln):
+                    tid_len_mm[tid] = ln
         fn = _parse_frame_num_from_ply_name(Path(ply).name)
         if fn is None or fps <= 0:
             continue
@@ -575,23 +594,20 @@ def build_track_weights_minute_top5(
         if vals:
             minute_avg[b] = float(np.mean(vals))
     top5 = sorted(tid_max.values(), reverse=True)[:5]
-    return tid_max, minute_avg, top5
+    return tid_max, tid_len_mm, minute_avg, top5
 
 
 def finalize_preview_video_with_weights(
     video_buffer: List[Dict[str, Any]],
     *,
     fps_video: float,
-    fps_timeline: float,
-    minute_interval_sec: float,
     weights_by_track: Dict[int, float],
-    minute_avg: Dict[int, float],
-    top5: List[float],
+    lengths_by_track_mm: Dict[int, float],
     class_names: Dict[int, str],
     svo_name: str,
     output_images_folder: Path,
 ) -> None:
-    """Redraw buffered frames with DGCNN weights + top-5 / per-minute lines; write side-by-side mp4."""
+    """Redraw buffered frames with DGCNN weight/length labels; write side-by-side mp4."""
     if not video_buffer:
         return
     out_frames: List[np.ndarray] = []
@@ -599,6 +615,8 @@ def finalize_preview_video_with_weights(
         left_raw = entry["left_raw"]
         right = entry["right"]
         frame_idx = int(entry["frame_idx"])
+        frame_name = entry.get("frame_name", f"frame_{frame_idx + 1:06d}")
+        num_dets = int(entry.get("num_dets", 0))
         boxes = np.asarray(entry["boxes"], dtype=np.float32)
         cls_ids = np.asarray(entry["class_ids"], dtype=np.int64)
         _tid = entry.get("track_ids")
@@ -613,20 +631,13 @@ def finalize_preview_video_with_weights(
             tids,
             class_names,
             weights_by_track=weights_by_track,
+            lengths_by_track_mm=lengths_by_track_mm,
         )
-        bucket = int((frame_idx / max(fps_timeline, 1e-6)) // minute_interval_sec)
-        mav = minute_avg.get(bucket)
-        mav_s = f"{mav:.0f} g" if mav is not None and np.isfinite(mav) else "--"
-        top5_s = ", ".join(f"{w:.0f}" for w in top5 if np.isfinite(w)) if top5 else "--"
-        lines = [
-            f"Top-5 weights (g, all fish so far): {top5_s}",
-            f"This {int(minute_interval_sec)}s window ~min {bucket + 1}: avg {mav_s}",
-        ]
-        draw_overlay_header(left_disp, lines)
-        info = f"[{frame_idx + 1}] Detections"
-        cv2.putText(
-            left_disp, info, (10, left_disp.shape[0] - 24), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 255, 0), 2, cv2.LINE_AA
-        )
+        info = f"[{frame_idx + 1}] {frame_name} | Detections: {num_dets}"
+        cv2.putText(left_disp, info, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2, cv2.LINE_AA)
+
+        cv2.putText(left_disp, "Detection", (10, left_disp.shape[0] - 20),
+                     cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2, cv2.LINE_AA)
         combined = np.hstack([left_disp, right])
         out_frames.append(combined)
     video_path = output_images_folder / f"{svo_name}_preview.mp4"
@@ -1362,7 +1373,8 @@ def process_single_svo2(svo_path, output_base, yolo_model, sam_predictor, sam_de
         fps_timeline = 30.0
     print(f"  Timeline FPS (for per-minute buckets): {fps_timeline:.2f}")
 
-    defer_video = bool(do_weight_estimation and weight_overlay_video and not save_images)
+    # 只要跑过 fish 内 DGCNN，就延迟写预览，以便叠加真实 weight/length（不再依赖 --weight-overlay-video）
+    defer_video = bool(do_weight_estimation and not save_images)
     video_frames: List[np.ndarray] = []
     video_defer_buffer: List[Dict[str, Any]] = []
     idx = 0
@@ -1400,6 +1412,8 @@ def process_single_svo2(svo_path, output_base, yolo_model, sam_predictor, sam_de
             if frame_stride > 1 and (idx % frame_stride) != 0:
                 idx += 1
                 continue
+
+            defer_left_payload: Optional[Dict[str, Any]] = None
             
             frame_name = f"frame_{idx+1:06d}"
             if idx % 30 == 0:
@@ -1464,6 +1478,18 @@ def process_single_svo2(svo_path, output_base, yolo_model, sam_predictor, sam_de
                         depth_stats_list = []
                     else:
                         active_boxes_array = np.array(active_boxes)
+                        if defer_video and len(active_boxes) > 0:
+                            cls_all_np = (
+                                results.boxes.cls.cpu().numpy().astype(int)
+                                if results.boxes.cls is not None
+                                else np.zeros(len(current_boxes), dtype=int)
+                            )
+                            ac = np.array(active_detections, dtype=int)
+                            defer_left_payload = {
+                                "boxes": active_boxes_array.astype(np.float32),
+                                "class_ids": cls_all_np[ac],
+                                "track_ids": np.array(active_track_ids, dtype=np.int64),
+                            }
                         all_masks = segment_with_sam(sam_predictor, img, active_boxes_array, sam_device)
                         individual_masks = all_masks if all_masks else []
                         
@@ -1514,34 +1540,25 @@ def process_single_svo2(svo_path, output_base, yolo_model, sam_predictor, sam_de
                 previous_boxes = None
                 depth_stats_list = []
             
-            # Left panel: DGCNN mass (g) after finalize — not YOLO conf, not depth (depth is mm to camera, not mass)
-            if defer_video and num_dets > 0 and results is not None and results.boxes is not None:
-                bx = results.boxes.xyxy.cpu().numpy()
-                cls_np = (
-                    results.boxes.cls.cpu().numpy().astype(int)
-                    if results.boxes.cls is not None
-                    else np.zeros(len(bx), dtype=int)
-                )
-                tid_np = (
-                    results.boxes.id.cpu().numpy().astype(int)
-                    if results.boxes.id is not None
-                    else np.zeros(len(bx), dtype=int)
-                )
+            # Left panel: 延迟模式用与 PLY 文件名一致的 active track_id，否则 DGCNN 字典对不上会全是 "--"
+            if defer_video and defer_left_payload is not None:
                 video_defer_buffer.append(
                     {
                         "left_raw": img.copy(),
                         "right": right_display.copy(),
                         "frame_idx": idx,
-                        "boxes": bx,
-                        "class_ids": cls_np,
-                        "track_ids": tid_np,
+                        "frame_name": frame_name,
+                        "num_dets": num_dets,
+                        **defer_left_payload,
                     }
                 )
                 left_display = img.copy()
-            else:
+            elif not defer_video:
                 left_display = draw_fish_boxes_with_weight(
                     img.copy(), results, class_names, weights_by_track=None
                 )
+            else:
+                left_display = img.copy()
             info = f"[{idx + 1}] {frame_name} | Detections: {num_dets}"
             cv2.putText(left_display, info, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2, cv2.LINE_AA)
             cv2.putText(left_display, "Detection", (10, left_display.shape[0] - 20), 
@@ -1771,26 +1788,30 @@ def process_single_svo2(svo_path, output_base, yolo_model, sam_predictor, sam_de
         if not save_images:
             if defer_video and video_defer_buffer:
                 wdict: Dict[int, float] = {}
-                mavg: Dict[int, float] = {}
-                top5: List[float] = []
+                ldict: Dict[int, float] = {}
                 wjson = output_base / "weight_estimation" / "weight_estimation_results.json"
                 if wjson.is_file():
                     try:
                         wr = json.loads(wjson.read_text(encoding="utf-8"))
-                        per = wr.get("per_file") or []
-                        wdict, mavg, top5 = build_track_weights_minute_top5(
-                            per, fps=fps_timeline, minute_interval_sec=minute_interval_sec
-                        )
+                        summary = wr.get("summary") or {}
+                        summary_wg = float(summary.get("avg_predicted_weight_g", float("nan")))
+                        summary_lmm = float(summary.get("avg_length_input_topk", float("nan")))
+                        all_tids: set = set()
+                        for _e in video_defer_buffer:
+                            _t = _e.get("track_ids")
+                            if _t is not None:
+                                all_tids.update(int(x) for x in np.asarray(_t).ravel())
+                        if np.isfinite(summary_wg):
+                            wdict = {tid: summary_wg for tid in all_tids}
+                        if np.isfinite(summary_lmm):
+                            ldict = {tid: summary_lmm for tid in all_tids}
                     except Exception as e:
                         print(f"  WARNING: Could not parse weight results for video overlay: {e}")
                 finalize_preview_video_with_weights(
                     video_defer_buffer,
                     fps_video=10.0,
-                    fps_timeline=fps_timeline,
-                    minute_interval_sec=minute_interval_sec,
                     weights_by_track=wdict,
-                    minute_avg=mavg,
-                    top5=top5,
+                    lengths_by_track_mm=ldict,
                     class_names=class_names,
                     svo_name=svo_name,
                     output_images_folder=output_images_folder,
diff --git a/FishMeasure/predict_weigth_from_svo2.py b/FishMeasure/predict_weigth_from_svo2.py
index 971b457..4cbf8c5 100755
--- a/FishMeasure/predict_weigth_from_svo2.py
+++ b/FishMeasure/predict_weigth_from_svo2.py
@@ -102,13 +102,20 @@ def _run_fish_video_evaluation_subprocess(args: argparse.Namespace, *, batch_fol
         cmd.append("--use-flatness-filter")
         cmd.extend(["--flatness-threshold", str(args.flatness_threshold)])
 
-    if getattr(args, "fish_video_weight_overlay", False):
-        wck = Path(args.weight_checkpoint).expanduser().resolve()
+    # 始终在 fish 内跑 DGCNN，生成 weight_estimation_results.json，预览视频才能叠加 weight/length；
+    # predict 后续会合并该 JSON，避免重复跑 test_dgcnn。
+    wck = Path(args.weight_checkpoint).expanduser().resolve()
+    if wck.is_file():
         cmd.extend(
             [
                 "--run-weight-estimation",
                 "--weight-estimator-checkpoint",
                 str(wck),
+            ]
+        )
+    if getattr(args, "fish_video_weight_overlay", False):
+        cmd.extend(
+            [
                 "--weight-overlay-video",
                 "--minute-interval-sec",
                 str(getattr(args, "minute_interval_sec", 60.0)),
@@ -507,8 +514,8 @@ def main() -> None:
     parser.add_argument(
         "--fish-video-weight-overlay",
         action="store_true",
-        help="Run fish_video with DGCNN + preview video overlay (fish weight g, top-5, per-window avg). "
-        "Avoids a duplicate test_dgcnn pass when weight_estimation_results.json is present.",
+        help="Extra on-video header lines (top-5 / per-minute bucket). "
+        "DGCNN in fish + preview weight/length labels are already enabled when weight checkpoint exists.",
     )
     parser.add_argument(
         "--minute-interval-sec",
diff --git a/README.md b/README.md
index 7f727c8..402394f 100644
--- a/README.md
+++ b/README.md
@@ -10,7 +10,8 @@
 | `FishMeasure/` | 双目链路：`predict_weigth_from_svo2.py`、`fish_video_weight_evaluation.py`、`weight_estimator/`、`pointcloud_classifier/` 等 |
 | `FishAction/` | 行为推断：`predict_video_x3d_3class.py`、`train_pytorchvideo_x3d.py`、`checkpoints/`（X3D） |
 | `packaging/` | **单一 Conda 环境**：网关 + 两条算法依赖定义，见 [`packaging/README.md`](packaging/README.md) |
-| `scripts/run_fishserver.sh` | 在已激活的 `fishserver` 环境中一键启动 uvicorn |
+| `scripts/start_fresh.sh` | 在已激活的 `fishserver` 环境中清空缓存后启动 uvicorn |
+| `scripts/start_no_fresh.sh` | 保留 SQLite 与推理缓存启动 uvicorn |
 
 ### 一键打包成「单环境」运行（推荐服务器）
 
@@ -19,7 +20,7 @@ bash packaging/bootstrap_fishserver.sh
 conda activate fishserver
 bash packaging/patch_cuda_torch.sh    # Linux + NVIDIA 时建议
 # 再按 packaging/README.md 安装 ZED SDK 与 pyzed
-PORT=8001 bash scripts/run_fishserver.sh
+PORT=8001 bash scripts/start_fresh.sh
 ```
 
 多 Conda 环境、分别设置 `PYTHON_FISH_MEASURE` / `PYTHON_FISH_ACTION` 的方式仍支持，见 `fish_api/README.md`。
diff --git a/fish_api/README.md b/fish_api/README.md
index 512b7b7..eada28f 100644
--- a/fish_api/README.md
+++ b/fish_api/README.md
@@ -13,7 +13,7 @@ FastAPI 网关：分块接收 **SVO2**（FishMeasure）与 **MP4**（FishAction
 - `PYTHON_FISH_MEASURE` — 运行 `predict_weigth_from_svo2.py` 的解释器路径
 - `PYTHON_FISH_ACTION` — 运行 `predict_video_x3d_3class.py` 的解释器路径
 
-**单一 Conda 环境**：若 FishMeasure 与 FishAction 已与网关停在同一个 env（例如仓库根目录 [`packaging/conda-fishserver.yaml`](../packaging/conda-fishserver.yaml) 定义的 `fishserver`），则**不要**设置上述两个变量，子进程会使用当前 `uvicorn` 的 Python。可用 [`scripts/run_fishserver.sh`](../scripts/run_fishserver.sh) 启动。
+**单一 Conda 环境**：若 FishMeasure 与 FishAction 已与网关停在同一个 env（例如仓库根目录 [`packaging/conda-fishserver.yaml`](../packaging/conda-fishserver.yaml) 定义的 `fishserver`），则**不要**设置上述两个变量，子进程会使用当前 `uvicorn` 的 Python。可用 [`scripts/start_fresh.sh`](../scripts/start_fresh.sh)（清空后启动）或 [`scripts/start_no_fresh.sh`](../scripts/start_no_fresh.sh)（保留缓存）启动。
 
 ## 配置（环境变量）
 
@@ -22,14 +22,12 @@ FastAPI 网关：分块接收 **SVO2**（FishMeasure）与 **MP4**（FishAction
 | `PUBLIC_BASE_URL` | 返回 JSON 中 `video_left` / `video_right` 的前缀（勿带末尾 `/`） | `http://127.0.0.1:8000` |
 | `INGEST_API_KEY` | 非空时，`/api/v1/ingest/*` 需请求头 `X-API-Key` | 空（不校验） |
 | `STREAM_TMP_DIR` | 分块上传临时目录 | `<repo>/fish_api/.data/ingest` |
-| `MEDIA_ROOT` | 对外托管的 `latest_left.mp4` / `latest_right.mp4` | `<repo>/fish_api/.data/media` |
+| `MEDIA_ROOT` | 对外托管每次测量生成的 `*_left.mp4` / `*_right.mp4` | `<repo>/fish_api/.data/media` |
 | `FISH_MEASURE_ROOT` | `FishMeasure` 根目录 | 自动相对仓库 |
 | `FISH_ACTION_ROOT` | `FishAction` 根目录 | 自动相对仓库 |
 | `MEASURE_OUTPUT_ROOT` | 传给 `--save-output` 的目录 | `FishMeasure/output_weight_estimator` |
 | `YOLO_MODEL` / `WEIGHT_CHECKPOINT` / `ACTION_CHECKPOINT` | 模型路径 | 与仓库内脚本默认一致 |
 | `SAM_DEVICE` | `cuda` 或 `cpu` | `cuda` |
-| `DEFAULT_FISH_SPECIES` | GET 中 `result[].type` | `大黄鱼` |
-
 可在 `fish_api/.env` 中填写上述变量（`pydantic-settings` 会读取）。
 
 ## 安装与启动
@@ -39,17 +37,18 @@ cd fish_api
 uv sync
 # 可选：包含 httpx，便于本地用 FastAPI TestClient 做冒烟测试
 # uv sync --group dev
-uv run uvicorn app.main:app --host 0.0.0.0 --port 8000
+bash start_fresh.sh    # 清空 SQLite / 缓存后启动；保留缓存用 start_no_fresh.sh
+# 或：uv run uvicorn app.main:app --host 0.0.0.0 --port 8000（需自行 prestart）
 ```
 
 OpenAPI：`http://127.0.0.1:8000/docs`
 
 ## 对外 GET（由其它系统轮询）
 
-- `GET /api/v1/biomass/real/camera/` — 双目 / 称重结果（最新一次成功快照）
-- `GET /api/v1/biomass/health/result/` — 行为 / 健康（最新一次成功快照）
+- `GET /api/v1/biomass/real/camera/` — 每次 GET 消费一条未投递的称重快照（SQLite 游标）；无新数据时 `data.result` 为空，响应头 `X-Fish-Biomass-New: 0`
+- `GET /api/v1/biomass/health/result/` — 同上，行为 / 健康快照队列
 
-失败时返回 `code: 500`，`msg` 为错误信息，`data` 为空结构。
+`result` 中每条鱼含算法字段：`id`（跟踪 ID）、`weight`（g）、`length`（mm）。不可交付或失败的推理不会进入客户端队列。
 
 ## 流式输入（分块上传）
 
@@ -82,7 +81,7 @@ MP4 将 `svo` 换成 `mp4`，本地文件换成 `clip.mp4`，轮询 `GET /api/v1
 
 ## 视频 URL
 
-FishMeasure 跑完后在输出目录查找 `*preview*.mp4`，复制到 `MEDIA_ROOT/latest_left.mp4` 与 `latest_right.mp4`（仅一个文件时左右 URL 可能相同）。确保 `PUBLIC_BASE_URL` 与前端/文档中的域名端口一致（例如 `http://192.168.3.33:8888`另起反向代理时，应把该值配成对外可达的 API 根）。
+FishMeasure 跑完后在输出目录查找 `*preview*.mp4`，复制到 `MEDIA_ROOT/`，文件名为 `{UTC时间戳}_{svo_stem}_left.mp4` / `_right.mp4`（每次测量不覆盖；仅一个预览文件时可能左右 URL 指向同一逻辑源经 SBS 拆分）。确保 `PUBLIC_BASE_URL` 与前端/文档中的域名端口一致。
 
 ## 演进建议
 
diff --git a/fish_api/app/db.py b/fish_api/app/db.py
index 2622465..eacc7fa 100644
--- a/fish_api/app/db.py
+++ b/fish_api/app/db.py
@@ -8,6 +8,7 @@ weight_prediction.json、临时 pred.json 等），由 fish_api 在子进程结
 from __future__ import annotations
 
 import json
+import math
 import shutil
 import sqlite3
 from datetime import datetime, timezone
@@ -222,6 +223,86 @@ def get_latest_health(settings: Settings) -> HealthSnapshot:
         conn.close()
 
 
+def _coerce_finite_number(v: Any) -> Optional[float]:
+    if v is None:
+        return None
+    if isinstance(v, bool):
+        return None
+    if isinstance(v, (int, float)):
+        x = float(v)
+        return x if math.isfinite(x) else None
+    if isinstance(v, str):
+        s = v.strip()
+        if not s:
+            return None
+        try:
+            x = float(s)
+            return x if math.isfinite(x) else None
+        except ValueError:
+            return None
+    return None
+
+
+def _coerce_track_id(v: Any) -> Optional[int]:
+    # bool is a subclass of int in Python
+    if isinstance(v, bool):
+        return None
+    if isinstance(v, int):
+        return v if v >= 0 else None
+    if isinstance(v, str):
+        try:
+            i = int(v.strip(), 10)
+            return i if i >= 0 else None
+        except ValueError:
+            return None
+    return None
+
+
+def measure_result_deliverable(result: Any, error: Optional[str]) -> bool:
+    """至少一条记录含有效 track id 与有限数值的 weight(g)、length(mm)。"""
+    if error:
+        return False
+    if not isinstance(result, list) or not result:
+        return False
+    for it in result:
+        if not isinstance(it, dict):
+            continue
+        tid = _coerce_track_id(it.get("id"))
+        w = _coerce_finite_number(it.get("weight"))
+        ln = _coerce_finite_number(it.get("length"))
+        if tid is not None and w is not None and ln is not None:
+            return True
+    return False
+
+
+def measure_snapshot_deliverable(snap: MeasureSnapshot) -> bool:
+    return measure_result_deliverable(snap.result, snap.error)
+
+
+def health_snapshot_deliverable(snap: HealthSnapshot) -> bool:
+    if snap.error:
+        return False
+    b = (snap.behavior_result or "").strip()
+    h = (snap.health_result or "").strip()
+    r = (snap.raw_class_en or "").strip()
+    return bool(b or h or r)
+
+
+def _health_row_deliverable(
+    behavior_result: str,
+    health_result: str,
+    raw_class_en: str,
+    error: Optional[str],
+) -> bool:
+    snap = HealthSnapshot(
+        behavior_result=behavior_result or "",
+        health_result=health_result or "",
+        raw_class_en=raw_class_en or "",
+        error=error,
+    )
+    return health_snapshot_deliverable(snap)
+
+
 def _last_delivered_id(
     conn: sqlite3.Connection, kind: str, snapshots_table: str
 ) -> int:
@@ -243,48 +324,55 @@ def _last_delivered_id(
 def pop_next_measure(
     settings: Settings,
 ) -> Tuple[MeasureSnapshot, bool, Optional[int]]:
-    """取队首未投递的 measure 快照并推进游标；无未投递时 has_new=False。"""
+    """取队首未投递且可交付的 measure 快照并推进游标；跳过不可交付行仅推进游标。"""
     init_db(settings)
     conn = _connect(settings.sqlite_path)
     try:
         conn.execute("BEGIN IMMEDIATE")
         last_id = _last_delivered_id(conn, "measure", "measure_snapshots")
 
-        next_row = conn.execute(
-            """
-            SELECT id, created_at, result_json, video_left, video_right,
-                   error, raw_prediction_path
-            FROM measure_snapshots
-            WHERE id > ?
-            ORDER BY id ASC
-            LIMIT 1
-            """,
-            (last_id,),
-        ).fetchone()
+        while True:
+            next_row = conn.execute(
+                """
+                SELECT id, created_at, result_json, video_left, video_right,
+                       error, raw_prediction_path
+                FROM measure_snapshots
+                WHERE id > ?
+                ORDER BY id ASC
+                LIMIT 1
+                """,
+                (last_id,),
+            ).fetchone()
+
+            if next_row is None:
+                conn.commit()
+                return MeasureSnapshot(result=[], video_left="", video_right=""), False, None
+
+            nid = int(next_row["id"])
+            err: Optional[str] = next_row["error"]
+            data: Any = json.loads(next_row["result_json"])
+            if not isinstance(data, list):
+                data = []
+
+            conn.execute(
+                "UPDATE delivery_cursor SET last_delivered_id = ? WHERE kind = ?",
+                (nid, "measure"),
+            )
+
+            if not measure_result_deliverable(data, err):
+                last_id = nid
+                continue
 
-        if next_row is None:
             conn.commit()
-            return MeasureSnapshot(result=[], video_left="", video_right=""), False, None
-
-        nid = int(next_row["id"])
-        conn.execute(
-            "UPDATE delivery_cursor SET last_delivered_id = ? WHERE kind = ?",
-            (nid, "measure"),
-        )
-        conn.commit()
-
-        data: Any = json.loads(next_row["result_json"])
-        if not isinstance(data, list):
-            data = []
-        snap = MeasureSnapshot(
-            result=data,
-            video_left=next_row["video_left"] or "",
-            video_right=next_row["video_right"] or "",
-            updated_at=_parse_dt(next_row["created_at"]),
-            error=next_row["error"],
-            raw_prediction_path=next_row["raw_prediction_path"],
-        )
-        return snap, True, nid
+            snap = MeasureSnapshot(
+                result=data,
+                video_left=next_row["video_left"] or "",
+                video_right=next_row["video_right"] or "",
+                updated_at=_parse_dt(next_row["created_at"]),
+                error=err,
+                raw_prediction_path=next_row["raw_prediction_path"],
+            )
+            return snap, True, nid
     except Exception:
         conn.rollback()
         raise
@@ -293,44 +381,54 @@ def pop_next_measure(
 
 
 def pop_next_health(settings: Settings) -> Tuple[HealthSnapshot, bool, Optional[int]]:
-    """取队首未投递的 health 快照并推进游标；无未投递时 has_new=False。"""
+    """取队首未投递且可交付的 health 快照并推进游标；跳过不可交付行仅推进游标。"""
     init_db(settings)
     conn = _connect(settings.sqlite_path)
     try:
         conn.execute("BEGIN IMMEDIATE")
         last_id = _last_delivered_id(conn, "health", "health_snapshots")
 
-        next_row = conn.execute(
-            """
-            SELECT id, created_at, behavior_result, health_result,
-                   raw_class_en, error
-            FROM health_snapshots
-            WHERE id > ?
-            ORDER BY id ASC
-            LIMIT 1
-            """,
-            (last_id,),
-        ).fetchone()
+        while True:
+            next_row = conn.execute(
+                """
+                SELECT id, created_at, behavior_result, health_result,
+                       raw_class_en, error
+                FROM health_snapshots
+                WHERE id > ?
+                ORDER BY id ASC
+                LIMIT 1
+                """,
+                (last_id,),
+            ).fetchone()
+
+            if next_row is None:
+                conn.commit()
+                return HealthSnapshot(behavior_result="", health_result=""), False, None
+
+            nid = int(next_row["id"])
+            beh = next_row["behavior_result"] or ""
+            hlth = next_row["health_result"] or ""
+            raw_en = next_row["raw_class_en"] or ""
+            err: Optional[str] = next_row["error"]
+
+            conn.execute(
+                "UPDATE delivery_cursor SET last_delivered_id = ? WHERE kind = ?",
+                (nid, "health"),
+            )
+
+            if not _health_row_deliverable(beh, hlth, raw_en, err):
+                last_id = nid
+                continue
 
-        if next_row is None:
             conn.commit()
-            return HealthSnapshot(behavior_result="", health_result=""), False, None
-
-        nid = int(next_row["id"])
-        conn.execute(
-            "UPDATE delivery_cursor SET last_delivered_id = ? WHERE kind = ?",
-            (nid, "health"),
-        )
-        conn.commit()
-
-        snap = HealthSnapshot(
-            behavior_result=next_row["behavior_result"] or "",
-            health_result=next_row["health_result"] or "",
-            updated_at=_parse_dt(next_row["created_at"]),
-            error=next_row["error"],
-            raw_class_en=next_row["raw_class_en"] or "",
-        )
-        return snap, True, nid
+            snap = HealthSnapshot(
+                behavior_result=beh,
+                health_result=hlth,
+                updated_at=_parse_dt(next_row["created_at"]),
+                error=err,
+                raw_class_en=raw_en,
+            )
+            return snap, True, nid
     except Exception:
         conn.rollback()
         raise
diff --git a/fish_api/app/prestart_fresh.py b/fish_api/app/prestart_fresh.py
index f68deb7..704da12 100644
--- a/fish_api/app/prestart_fresh.py
+++ b/fish_api/app/prestart_fresh.py
@@ -1,6 +1,6 @@
 """启动前清空状态：SQLite、watch 旧 JSON、测量/行为运行时目录。
 
-由 start.sh / start_fresh.sh 在 uvicorn 之前调用，使 FishMeasure 与 FishAction 均在无缓存下重新推理。
+由 start_fresh.sh 在 uvicorn 之前调用，使 FishMeasure 与 FishAction 均在无缓存下重新推理。
 """
 
 from __future__ import annotations
diff --git a/fish_api/app/routers/ingest.py b/fish_api/app/routers/ingest.py
index 39794fe..bd8ba79 100644
--- a/fish_api/app/routers/ingest.py
+++ b/fish_api/app/routers/ingest.py
@@ -5,7 +5,12 @@ from pathlib import Path
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response
 
-from app.db import save_health_snapshot, save_measure_snapshot
+from app.db import (
+    health_snapshot_deliverable,
+    measure_snapshot_deliverable,
+    save_health_snapshot,
+    save_measure_snapshot,
+)
 from app.deps import require_ingest_auth
 from app.services import action as action_svc
 from app.services import measure as measure_svc
@@ -16,7 +21,7 @@ from app.services.sessions import (
     write_chunk,
 )
 from app.settings import Settings, get_settings
-from app.state import HealthSnapshot, MeasureSnapshot, app_state
+from app.state import app_state
 
 router = APIRouter(prefix="/api/v1/ingest", tags=["ingest"])
 
@@ -28,19 +33,12 @@ async def _measure_job_serial(svo_path: Path, settings: Settings) -> None:
             snap = await asyncio.to_thread(
                 measure_svc.run_full_measure, svo_path, settings
             )
-            save_measure_snapshot(settings, snap, source_path=str(svo_path.resolve()))
+            if measure_snapshot_deliverable(snap):
+                save_measure_snapshot(
+                    settings, snap, source_path=str(svo_path.resolve())
+                )
             app_state.measure_status = "idle"
-        except Exception as e:
-            save_measure_snapshot(
-                settings,
-                MeasureSnapshot(
-                    result=[],
-                    video_left="",
-                    video_right="",
-                    error=str(e),
-                ),
-                source_path=str(svo_path.resolve()),
-            )
+        except Exception:
             app_state.measure_status = "error"
 
 
@@ -51,18 +49,12 @@ async def _action_job_serial(mp4_path: Path, settings: Settings) -> None:
             snap = await asyncio.to_thread(
                 action_svc.run_full_action, mp4_path, settings
             )
-            save_health_snapshot(settings, snap, source_path=str(mp4_path.resolve()))
+            if health_snapshot_deliverable(snap):
+                save_health_snapshot(
+                    settings, snap, source_path=str(mp4_path.resolve())
+                )
             app_state.action_status = "idle"
-        except Exception as e:
-            save_health_snapshot(
-                settings,
-                HealthSnapshot(
-                    behavior_result="",
-                    health_result="",
-                    error=str(e),
-                ),
-                source_path=str(mp4_path.resolve()),
-            )
+        except Exception:
             app_state.action_status = "error"
 
 
diff --git a/fish_api/app/services/action_watch.py b/fish_api/app/services/action_watch.py
index 26222bc..78b6b8f 100644
--- a/fish_api/app/services/action_watch.py
+++ b/fish_api/app/services/action_watch.py
@@ -6,10 +6,15 @@ from typing import Dict, Set
 
 from loguru import logger
 
-from app.db import add_watch_processed, load_watch_processed, save_health_snapshot
+from app.db import (
+    add_watch_processed,
+    health_snapshot_deliverable,
+    load_watch_processed,
+    save_health_snapshot,
+)
 from app.services import action as action_svc
 from app.settings import Settings
-from app.state import HealthSnapshot, app_state
+from app.state import app_state
 from app.watch_idle import IdleWatchWarnState, idle_warn_interval_sec, maybe_warn_idle_watch
 
 _ACTION_IDLE_WARN_INTERVAL_SEC = idle_warn_interval_sec(
@@ -52,7 +57,13 @@ async def _run_inference_and_state(
         app_state.action_status = "running"
         try:
             snap = await asyncio.to_thread(action_svc.run_full_action, mp4, settings)
-            save_health_snapshot(settings, snap, source_path=key)
+            if health_snapshot_deliverable(snap):
+                save_health_snapshot(settings, snap, source_path=key)
+            else:
+                logger.warning(
+                    "[action-watch] no deliverable health snapshot for {}, skip SQLite",
+                    mp4.name,
+                )
             app_state.action_status = "idle"
             processed.add(key)
             if settings.action_watch_use_state_file:
@@ -60,18 +71,11 @@ async def _run_inference_and_state(
             pred = (snap.raw_class_en or "").strip()
             logger.info("[action-watch] done: {} -> {}", mp4.name, pred)
         except Exception as e:
-            save_health_snapshot(
-                settings,
-                HealthSnapshot(
-                    behavior_result="",
-                    health_result="",
-                    error=str(e),
-                ),
-                source_path=key,
-            )
-            app_state.action_status = "error"
             logger.exception("[action-watch] error on {}: {}", mp4, e)
-            raise
+            app_state.action_status = "idle"
+            processed.add(key)
+            if settings.action_watch_use_state_file:
+                add_watch_processed(settings, key, "action")
 
 
 async def watch_tick(
@@ -106,12 +110,9 @@ async def watch_tick(
             stability[key] = (size, cnt + 1)
         _, cnt = stability[key]
         if cnt >= settings.action_watch_stable_polls:
-            try:
-                await _run_inference_and_state(mp4, settings, processed, state_file)
-                stability.pop(key, None)
-                did = True
-            except Exception:
-                stability[key] = (size, 1)
+            await _run_inference_and_state(mp4, settings, processed, state_file)
+            stability.pop(key, None)
+            did = True
     for k in list(stability.keys()):
         if k not in seen_keys:
             del stability[k]
diff --git a/fish_api/app/services/measure.py b/fish_api/app/services/measure.py
index 59c9606..c012625 100644
--- a/fish_api/app/services/measure.py
+++ b/fish_api/app/services/measure.py
@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import json
+import math
 import os
+import re
 import shutil
 import subprocess
 import sys
-from datetime import date, datetime, timezone
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -59,6 +61,8 @@ def _predict_weigth_from_svo2_extra_args(settings: Settings) -> List[str]:
                 str(settings.predict_minute_interval_sec),
             ]
         )
+    if not settings.measure_reuse_existing_clouds:
+        out.append("--no-reuse-existing-clouds")
     return out
 
 
@@ -106,13 +110,116 @@ def run_measure_subprocess(svo_path: Path, settings: Settings) -> None:
         )
 
 
-def _load_weight_json(svo_path: Path, settings: Settings) -> Dict[str, Any]:
+def _summary_entry_matches_svo(item: Dict[str, Any], svo_path: Path) -> bool:
     stem = svo_path.stem
-    candidate = settings.measure_output_root / stem / "weight_prediction.json"
-    if not candidate.is_file():
-        raise FileNotFoundError(f"Expected output missing: {candidate}")
-    with open(candidate, encoding="utf-8") as f:
-        return json.load(f)
+    resolved = str(svo_path.resolve())
+    svo_key = item.get("svo")
+    if svo_key:
+        try:
+            if Path(str(svo_key)).resolve() == svo_path.resolve():
+                return True
+        except OSError:
+            pass
+        if str(svo_key) == resolved:
+            return True
+    if item.get("svo_name") == stem:
+        return True
+    return False
+
+
+def _load_weight_json(svo_path: Path, settings: Settings) -> Dict[str, Any]:
+    """读取 FishMeasure 合并结果。优先 per-SVO 的 weight_prediction.json；否则从 weight_predictions_summary.json 取匹配项（predict 脚本在权重步失败时仍 exit 0 只写 summary）。"""
+    stem = svo_path.stem
+    root = settings.measure_output_root
+    candidate = root / stem / "weight_prediction.json"
+    if candidate.is_file():
+        with open(candidate, encoding="utf-8") as f:
+            return json.load(f)
+
+    summary_path = root / "weight_predictions_summary.json"
+    if summary_path.is_file():
+        with open(summary_path, encoding="utf-8") as f:
+            summary_list: Any = json.load(f)
+        if isinstance(summary_list, list):
+            for item in reversed(summary_list):
+                if not isinstance(item, dict):
+                    continue
+                if not _summary_entry_matches_svo(item, svo_path):
+                    continue
+                err = item.get("error")
+                if err:
+                    raise RuntimeError(
+                        f"FishMeasure 权重步骤失败（{svo_path.name}）: {err}"
+                    )
+                if item.get("per_cloud") or item.get("per_file") or item.get(
+                    "dgcnn_summary"
+                ):
+                    return item
+                break
+
+    combined_path = root / "weight_prediction.json"
+    if combined_path.is_file():
+        with open(combined_path, encoding="utf-8") as f:
+            combined: Any = json.load(f)
+        if isinstance(combined, dict) and combined.get("combined"):
+            names = combined.get("svo_names") or []
+            if stem in names:
+                return combined
+
+    raise FileNotFoundError(
+        f"未找到测量结果 JSON：{candidate}（且 summary 中无本条 SVO 的成功记录）"
+    )
+
+
+_TID_RE = re.compile(r"_tid(\d+)")
+
+
+def _parse_tid_from_ply_name(name: str) -> Optional[int]:
+    """与 FishMeasure/fish_video_weight_evaluation._parse_tid_from_ply_name 一致。"""
+    m = _TID_RE.search(name)
+    return int(m.group(1)) if m else None
+
+
+def _safe_media_prefix(stem: str) -> str:
+    s = re.sub(r"[^\w.\-]+", "_", stem, flags=re.UNICODE).strip("._") or "svo"
+    return s[:120]
+
+
+def _result_from_weight_prediction(data: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """按 track_id 聚合：体重取 max(predicted_weight_g)，体长取达到 max 的那条 PLY 的 length_input (mm)。"""
+    items = data.get("per_cloud") or data.get("per_file") or []
+    if not isinstance(items, list):
+        return []
+    # tid -> (max_weight_g, length_mm at max weight)
+    best: Dict[int, Tuple[float, float]] = {}
+    for it in items:
+        if not isinstance(it, dict):
+            continue
+        ply = it.get("ply")
+        if not ply:
+            continue
+        tid = _parse_tid_from_ply_name(Path(str(ply)).name)
+        if tid is None:
+            continue
+        try:
+            wg = float(it.get("predicted_weight_g", float("nan")))
+        except (TypeError, ValueError):
+            continue
+        if not math.isfinite(wg):
+            continue
+        try:
+            ln = float(it.get("length_input", float("nan")))
+        except (TypeError, ValueError):
+            ln = float("nan")
+        if tid not in best or wg > best[tid][0]:
+            best[tid] = (wg, ln)
+    out: List[Dict[str, Any]] = []
+    for tid in sorted(best.keys()):
+        wg, ln = best[tid]
+        if not math.isfinite(ln):
+            continue
+        out.append({"id": tid, "weight": wg, "length": ln})
+    return out
 
 
 def _find_preview_videos(output_dir: Path) -> Tuple[Optional[Path], Optional[Path]]:
@@ -173,10 +280,12 @@ def _publish_media(
     left: Optional[Path],
     right: Optional[Path],
     settings: Settings,
+    file_prefix: str,
 ) -> Tuple[str, str]:
     settings.media_root.mkdir(parents=True, exist_ok=True)
-    left_dst = settings.media_root / "latest_left.mp4"
-    right_dst = settings.media_root / "latest_right.mp4"
+    safe_p = _safe_media_prefix(file_prefix)
+    left_dst = settings.media_root / f"{safe_p}_left.mp4"
+    right_dst = settings.media_root / f"{safe_p}_right.mp4"
     base = settings.public_base_url.rstrip("/")
 
     if left is not None and left == right and left.is_file():
@@ -200,37 +309,39 @@ def _publish_media(
 def build_measure_snapshot(svo_path: Path, settings: Settings) -> MeasureSnapshot:
     data = _load_weight_json(svo_path, settings)
     summary = data.get("dgcnn_summary") or data.get("weight_summary") or {}
-    length_mm = summary.get("avg_length_input")
+
     weight_g = summary.get("avg_predicted_weight_g")
-    if length_mm is None:
-        length_mm = data.get("avg_length_input")
+    length_mm = summary.get("avg_length_input_topk")
     if weight_g is None:
         weight_g = data.get("avg_predicted_weight_g")
+    if length_mm is None:
+        length_mm = summary.get("avg_length_input") or data.get("avg_length_input")
 
-    today = date.today().isoformat()
-    result_item = {
-        "id": 1,
-        "type": settings.default_fish_species,
-        "length": "" if length_mm is None else str(int(round(float(length_mm)))),
-        "weight": "" if weight_g is None else str(int(round(float(weight_g)))),
-        "date": today,
-    }
+    result: List[Dict[str, Any]] = []
+    if weight_g is not None and length_mm is not None:
+        try:
+            w = float(weight_g)
+            l = float(length_mm)
+            if math.isfinite(w) and math.isfinite(l):
+                result = [{"id": 1, "weight": w, "length": l}]
+        except (TypeError, ValueError):
+            pass
 
     logger.info(
-        "[FishMeasure] parsed {}\navg_length_mm={} avg_weight_g={}\nweight_summary:\n{}",
+        "[FishMeasure] parsed {}\navg_weight_g(top5)={} avg_length_mm(top5)={}\nresult:\n{}\ndgcnn_summary:\n{}",
         svo_path.name,
-        length_mm,
         weight_g,
+        length_mm,
+        format_json_pretty(result),
         format_json_pretty(summary if summary else {}),
     )
-    logger.info(
-        "[FishMeasure] API result_item:\n{}",
-        format_json_pretty(result_item),
-    )
 
     out_dir = Path(data.get("output_dir", settings.measure_output_root / svo_path.stem))
     lv, rv = _find_preview_videos(out_dir)
-    v_left, v_right = _publish_media(lv, rv, settings)
+    prefix = (
+        f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S')}_{svo_path.stem}"
+    )
+    v_left, v_right = _publish_media(lv, rv, settings, prefix)
     logger.info(
         "[FishMeasure] media preview_paths={} {} | published_left={} published_right={}",
         lv,
@@ -240,7 +351,7 @@ def build_measure_snapshot(svo_path: Path, settings: Settings) -> MeasureSnapsho
     )
 
     return MeasureSnapshot(
-        result=[result_item],
+        result=result,
         video_left=v_left,
         video_right=v_right,
         updated_at=datetime.now(timezone.utc),
diff --git a/fish_api/app/services/measure_watch.py b/fish_api/app/services/measure_watch.py
index 0cd1c0a..e34317f 100644
--- a/fish_api/app/services/measure_watch.py
+++ b/fish_api/app/services/measure_watch.py
@@ -8,10 +8,15 @@ from typing import Dict, Set
 
 from loguru import logger
 
-from app.db import add_watch_processed, load_watch_processed, save_measure_snapshot
+from app.db import (
+    add_watch_processed,
+    load_watch_processed,
+    measure_snapshot_deliverable,
+    save_measure_snapshot,
+)
 from app.services import measure as measure_svc
 from app.settings import Settings
-from app.state import MeasureSnapshot, app_state
+from app.state import app_state
 from app.watch_idle import IdleWatchWarnState, idle_warn_interval_sec, maybe_warn_idle_watch
 
 _MEASURE_IDLE_WARN_INTERVAL_SEC = idle_warn_interval_sec(
@@ -54,7 +59,13 @@ async def _run_measure_and_state(
         app_state.measure_status = "running"
         try:
             snap = await asyncio.to_thread(measure_svc.run_full_measure, svo, settings)
-            save_measure_snapshot(settings, snap, source_path=key)
+            if measure_snapshot_deliverable(snap):
+                save_measure_snapshot(settings, snap, source_path=key)
+            else:
+                logger.warning(
+                    "[measure-watch] no deliverable measure rows for {}, skip SQLite",
+                    svo.name,
+                )
             app_state.measure_status = "idle"
             processed.add(key)
             if settings.measure_watch_use_state_file:
@@ -62,20 +73,19 @@ async def _run_measure_and_state(
             r0 = snap.result[0] if snap.result else {}
             w = r0.get("weight", "")
             logger.info("[measure-watch] done: {} weight={!r}", svo.name, w)
+        except (RuntimeError, FileNotFoundError) as e:
+            # FishMeasure 常见失败：无点云、缺 JSON 等，避免整段 traceback 刷屏
+            logger.warning("[measure-watch] measure failed for {}: {}", svo.name, e)
+            app_state.measure_status = "idle"
+            processed.add(key)
+            if settings.measure_watch_use_state_file:
+                add_watch_processed(settings, key, "measure")
         except Exception as e:
-            save_measure_snapshot(
-                settings,
-                MeasureSnapshot(
-                    result=[],
-                    video_left="",
-                    video_right="",
-                    error=str(e),
-                ),
-                source_path=key,
-            )
-            app_state.measure_status = "error"
             logger.exception("[measure-watch] error on {}: {}", svo, e)
-            raise
+            app_state.measure_status = "idle"
+            processed.add(key)
+            if settings.measure_watch_use_state_file:
+                add_watch_processed(settings, key, "measure")
 
 
 async def watch_tick(
@@ -109,12 +119,9 @@ async def watch_tick(
             stability[key] = (size, cnt + 1)
         _, cnt = stability[key]
         if cnt >= settings.measure_watch_stable_polls:
-            try:
-                await _run_measure_and_state(svo, settings, processed, state_file)
-                stability.pop(key, None)
-                did = True
-            except Exception:
-                stability[key] = (size, 1)
+            await _run_measure_and_state(svo, settings, processed, state_file)
+            stability.pop(key, None)
+            did = True
     for k in list(stability.keys()):
         if k not in seen_keys:
             del stability[k]
diff --git a/fish_api/app/settings.py b/fish_api/app/settings.py
index 7f9fd90..cb186c0 100644
--- a/fish_api/app/settings.py
+++ b/fish_api/app/settings.py
@@ -81,6 +81,8 @@ class Settings(BaseSettings):
     predict_flatness_threshold: float = 55.0
     measure_weight_top_k: int = 5
     measure_weight_top_by_length: bool = True
+    #: 为 False 时向 predict 传 --no-reuse-existing-clouds，每次强制跑 fish_video（避免误用空/陈旧 cloud；可设 True 加速重复跑同一 SVO）
+    measure_reuse_existing_clouds: bool = False
     #: 为 True 时 fish_video 内联 DGCNN + 预览叠加（更重；需 fish_video 已支持）
     predict_fish_video_weight_overlay: bool = False
     predict_minute_interval_sec: float = 60.0
diff --git a/fish_api/start_fresh.sh b/fish_api/start_fresh.sh
index 0d52803..a6fc8b6 100755
--- a/fish_api/start_fresh.sh
+++ b/fish_api/start_fresh.sh
@@ -1,9 +1,32 @@
 #!/usr/bin/env bash
-# 与 start.sh 相同（历史名称保留）：清空状态后启动 Fish API。
+# 清空 SQLite、watch 状态与测量/行为运行时目录后启动 Fish API（uvicorn）。
 #
 #   bash fish_api/start_fresh.sh
 #   PORT=8001 HOST=0.0.0.0 bash fish_api/start_fresh.sh
 #
+# 首次使用请先：cd fish_api && uv sync
+#
 set -euo pipefail
+
 DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-exec bash "$DIR/start.sh"
+cd "$DIR"
+
+export PUBLIC_BASE_URL="${PUBLIC_BASE_URL:-http://127.0.0.1:8000}"
+unset PYTHON_FISH_MEASURE PYTHON_FISH_ACTION 2>/dev/null || true
+
+if command -v uv >/dev/null 2>&1; then
+  PY=(uv run python)
+else
+  PY=(python3)
+fi
+
+"${PY[@]}" -m app.prestart_fresh
+
+PORT="${PORT:-8000}"
+HOST="${HOST:-0.0.0.0}"
+
+if command -v uv >/dev/null 2>&1; then
+  exec uv run uvicorn app.main:app --host "$HOST" --port "$PORT"
+else
+  exec uvicorn app.main:app --host "$HOST" --port "$PORT"
+fi
diff --git a/fish_api/start.sh b/fish_api/start_no_fresh.sh
old mode 100644
new mode 100755
similarity index 60%
rename from fish_api/start.sh
rename to fish_api/start_no_fresh.sh
index 6a177c0..988885d
--- a/fish_api/start.sh
+++ b/fish_api/start_no_fresh.sh
@@ -1,8 +1,8 @@
 #!/usr/bin/env bash
-# 一键启动 Fish API：先清空 SQLite、watch 状态与运行时目录（与 start_fresh.sh 相同），再启动 uvicorn。
+# 启动 Fish API（uvicorn），不执行 prestart_fresh，保留 SQLite 与 measure/action 缓存。
 #
-#   bash fish_api/start.sh
-#   PORT=8001 HOST=0.0.0.0 bash fish_api/start.sh
+#   bash fish_api/start_no_fresh.sh
+#   PORT=8001 HOST=0.0.0.0 bash fish_api/start_no_fresh.sh
 #
 # 首次使用请先：cd fish_api && uv sync
 #
@@ -14,14 +14,6 @@ cd "$DIR"
 export PUBLIC_BASE_URL="${PUBLIC_BASE_URL:-http://127.0.0.1:8000}"
 unset PYTHON_FISH_MEASURE PYTHON_FISH_ACTION 2>/dev/null || true
 
-if command -v uv >/dev/null 2>&1; then
-  PY=(uv run python)
-else
-  PY=(python3)
-fi
-
-"${PY[@]}" -m app.prestart_fresh
-
 PORT="${PORT:-8000}"
 HOST="${HOST:-0.0.0.0}"
 
diff --git a/packaging/README.md b/packaging/README.md
index 2fa9887..20641a0 100644
--- a/packaging/README.md
+++ b/packaging/README.md
@@ -89,7 +89,7 @@ pip install pyzed
 ```bash
 conda activate fishserver
 export PUBLIC_BASE_URL="http://192.168.x.x:8001"   # 与对外访问一致，影响 JSON 里 video_* URL
-PORT=8001 bash scripts/run_fishserver.sh
+PORT=8001 bash scripts/start_fresh.sh
 ```
 
 ## 5. 与「多环境」方案对比
diff --git a/packaging/bootstrap_fishserver.sh b/packaging/bootstrap_fishserver.sh
index 3f53c18..c24622b 100755
--- a/packaging/bootstrap_fishserver.sh
+++ b/packaging/bootstrap_fishserver.sh
@@ -72,5 +72,5 @@ echo "    conda activate fishserver"
 echo "  Or run Python directly:"
 echo "    $PY -c \"import sys; print(sys.executable)\""
 echo ""
-echo "  Then: PORT=8001 bash scripts/run_fishserver.sh"
+echo "  Then: PORT=8001 bash scripts/start_fresh.sh"
 echo "  (SVO) ZED SDK + pip install pyzed inside fishserver"
diff --git a/scripts/biomass_poller.py b/scripts/biomass_poller.py
index e0556de..0b75e40 100755
--- a/scripts/biomass_poller.py
+++ b/scripts/biomass_poller.py
@@ -4,8 +4,8 @@
 
 接口每次 GET 会「消费」一条未投递快照；仅当响应头 X-Fish-Biomass-New: 1（或 JSON code!=200）时打日志。
 
-  BIOMASS_API_BASE=http://127.0.0.1:8000 POLL_INTERVAL=5 \\
-    python scripts/biomass_poller.py
+  cd <仓库根目录>
+  BIOMASS_API_BASE=http://127.0.0.1:8000 POLL_INTERVAL=5 python3 scripts/biomass_poller.py
 
 依赖（与 fish_api dev 组一致）：httpx、loguru
   cd fish_api && pip install -e ".[dev]"
diff --git a/scripts/run_fishserver.sh b/scripts/run_fishserver.sh
deleted file mode 100755
index 96a2df3..0000000
--- a/scripts/run_fishserver.sh
+++ /dev/null
@@ -1,11 +0,0 @@
-#!/usr/bin/env bash
-# 仓库根目录入口：与 fish_api/start.sh 等价（启动前清空 SQLite、watch 状态、measure/action 输出目录等）
-#
-#   conda activate fishserver   # 若不用 uv
-#   export PUBLIC_BASE_URL=http://<本机对外IP>:8001
-#   PORT=8001 bash scripts/run_fishserver.sh
-#
-set -euo pipefail
-
-ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-exec bash "$ROOT/fish_api/start.sh"
diff --git a/scripts/start_fishapi_fresh.sh b/scripts/start_fishapi_fresh.sh
deleted file mode 100755
index 8874dfc..0000000
--- a/scripts/start_fishapi_fresh.sh
+++ /dev/null
@@ -1,10 +0,0 @@
-#!/usr/bin/env bash
-# 与 scripts/run_fishserver.sh 等价（历史名称保留）；均会先执行 app.prestart_fresh 再启动 uvicorn
-#
-#   bash scripts/start_fishapi_fresh.sh
-#   PORT=8001 bash scripts/start_fishapi_fresh.sh
-#
-set -euo pipefail
-
-ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-exec bash "$ROOT/fish_api/start_fresh.sh"
diff --git a/scripts/start_fresh.sh b/scripts/start_fresh.sh
new file mode 100755
index 0000000..deab4f8
--- /dev/null
+++ b/scripts/start_fresh.sh
@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+# 仓库根入口：等价于 fish_api/start_fresh.sh（先 prestart_fresh 再 uvicorn）
+#
+#   bash scripts/start_fresh.sh
+#   PORT=8001 bash scripts/start_fresh.sh
+#
+set -euo pipefail
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+exec bash "$ROOT/fish_api/start_fresh.sh"
diff --git a/scripts/start_no_fresh.sh b/scripts/start_no_fresh.sh
new file mode 100755
index 0000000..7549e03
--- /dev/null
+++ b/scripts/start_no_fresh.sh
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+# 仓库根入口：等价于 fish_api/start_no_fresh.sh（保留 SQLite 与缓存）
+#
+#   bash scripts/start_no_fresh.sh
+#
+set -euo pipefail
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+exec bash "$ROOT/fish_api/start_no_fresh.sh"