fix: preserve handled display cabinet cases

Before JPEG encoding and OTA upload, paint the configured [[zones]]
polygons (yellow) and the [trash].roi (red) directly onto a copied
Frame.rgb so uploaded alarm snapshots visually carry the calibrated
regions. Normalized coordinates are clamped to image bounds, the source
frame stays untouched for downstream runtime processing, and
non-alert/disabled paths are unchanged. Adds stdlib-only polygon
fill/outline helpers plus focused unit tests.

Co-Authored-By: Claude <noreply@anthropic.com>

Dockerfile

@@ -11,6 +11,7 @@ RUN sed -i 's|http://deb.debian.org/debian|http://mirrors.aliyun.com/debian|g; s
     apt-get update && apt-get install -y --no-install-recommends \
     ca-certificates \
     ffmpeg \
+    fonts-noto-cjk \
     tzdata \
     && rm -rf /var/lib/apt/lists/*
 

README_zh.md

@@ -184,6 +184,15 @@ diagnostics_path = "logs/runtime_diagnostics.jsonl"
 [case_sink]
 path = "logs/cases.jsonl"
 
+[alarm_snapshot_upload]
+enabled = true
+service_url = "https://ota.zhengxinshipin.com"
+secret = "change-me-in-production"
+object_key_prefix = "cold-display-guard/alarms"
+connect_timeout_seconds = 5
+read_timeout_seconds = 20
+encode_timeout_seconds = 10
+
 [webhook_retry_sink]
 path = "logs/webhook_retry.jsonl"
 
@@ -191,6 +200,7 @@ path = "logs/webhook_retry.jsonl"
 enabled = true
 event_url = "https://example.com/runtime-events"
 case_url = "https://example.com/case-events"
+source_id = "cold-display-guard"
 callback_token = "shared-secret"
 connect_timeout_seconds = 3
 read_timeout_seconds = 5
@@ -206,6 +216,30 @@ retry_max_backoff_seconds = 1800
 - `logs/webhook_retry.jsonl`：Webhook 重试队列状态快照
 - `logs/webhook_delivery.jsonl`：Webhook 投递结果审计
 
+当某一轮识别结果里出现 `severity=alarm` 或 `severity=warning` 的事件时，运行时会直接复用当前检测帧：
+
+1. 用 `ffmpeg` 把当前 RGB 帧编码成 JPEG
+2. 通过 `https://ota.zhengxinshipin.com` 的 chunk-upload API 上传
+3. 把上传返回的 `object_key` 追加到对应 webhook payload
+
+相关 webhook 字段：
+
+- `event_code`：下游事件列表可直接使用的稳定编码，当前取批次 ID
+- `camera_id` / `camera_ip`：来源设备和摄像头 IP
+- `zone_id` / `zone_label`：所属区域
+- `started_at`：开始计时时间点
+- `ended_at` / `removed_at`：取出时间点
+- `dwell_seconds`：当前批次累计计时时长
+- `is_discarded` / `discarded_at`：是否已丢弃及丢弃时间点
+- `created_at`：该条外部事件记录的创建时间
+- `alerted_at` / `alarm_at`：时长告警时间点
+- `updated_at`：该条外部事件记录的最新更新时间
+- `snapshot_upload_status`：`uploaded` 或 `error`
+- `snapshot_object_key`：上传成功后的 OSS 路径
+- `snapshot_file_name`：上传文件名
+- `snapshot_captured_at`：抓帧时间
+- `snapshot_upload_error`：上传失败原因，仅失败时返回
+
 ## 本地测试
 
 ```bash

config/example.toml

@@ -5,7 +5,9 @@ timezone = "Asia/Shanghai"
 rtsp_url = ""
 
 [thresholds]
+pre_warning_seconds = 900
 max_dwell_seconds = 1200
+alarm_removal_seconds = 1800
 trash_confirmation_seconds = 120
 
 [layout]
@@ -39,6 +41,7 @@ roi = [[0.776842, 0.486901], [0.896842, 0.522456], [0.841053, 0.857427], [0.7168
 sample_stride_pixels = 4
 occupancy_mean_delta = 55.0
 occupancy_dark_luma_threshold = 80.0
+occupancy_absolute_dark_fraction = 0.0
 occupancy_dark_fraction = 0.06
 occupancy_texture_dark_fraction = 0.04
 occupancy_bright_luma_threshold = 220.0
@@ -58,6 +61,15 @@ path = "logs/events.jsonl"
 [case_sink]
 path = "logs/cases.jsonl"
 
+[alarm_snapshot_upload]
+enabled = true
+service_url = "https://ota.zhengxinshipin.com"
+secret = "change-me-in-production"
+object_key_prefix = "cold-display-guard/alarms"
+connect_timeout_seconds = 5
+read_timeout_seconds = 20
+encode_timeout_seconds = 10
+
 [webhook_retry_sink]
 path = "logs/webhook_retry.jsonl"
 
@@ -68,6 +80,7 @@ path = "logs/webhook_delivery.jsonl"
 enabled = false
 event_url = ""
 case_url = ""
+source_id = ""
 callback_token = ""
 connect_timeout_seconds = 3
 read_timeout_seconds = 5

docs/superpowers/plans/2026-06-09-alarm-snapshot-upload.md

@@ -0,0 +1,111 @@
+# Alarm Snapshot Upload Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Capture one current-frame snapshot for alerting runtime events, upload it to the OTA chunk-upload service, and include the returned path in outbound webhook payloads.
+
+**Architecture:** Keep `BatchEngine` unchanged and treat snapshot upload as runtime-side enrichment. Reuse the already captured RGB frame from the active detection loop, encode it to JPEG with `ffmpeg`, upload it through the documented token/init/chunk/complete flow, then merge the returned `object_key` into the webhook payload for alert-level batch events and the derived case events from the same cycle.
+
+**Tech Stack:** Python 3.12 standard library backend, existing `ffmpeg` dependency, JSONL webhook retry flow, unittest.
+
+---
+
+### Task 1: Snapshot Upload Client
+
+**Files:**
+- Create: `src/cold_display_guard/alarm_snapshots.py`
+- Test: `tests/test_alarm_snapshots.py`
+
+- [ ] **Step 1: Write the failing test**
+  Add tests that cover:
+  - loading upload settings from config
+  - encoding a current RGB frame into JPEG via injected encoder helper
+  - successful OTA upload flow returning `object_key`
+  - disabled or non-alert events skipping upload
+
+- [ ] **Step 2: Run test to verify it fails**
+  Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_alarm_snapshots.py -v`
+  Expected: FAIL because the snapshot upload module does not exist yet.
+
+- [ ] **Step 3: Write minimal implementation**
+  Implement:
+  - upload settings parsing with defaults for `https://ota.zhengxinshipin.com` and secret `change-me-in-production`
+  - current-frame JPEG encoding
+  - token/init/chunk/complete upload workflow with injectable HTTP helpers for tests
+  - per-cycle alert snapshot metadata structure carrying `object_key`, file name, and upload status
+
+- [ ] **Step 4: Run test to verify it passes**
+  Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_alarm_snapshots.py -v`
+  Expected: PASS
+
+### Task 2: Runtime And Webhook Integration
+
+**Files:**
+- Modify: `src/cold_display_guard/main.py`
+- Modify: `src/cold_display_guard/webhooks.py`
+- Test: `tests/test_main.py`
+- Test: `tests/test_webhooks.py`
+
+- [ ] **Step 1: Write the failing test**
+  Add tests that cover:
+  - runtime uploads one snapshot when a cycle contains alert-severity events
+  - webhook payload includes uploaded `object_key` for alert batch events
+  - derived case webhook payload includes the same snapshot path for matching case-creation events
+  - upload failure does not block webhook delivery and instead records failure metadata in payload
+
+- [ ] **Step 2: Run test to verify it fails**
+  Run:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_main.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py -v`
+  Expected: FAIL because runtime/webhook code does not accept snapshot metadata yet.
+
+- [ ] **Step 3: Write minimal implementation**
+  Implement:
+  - alert event selection based on event severity
+  - one-per-cycle snapshot upload using the current frame
+  - payload enrichment for batch-event and matching case-event webhooks
+  - retry queue persistence of the already enriched payload
+
+- [ ] **Step 4: Run test to verify it passes**
+  Run:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_main.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py -v`
+  Expected: PASS
+
+### Task 3: Config, Secrets, Docs, And Final Verification
+
+**Files:**
+- Modify: `src/cold_display_guard/config.py`
+- Modify: `src/cold_display_guard/manage_api.py`
+- Modify: `config/example.toml`
+- Modify: `README_zh.md`
+- Test: `tests/test_config.py`
+- Test: `tests/test_manage_api.py`
+
+- [ ] **Step 1: Write the failing test**
+  Extend tests so:
+  - config formatting writes snapshot-upload settings
+  - management config payload strips sensitive upload secret
+
+- [ ] **Step 2: Run test to verify it fails**
+  Run:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_config.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
+  Expected: FAIL because snapshot upload settings are not exposed/formatted yet.
+
+- [ ] **Step 3: Write minimal implementation**
+  Implement:
+  - config keys for snapshot upload URL, secret, object prefix, enable flag, and timeout/chunk settings
+  - config payload secret stripping
+  - README updates for alert snapshot upload behavior and returned webhook fields
+
+- [ ] **Step 4: Run targeted and full verification**
+  Run:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_alarm_snapshots.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_main.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_config.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest discover -s tests -v`
+  - `cd web && pnpm build`
+  Expected: PASS

src/cold_display_guard/cases.py

@@ -8,15 +8,19 @@ from typing import Any
 
 
 EVENT_CASE_TYPES = {
+    "time_pre_warning": "pre_warning",
     "time_alarm": "time_alarm",
     "batch_pending_disposal": "pending_disposal",
+    "alarm_removal_timeout": "alarm_removal_timeout",
     "warning_escalated": "warning_escalated",
 }
 
 CASE_PRIORITY = {
-    "time_alarm": 1,
-    "pending_disposal": 2,
-    "warning_escalated": 3,
+    "pre_warning": 1,
+    "time_alarm": 2,
+    "pending_disposal": 3,
+    "alarm_removal_timeout": 4,
+    "warning_escalated": 5,
 }
 
 
@@ -139,10 +143,11 @@ class CaseStore:
 
         case_id = build_case_id(batch_id)
         existing = self._cases.get(case_id)
-        if event_name == "batch_discarded":
+        if event_name in {"batch_discarded", "pre_warning_handled"}:
             if existing is None or existing.case_status == "handled":
                 return None
-            return self._close_case(existing, when, handled_source="auto_closed")
+            handled_source = "auto_removed_before_alarm" if event_name == "pre_warning_handled" else "auto_closed"
+            return self._close_case(existing, when, handled_source=handled_source)
 
         case_type = EVENT_CASE_TYPES.get(event_name)
         if case_type is None:

src/cold_display_guard/config.py

@@ -23,7 +23,9 @@ def load_settings(path: str | Path) -> EngineSettings:
 
     return EngineSettings(
         camera_id=str(data.get("camera_id", "cold_display_cam_01")),
+        pre_warning_seconds=int(thresholds.get("pre_warning_seconds", 0)),
         max_dwell_seconds=int(thresholds.get("max_dwell_seconds", 10_800)),
+        alarm_removal_seconds=int(thresholds.get("alarm_removal_seconds", 0)),
         trash_confirmation_seconds=int(thresholds.get("trash_confirmation_seconds", 120)),
         zone_ids=zone_ids,
     )
@@ -135,7 +137,9 @@ def format_config_document(data: dict[str, Any]) -> str:
 
     thresholds = data.get("thresholds", {})
     lines.append("[thresholds]")
+    lines.append(f'pre_warning_seconds = {int(thresholds.get("pre_warning_seconds", 0))}')
     lines.append(f'max_dwell_seconds = {int(thresholds.get("max_dwell_seconds", 10_800))}')
+    lines.append(f'alarm_removal_seconds = {int(thresholds.get("alarm_removal_seconds", 0))}')
     lines.append(f'trash_confirmation_seconds = {int(thresholds.get("trash_confirmation_seconds", 120))}')
     lines.append("")
 
@@ -199,6 +203,29 @@ def format_config_document(data: dict[str, Any]) -> str:
         lines.append(f'path = "{_escape(str(case_sink.get("path", "logs/cases.jsonl")))}"')
         lines.append("")
 
+    alarm_snapshot_upload = data.get("alarm_snapshot_upload", {})
+    if alarm_snapshot_upload:
+        lines.append("[alarm_snapshot_upload]")
+        for key in (
+            "connect_timeout_seconds",
+            "encode_timeout_seconds",
+            "enabled",
+            "object_key_prefix",
+            "read_timeout_seconds",
+            "secret",
+            "service_url",
+        ):
+            if key not in alarm_snapshot_upload:
+                continue
+            value = alarm_snapshot_upload[key]
+            if isinstance(value, bool):
+                lines.append(f"{key} = {str(value).lower()}")
+            elif isinstance(value, int | float):
+                lines.append(f"{key} = {value}")
+            else:
+                lines.append(f'{key} = "{_escape(str(value))}"')
+        lines.append("")
+
     webhook_retry_sink = data.get("webhook_retry_sink", {})
     if webhook_retry_sink:
         lines.append("[webhook_retry_sink]")
@@ -219,6 +246,7 @@ def format_config_document(data: dict[str, Any]) -> str:
             "retry_batch_limit",
             "retry_max_attempts",
             "retry_max_backoff_seconds",
+            "source_id",
         ):
             if key not in webhooks:
                 continue

src/cold_display_guard/engine.py

@@ -34,7 +34,9 @@ class BatchEngine:
         if appeared_zones and self.pending_disposal:
             events.extend(self._mark_pending_as_returned(observation.ts, appeared_zones))
 
+        events.extend(self._apply_pre_warnings(observation.ts, previous_zone_counts))
         events.extend(self._apply_time_alarms(observation.ts, previous_zone_counts))
+        events.extend(self._apply_alarm_removal_timeouts(observation.ts))
 
         pending_count_before_zone_transitions = len(self.pending_disposal)
         for zone_id, new_count in zone_counts.items():
@@ -99,7 +101,14 @@ class BatchEngine:
             if zone_id not in self._zone_counts:
                 continue
             event_name = str(event.get("event", ""))
-            if event_name in {"batch_started", "batch_count_changed", "mixed_batch_violation", "time_alarm"}:
+            if event_name in {
+                "batch_started",
+                "batch_count_changed",
+                "mixed_batch_violation",
+                "time_pre_warning",
+                "time_alarm",
+                "alarm_removal_timeout",
+            }:
                 if active_zone_counts is not None and active_counts.get(zone_id, 0) <= 0:
                     self.active_by_zone.pop(zone_id, None)
                     self._zone_counts[zone_id] = 0
@@ -131,9 +140,19 @@ class BatchEngine:
             last_count=max(1, int(event.get("current_count", 1) or 1)),
             state=str(event.get("state", "active") or "active"),
         )
+        batch.pre_warned_at = parse_event_datetime(event.get("pre_warned_at"))
         batch.alerted_at = parse_event_datetime(event.get("alerted_at"))
         if batch.alerted_at is not None:
             batch.state = "alerted"
+            if self.settings.alarm_removal_seconds > 0:
+                batch.alarm_removal_deadline = parse_event_datetime(event.get("alarm_removal_deadline"))
+                if batch.alarm_removal_deadline is None:
+                    batch.alarm_removal_deadline = batch.alerted_at + self.settings.alarm_removal_window
+        elif batch.pre_warned_at is not None:
+            batch.state = "pre_warning"
+        batch.alarm_removal_timed_out_at = parse_event_datetime(event.get("alarm_removal_timed_out_at"))
+        if batch.alarm_removal_timed_out_at is not None:
+            batch.state = "alarm_removal_timeout"
         batch.dwell_seconds = max(0, int(event.get("dwell_seconds", 0) or 0))
         return batch
 
@@ -162,10 +181,44 @@ class BatchEngine:
             self.pending_disposal.append(batch)
             return self._event("batch_pending_disposal", when, batch, severity="warning")
 
+        if batch.pre_warned_at is not None:
+            batch.state = "handled"
+            self.closed_batches.append(batch)
+            return self._event(
+                "pre_warning_handled",
+                when,
+                batch,
+                severity="info",
+                handled_source="auto_removed_before_alarm",
+            )
+
         batch.state = "consumed"
         self.closed_batches.append(batch)
         return self._event("batch_consumed", when, batch, severity="info")
 
+    def _apply_pre_warnings(self, when: datetime, zone_counts: dict[str, int]) -> list[dict[str, Any]]:
+        if self.settings.pre_warning_seconds <= 0:
+            return []
+        events: list[dict[str, Any]] = []
+        for zone_id, batch in self.active_by_zone.items():
+            if batch.pre_warned_at is not None or batch.alerted_at is not None:
+                continue
+            dwell_seconds = batch.current_dwell_seconds(when)
+            if dwell_seconds < self.settings.pre_warning_seconds:
+                continue
+            batch.state = "pre_warning"
+            batch.pre_warned_at = when
+            events.append(
+                self._event(
+                    "time_pre_warning",
+                    when,
+                    batch,
+                    severity="warning",
+                    current_count=zone_counts.get(zone_id, batch.last_count),
+                )
+            )
+        return events
+
     def _apply_time_alarms(self, when: datetime, zone_counts: dict[str, int]) -> list[dict[str, Any]]:
         events: list[dict[str, Any]] = []
         for zone_id, batch in self.active_by_zone.items():
@@ -176,6 +229,8 @@ class BatchEngine:
                 continue
             batch.state = "alerted"
             batch.alerted_at = when
+            if self.settings.alarm_removal_seconds > 0:
+                batch.alarm_removal_deadline = when + self.settings.alarm_removal_window
             events.append(
                 self._event(
                     "time_alarm",
@@ -187,6 +242,30 @@ class BatchEngine:
             )
         return events
 
+    def _apply_alarm_removal_timeouts(self, when: datetime) -> list[dict[str, Any]]:
+        if self.settings.alarm_removal_seconds <= 0:
+            return []
+        events: list[dict[str, Any]] = []
+        for batch in self.active_by_zone.values():
+            if batch.alerted_at is None or batch.alarm_removal_deadline is None:
+                continue
+            if batch.alarm_removal_timed_out_at is not None:
+                continue
+            if when <= batch.alarm_removal_deadline:
+                continue
+            batch.state = "alarm_removal_timeout"
+            batch.alarm_removal_timed_out_at = when
+            events.append(
+                self._event(
+                    "alarm_removal_timeout",
+                    when,
+                    batch,
+                    severity="alarm",
+                    reason="alarmed_batch_not_removed_after_alarm_window",
+                )
+            )
+        return events
+
     def _mark_mixed_batch(
         self,
         zone_id: str,
@@ -273,13 +352,21 @@ class BatchEngine:
             "state": batch.state,
             "started_at": batch.started_at.isoformat(),
             "dwell_seconds": batch.current_dwell_seconds(when),
+            "pre_warning_seconds": self.settings.pre_warning_seconds,
             "max_dwell_seconds": self.settings.max_dwell_seconds,
+            "alarm_removal_seconds": self.settings.alarm_removal_seconds,
         }
         zone_index = self._zone_index(batch.zone_id)
         if zone_index is not None:
             payload["zone_index"] = zone_index
+        if batch.pre_warned_at is not None:
+            payload["pre_warned_at"] = batch.pre_warned_at.isoformat()
         if batch.alerted_at is not None:
             payload["alerted_at"] = batch.alerted_at.isoformat()
+        if batch.alarm_removal_deadline is not None:
+            payload["alarm_removal_deadline"] = batch.alarm_removal_deadline.isoformat()
+        if batch.alarm_removal_timed_out_at is not None:
+            payload["alarm_removal_timed_out_at"] = batch.alarm_removal_timed_out_at.isoformat()
         if batch.ended_at is not None:
             payload["ended_at"] = batch.ended_at.isoformat()
         if batch.disposal_deadline is not None:
@@ -290,9 +377,9 @@ class BatchEngine:
         return payload
 
     def _event_severity(self, event_name: str) -> str:
-        if event_name == "time_alarm":
+        if event_name in {"time_alarm", "alarm_removal_timeout"}:
             return "alarm"
-        if event_name in {"warning_escalated", "batch_pending_disposal"}:
+        if event_name in {"warning_escalated", "batch_pending_disposal", "time_pre_warning"}:
             return "warning"
         if event_name.endswith("_violation"):
             return "warning"

src/cold_display_guard/main.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from pathlib import Path
 from zoneinfo import ZoneInfo
 
+from cold_display_guard.alarm_snapshots import capture_alert_snapshot
 from cold_display_guard.cases import CaseStore, append_case_snapshots, load_case_snapshots
 from cold_display_guard.config import load_config_document, load_settings, resolve_config_path, resolve_project_root
 from cold_display_guard.engine import BatchEngine
@@ -104,6 +105,7 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
             events = engine.process(observation)
             append_jsonl(event_path, events)
             case_snapshots = persist_case_updates(case_store, case_path, events)
+            snapshot_upload = capture_runtime_alarm_snapshot(frame, events, config, now=when)
             deliver_runtime_webhooks(
                 events,
                 case_snapshots,
@@ -111,6 +113,7 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
                 webhook_delivery_path,
                 retry_path=webhook_retry_path,
                 now=when,
+                snapshot_upload=snapshot_upload,
             )
             append_jsonl(
                 diagnostics_path,
@@ -168,11 +171,31 @@ def load_case_store(path: Path) -> CaseStore:
 
 
 def persist_case_updates(case_store: CaseStore, path: Path, events: list[dict[str, object]]) -> list[dict[str, object]]:
+    case_store = load_case_store(path)
     snapshots = case_store.apply_batch_events(events)
     append_case_snapshots(path, snapshots)
     return snapshots
 
 
+def capture_runtime_alarm_snapshot(
+    frame,
+    events: list[dict[str, object]],
+    config: dict[str, object],
+    *,
+    now: datetime | None = None,
+    jpeg_encoder=None,
+    uploader=None,
+) -> dict[str, object] | None:
+    return capture_alert_snapshot(
+        frame,
+        events,
+        config,
+        now=now,
+        jpeg_encoder=jpeg_encoder,
+        uploader=uploader,
+    )
+
+
 def deliver_runtime_webhooks(
     events: list[dict[str, object]],
     case_snapshots: list[dict[str, object]],
@@ -182,9 +205,26 @@ def deliver_runtime_webhooks(
     retry_path: Path | None = None,
     http_post=None,
     now: datetime | None = None,
+    snapshot_upload: dict[str, object] | None = None,
 ) -> None:
-    send_batch_event_webhooks(events, config, audit_path, retry_path=retry_path, http_post=http_post, now=now)
-    send_case_webhooks(case_snapshots, config, audit_path, retry_path=retry_path, http_post=http_post, now=now)
+    send_batch_event_webhooks(
+        events,
+        config,
+        audit_path,
+        retry_path=retry_path,
+        http_post=http_post,
+        now=now,
+        snapshot_upload=snapshot_upload,
+    )
+    send_case_webhooks(
+        case_snapshots,
+        config,
+        audit_path,
+        retry_path=retry_path,
+        http_post=http_post,
+        now=now,
+        snapshot_upload=snapshot_upload,
+    )
     if retry_path is not None:
         drain_webhook_retries(config, retry_path, audit_path, http_post=http_post, now=now)
 

src/cold_display_guard/manage_api.py

@@ -311,6 +311,8 @@ def config_payload(ctx: ManageContext) -> dict[str, Any]:
     event_path = event_sink_path(ctx, data)
     case_path = case_sink_path(ctx, data)
     retry_path = webhook_retry_sink_path(ctx, data)
+    alarm_snapshot_upload = dict(data.get("alarm_snapshot_upload", {}) or {})
+    alarm_snapshot_upload.pop("secret", None)
     webhooks = dict(data.get("webhooks", {}) or {})
     webhooks.pop("callback_token", None)
     return {
@@ -329,6 +331,7 @@ def config_payload(ctx: ManageContext) -> dict[str, Any]:
         "event_sink": {"path": str(event_path)},
         "case_sink": {"path": str(case_path)},
         "webhook_retry_sink": {"path": str(retry_path)},
+        "alarm_snapshot_upload": alarm_snapshot_upload,
         "webhooks": webhooks,
     }
 

src/cold_display_guard/models.py

@@ -11,14 +11,24 @@ DEFAULT_ZONE_IDS = tuple(f"r{row}c{col}" for row in range(1, 3) for col in range
 @dataclass(frozen=True, slots=True)
 class EngineSettings:
     camera_id: str = "cold_display_cam_01"
+    pre_warning_seconds: int = 0
     max_dwell_seconds: int = 10_800
+    alarm_removal_seconds: int = 0
     trash_confirmation_seconds: int = 120
     zone_ids: tuple[str, ...] = DEFAULT_ZONE_IDS
 
+    @property
+    def pre_warning(self) -> timedelta:
+        return timedelta(seconds=self.pre_warning_seconds)
+
     @property
     def max_dwell(self) -> timedelta:
         return timedelta(seconds=self.max_dwell_seconds)
 
+    @property
+    def alarm_removal_window(self) -> timedelta:
+        return timedelta(seconds=self.alarm_removal_seconds)
+
     @property
     def trash_confirmation_window(self) -> timedelta:
         return timedelta(seconds=self.trash_confirmation_seconds)
@@ -56,7 +66,10 @@ class Batch:
     started_at: datetime
     last_count: int
     state: str = "active"
+    pre_warned_at: datetime | None = None
     alerted_at: datetime | None = None
+    alarm_removal_deadline: datetime | None = None
+    alarm_removal_timed_out_at: datetime | None = None
     ended_at: datetime | None = None
     pending_since: datetime | None = None
     disposal_deadline: datetime | None = None

src/cold_display_guard/vision.py

@@ -30,6 +30,7 @@ class RuntimeVisionSettings:
     occupancy_texture_delta: float = 18.0
     occupancy_dark_luma_threshold: float = 80.0
     occupancy_dark_fraction: float = 0.06
+    occupancy_absolute_dark_fraction: float = 0.0
     occupancy_texture_dark_fraction: float = 0.04
     occupancy_bright_luma_threshold: float = 220.0
     occupancy_bright_reflection_fraction: float = 0.18
@@ -236,6 +237,7 @@ def load_runtime_vision_settings(config: dict[str, Any]) -> RuntimeVisionSetting
         occupancy_texture_delta=float(runtime.get("occupancy_texture_delta", 18.0)),
         occupancy_dark_luma_threshold=float(runtime.get("occupancy_dark_luma_threshold", 80.0)),
         occupancy_dark_fraction=float(runtime.get("occupancy_dark_fraction", 0.06)),
+        occupancy_absolute_dark_fraction=float(runtime.get("occupancy_absolute_dark_fraction", 0.0)),
         occupancy_texture_dark_fraction=float(runtime.get("occupancy_texture_dark_fraction", 0.04)),
         occupancy_bright_luma_threshold=float(runtime.get("occupancy_bright_luma_threshold", 220.0)),
         occupancy_bright_reflection_fraction=float(runtime.get("occupancy_bright_reflection_fraction", 0.18)),
@@ -324,13 +326,18 @@ def metrics_indicate_occupied(
     dark_delta = dark_fraction - baseline_dark_fraction
     bright_reflection = is_bright_reflection(settings, dark_delta, bright_fraction)
     dark_occupied = dark_delta >= settings.occupancy_dark_fraction and not bright_reflection
+    absolute_dark_occupied = (
+        settings.occupancy_absolute_dark_fraction > 0
+        and dark_fraction >= settings.occupancy_absolute_dark_fraction
+        and not bright_reflection
+    )
     mean_occupied = mean_delta >= settings.occupancy_mean_delta and not bright_reflection
     texture_occupied = (
         texture_delta >= settings.occupancy_texture_delta
         and dark_delta >= settings.occupancy_texture_dark_fraction
         and not bright_reflection
     )
-    return dark_occupied or mean_occupied or texture_occupied
+    return dark_occupied or absolute_dark_occupied or mean_occupied or texture_occupied
 
 
 def is_bright_reflection(settings: RuntimeVisionSettings, dark_delta: float, bright_fraction: float) -> bool:

src/cold_display_guard/webhooks.py

@@ -7,6 +7,7 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Callable
 from urllib import request
+from urllib.parse import urlsplit
 
 
 @dataclass(frozen=True, slots=True)
@@ -14,6 +15,7 @@ class WebhookSettings:
     enabled: bool = False
     event_url: str = ""
     case_url: str = ""
+    source_id: str = ""
     callback_token: str = ""
     connect_timeout_seconds: float = 3.0
     read_timeout_seconds: float = 5.0
@@ -123,6 +125,7 @@ def load_webhook_settings(config: dict[str, Any]) -> WebhookSettings:
         enabled=bool(payload.get("enabled", False)),
         event_url=str(payload.get("event_url", "")),
         case_url=str(payload.get("case_url", "")),
+        source_id=str(payload.get("source_id", "")),
         callback_token=str(payload.get("callback_token", "")),
         connect_timeout_seconds=float(payload.get("connect_timeout_seconds", 3.0)),
         read_timeout_seconds=float(payload.get("read_timeout_seconds", 5.0)),
@@ -133,32 +136,88 @@ def load_webhook_settings(config: dict[str, Any]) -> WebhookSettings:
     )
 
 
-def build_batch_event_payload(event: dict[str, object]) -> dict[str, object]:
-    return {
+def build_batch_event_payload(
+    event: dict[str, object],
+    *,
+    camera_ip: str = "",
+    snapshot_upload: dict[str, object] | None = None,
+) -> dict[str, object]:
+    batch_id = str(event.get("batch_id", ""))
+    event_name = str(event.get("event", ""))
+    ts = str(event.get("ts", ""))
+    pre_warned_at = str(event.get("pre_warned_at", ""))
+    alerted_at = str(event.get("alerted_at", ""))
+    ended_at = str(event.get("ended_at", ""))
+    payload = {
         "kind": "batch_event",
-        "event": event.get("event", ""),
-        "ts": event.get("ts", ""),
-        "batch_id": event.get("batch_id", ""),
+        "event": event_name,
+        "event_code": batch_id,
+        "ts": ts,
+        "batch_id": batch_id,
         "camera_id": event.get("camera_id", ""),
+        "camera_ip": camera_ip,
         "zone_id": event.get("zone_id", ""),
         "zone_label": event.get("zone_label", ""),
         "severity": event.get("severity", ""),
         "state": event.get("state", ""),
+        "started_at": event.get("started_at", ""),
+        "ended_at": ended_at,
+        "removed_at": ended_at,
+        "dwell_seconds": event.get("dwell_seconds", ""),
+        "is_discarded": event_name == "batch_discarded",
+        "discarded_at": ts if event_name == "batch_discarded" else "",
+        "created_at": pre_warned_at or alerted_at or ts,
+        "pre_warned_at": pre_warned_at,
+        "alerted_at": alerted_at,
+        "alarm_at": alerted_at,
+        "updated_at": ts,
     }
+    return attach_snapshot_upload(payload, batch_id=batch_id, snapshot_upload=snapshot_upload)
 
 
-def build_case_event_payload(snapshot: dict[str, object]) -> dict[str, object]:
-    return {
+def build_case_event_payload(
+    snapshot: dict[str, object],
+    *,
+    camera_ip: str = "",
+    snapshot_upload: dict[str, object] | None = None,
+) -> dict[str, object]:
+    batch_id = str(snapshot.get("batch_id", ""))
+    created_at = str(snapshot.get("created_at", ""))
+    updated_at = str(snapshot.get("updated_at", ""))
+    handled_at = str(snapshot.get("handled_at", ""))
+    handled_source = str(snapshot.get("handled_source", ""))
+    event = snapshot_event(snapshot)
+    pre_warned_at = str(event.get("pre_warned_at", ""))
+    alerted_at = str(event.get("alerted_at", ""))
+    ended_at = str(event.get("ended_at", ""))
+    discarded = handled_source == "auto_closed"
+    payload = {
         "kind": "case_event",
         "action": infer_case_action(snapshot),
         "case_id": snapshot.get("case_id", ""),
+        "event_code": batch_id or snapshot.get("case_id", ""),
         "case_type": snapshot.get("case_type", ""),
         "case_status": snapshot.get("case_status", ""),
-        "batch_id": snapshot.get("batch_id", ""),
+        "batch_id": batch_id,
+        "camera_id": snapshot.get("camera_id", ""),
+        "camera_ip": camera_ip,
+        "zone_id": snapshot.get("zone_id", ""),
+        "zone_label": snapshot.get("zone_label", ""),
         "source_event": snapshot.get("source_event", ""),
-        "handled_source": snapshot.get("handled_source", ""),
-        "updated_at": snapshot.get("updated_at", ""),
+        "handled_source": handled_source,
+        "started_at": event.get("started_at", ""),
+        "ended_at": ended_at,
+        "removed_at": ended_at,
+        "dwell_seconds": event.get("dwell_seconds", ""),
+        "is_discarded": discarded,
+        "discarded_at": handled_at if discarded else "",
+        "created_at": pre_warned_at or alerted_at or created_at,
+        "pre_warned_at": pre_warned_at,
+        "alerted_at": alerted_at,
+        "alarm_at": alerted_at,
+        "updated_at": updated_at,
     }
+    return attach_snapshot_upload(payload, batch_id=batch_id, snapshot_upload=snapshot_upload)
 
 
 def infer_case_action(snapshot: dict[str, object]) -> str:
@@ -177,16 +236,20 @@ def send_batch_event_webhooks(
     retry_path: Path | None = None,
     http_post: HttpPost | None = None,
     now: datetime | None = None,
+    snapshot_upload: dict[str, object] | None = None,
 ) -> list[dict[str, object]]:
     settings = load_webhook_settings(config)
     if not settings.enabled or not settings.event_url:
         return []
+    camera_ip = infer_camera_ip(config)
     attempted_at = now or datetime.now(timezone.utc)
     deliveries: list[dict[str, object]] = []
     retry_updates: list[dict[str, object]] = []
     store = load_retry_store(retry_path) if retry_path is not None else None
     for event in events:
-        payload = build_batch_event_payload(event)
+        payload = build_batch_event_payload(event, camera_ip=camera_ip, snapshot_upload=snapshot_upload)
+        if settings.source_id:
+            payload["source_id"] = settings.source_id
         record = deliver_webhook(
             settings.event_url,
             payload,
@@ -223,16 +286,20 @@ def send_case_webhooks(
     retry_path: Path | None = None,
     http_post: HttpPost | None = None,
     now: datetime | None = None,
+    snapshot_upload: dict[str, object] | None = None,
 ) -> list[dict[str, object]]:
     settings = load_webhook_settings(config)
     if not settings.enabled or not settings.case_url:
         return []
+    camera_ip = infer_camera_ip(config)
     attempted_at = now or datetime.now(timezone.utc)
     deliveries: list[dict[str, object]] = []
     retry_updates: list[dict[str, object]] = []
     store = load_retry_store(retry_path) if retry_path is not None else None
     for snapshot in snapshots:
-        payload = build_case_event_payload(snapshot)
+        payload = build_case_event_payload(snapshot, camera_ip=camera_ip, snapshot_upload=snapshot_upload)
+        if settings.source_id:
+            payload["source_id"] = settings.source_id
         record = deliver_webhook(
             settings.case_url,
             payload,
@@ -455,6 +522,58 @@ def optional_int(value: object) -> int | None:
         return None
 
 
+def infer_camera_ip(config: dict[str, Any]) -> str:
+    stream = config.get("stream", {})
+    if not isinstance(stream, dict):
+        return ""
+    rtsp_url = str(stream.get("rtsp_url", "")).strip()
+    if not rtsp_url:
+        return ""
+    try:
+        return urlsplit(rtsp_url).hostname or ""
+    except ValueError:
+        return ""
+
+
+def snapshot_event(snapshot: dict[str, object]) -> dict[str, object]:
+    payload = snapshot.get("payload", {})
+    if not isinstance(payload, dict):
+        return {}
+    event = payload.get("event", {})
+    if not isinstance(event, dict):
+        return {}
+    return event
+
+
+def attach_snapshot_upload(
+    payload: dict[str, object],
+    *,
+    batch_id: str,
+    snapshot_upload: dict[str, object] | None,
+) -> dict[str, object]:
+    if not snapshot_upload:
+        return payload
+    batch_ids = {str(item).strip() for item in snapshot_upload.get("batch_ids", []) if str(item).strip()}
+    if batch_ids and batch_id not in batch_ids:
+        return payload
+    status = str(snapshot_upload.get("status", "")).strip()
+    if status:
+        payload["snapshot_upload_status"] = status
+    object_key = str(snapshot_upload.get("object_key", "")).strip()
+    if object_key:
+        payload["snapshot_object_key"] = object_key
+    file_name = str(snapshot_upload.get("file_name", "")).strip()
+    if file_name:
+        payload["snapshot_file_name"] = file_name
+    captured_at = str(snapshot_upload.get("captured_at", "")).strip()
+    if captured_at:
+        payload["snapshot_captured_at"] = captured_at
+    error_message = str(snapshot_upload.get("error", "")).strip()
+    if error_message:
+        payload["snapshot_upload_error"] = error_message
+    return payload
+
+
 def file_ends_with_newline(path: Path) -> bool:
     with path.open("rb") as handle:
         handle.seek(-1, 2)

tasks/lessons.md

@@ -0,0 +1,19 @@
+# Lessons
+
+- 2026-06-10: 远端接收端路由不能只凭已有相似服务或历史路径推断，必须先对用户指定的精确路径做真实 HTTP 探测，再决定配置值。
+  Prevention:
+  1. 对每个用户指定的 Webhook 路径，先在目标主机上用与真实请求接近的 `POST` 探测并记录状态码。
+  2. 如果存在多个相似路径，只能在验证过用户指定路径不可用后，才考虑回退到其它路径。
+  3. 切换远端配置前，先确认发送端容器对目标主机名或 IP 实际可达，避免写入不可解析的地址。
+
+- 2026-06-15: 告警截图叠加中文区域名时，不能依赖默认西文字体或手写点阵字形；这会在现场截图里表现为乱码或不可读文字。
+  Prevention:
+  1. 涉及中文截图叠字时，镜像必须安装可验证的 CJK 字体包，并在容器内用 `fc-match :lang=zh-cn` 确认命中 CJK 字体。
+  2. 部署后必须在目标容器内跑一次中文标签叠图烟测，确认真实渲染路径可用，而不只检查像素变化。
+  3. 字体渲染不可用时，回退文本必须转换成可读 ASCII 标识，例如 `区域 1` -> `R1`、`垃圾区` -> `TRASH`，避免继续绘制乱码中文。
+
+- 2026-06-15: 现场识别抖动排查时，不能先假设某个区域为空；用户指出区域 1、2、6 实际都有物后，原先单纯调高相对暗区阈值会压掉真实占用。
+  Prevention:
+  1. 调整视觉阈值前，必须向现场实际状态对齐，明确每个被分析区域当前应该是有物还是空。
+  2. 如果物品已存在于启动基线中，不能只依赖相对基线变化；需要绝对特征或重新采空基线来识别。
+  3. 对“正常取用”误报，应优先检查有物状态是否短暂掉空，并用判空确认帧数或滞后来处理抖动，而不是只提高占用阈值。

tasks/todo.md

@@ -1,31 +1,421 @@
 # Task Todo
 
+## Current Task: Runtime/API Case State Reopen Fix
+
+**Goal:** When the management API marks a display-cabinet case as handled, the runtime process must not later append a newer `open` snapshot for the same case from stale in-memory state.
+
+- [x] Add a failing regression test for API-written `handled` state being preserved when runtime persists later events.
+- [x] Fix runtime case persistence to reconcile with the latest JSONL snapshots before applying new events.
+- [x] Run targeted case/runtime tests.
+- [x] Record remote chain verification and deployment status.
+
+### Findings
+
+- On `xiaozheng@10.8.0.23`, `case_batch_000911` was marked `handled` at `2026-06-15T07:27:12Z`, then runtime appended a newer `open` snapshot for the same case at `2026-06-15T15:38:03+08:00`.
+- The API and runtime are separate processes sharing `logs/cases.jsonl`; runtime keeps a long-lived `CaseStore` loaded at startup and did not see the API-written handled snapshot.
+
+### Verification
+
+- RED:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests.test_main.RuntimeRestoreTests.test_persist_case_updates_preserves_api_handled_snapshot -v`
+  - Result before fix: failed because runtime appended a later `open` snapshot.
+- Local targeted verification:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests.test_main.RuntimeRestoreTests.test_persist_case_updates_preserves_api_handled_snapshot -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_cases.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_main.py -v`
+  - Result: all passed.
+- Remote deployment:
+  - Synced only `src/cold_display_guard/main.py` to `xiaozheng@10.8.0.23:/home/xiaozheng/cold_display_guard/src/cold_display_guard/main.py`.
+  - Ran `docker compose --env-file deploy/cold-display-guard.env -f deploy/docker-compose.yml up -d --build cold-display-guard-runtime`.
+  - Compose recreated `cold-display-guard-api` and `cold-display-guard-runtime`; health check returned `status=ok`.
+- Remote behavior check:
+  - Ran the same API-handled/runtime-later-event scenario inside `cold-display-guard-runtime` using a temp JSONL file.
+  - Result: `{"handled_source": "manual", "latest_status": "handled", "new_snapshots": 0}`.
+
 - [x] Review the current project instructions and check for task-relevant lessons.
-- [x] Check repository status before starting retry-queue work.
-- [x] Re-verify that `main` includes webhook case management before layering retries on top.
-- [x] Inspect the current webhook delivery path, config surface, runtime integration point, and manage API hooks.
-- [x] Write the detailed retry-queue implementation plan to `docs/superpowers/plans/2026-06-09-webhook-retry-queue.md`.
-- [x] Execute webhook retry queue backend TDD cycle.
-- [x] Execute runtime/manage API retry integration TDD cycle.
-- [x] Update documentation/config formatting for retry queue settings and sinks.
+- [x] Inspect the OTA upload API document and current runtime/webhook capture path.
+- [x] Create an isolated worktree for alarm snapshot upload implementation.
+- [x] Write the detailed implementation plan to `docs/superpowers/plans/2026-06-09-alarm-snapshot-upload.md`.
+- [x] Execute alarm snapshot upload client TDD cycle.
+- [x] Execute runtime and webhook payload integration TDD cycle.
+- [x] Update config surface, docs, and verification notes.
 - [x] Run targeted verification and final full verification.
 
 ## Notes
 
 - `tasks/lessons.md` is absent in this repository/worktree, so there were no prior session lessons to review.
-- Main branch merge result is available locally at `81f1709`; retry-queue work continues from branch `feat/webhook-retry-queue`.
+- Upload API reference: `/Users/glo/code/go/wenma/ai_manager/zd-ai-manager/chunk-upload-oss-service/UPLOAD_API.md`
+- User-provided upload target: `https://ota.zhengxinshipin.com`
+- User-provided token secret: `change-me-in-production`
 
 ## Review
 
-- Plan saved to `docs/superpowers/plans/2026-06-09-webhook-retry-queue.md`.
-- Chosen scope keeps the first outbound webhook attempt synchronous, then persists failures into a JSONL-backed retry queue with bounded backoff and dead-letter cutoff.
-- Retry queue observability and manual compensation will be exposed through the management API rather than the frontend in this phase.
-- Implemented queue-aware webhook delivery in `src/cold_display_guard/webhooks.py`, runtime retry draining in `src/cold_display_guard/main.py`, manage API retry list/drain endpoints in `src/cold_display_guard/manage_api.py`, and config/doc updates in `src/cold_display_guard/config.py`, `config/example.toml`, and `README_zh.md`.
+- Plan saved to `docs/superpowers/plans/2026-06-09-alarm-snapshot-upload.md`.
+- Chosen implementation keeps snapshot upload entirely outside `BatchEngine` and enriches webhook payloads from the runtime side using the already captured frame.
+- Implemented `src/cold_display_guard/alarm_snapshots.py` for JPEG encoding plus OTA chunk-upload orchestration, runtime integration in `src/cold_display_guard/main.py`, webhook payload enrichment in `src/cold_display_guard/webhooks.py`, config exposure/secret stripping in `src/cold_display_guard/config.py` and `src/cold_display_guard/manage_api.py`, and config/doc updates in `config/example.toml` and `README_zh.md`.
 - Targeted verification passed:
-  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_alarm_snapshots.py -v`
   - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_main.py -v`
-  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
-  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_config.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py tests/test_config.py tests/test_manage_api.py -v`
 - Final verification passed:
   - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest discover -s tests -v`
   - `cd web && pnpm install --frozen-lockfile && pnpm build`
+
+## Current Task: Webhook Payload Field Gap Check
+
+- [x] Pull the actual payload currently received by `video-recognition` and compare it against the required event list fields.
+- [x] Patch webhook payload builders to include the missing non-store fields required by the downstream table.
+- [x] Add or update focused webhook tests for the enriched payload shape.
+- [x] Run targeted verification and record the result here.
+
+### Current Findings
+
+- Current received payload only includes `batch_id`, `camera_id`, `event`, `kind`, `severity`, `source_id`, `state`, `ts`, `zone_id`, and `zone_label`.
+- Missing or not explicitly populated for the downstream event table: event code, camera IP, batch start time, removal time, dwell duration, discard flag, discard time, create time, alarm time, and update time.
+
+### Field Gap Verification
+
+- Actual receiver payload before the fix, from `video-recognition` result JSONL on `10.8.0.11`, confirmed only the base fields above and did not include the downstream table time/discard/IP fields.
+- Updated `src/cold_display_guard/webhooks.py` so both `batch_event` and `case_event` now include:
+  - `event_code`
+  - `camera_ip`
+  - `started_at`
+  - `ended_at`
+  - `removed_at`
+  - `dwell_seconds`
+  - `is_discarded`
+  - `discarded_at`
+  - `created_at`
+  - `alerted_at`
+  - `alarm_at`
+  - `updated_at`
+- `case_event` also now carries the missing contextual fields `camera_id`, `zone_id`, and `zone_label`.
+- Verification passed:
+  - `PYTHONPATH=src python3 -m unittest tests/test_webhooks.py -v`
+  - `PYTHONPATH=src python3 -m unittest tests/test_main.py -v`
+  - `PYTHONPATH=src python3 -m unittest discover -s tests -v`
+- Deployed updated code to `xiaozheng@10.8.0.11` without overwriting the remote `config/example.toml`, rebuilt `cold-display-guard:dev`, and restarted only `cold-display-guard-api` plus `cold-display-guard-runtime`.
+- Natural post-deploy traffic did not arrive during the 2-minute observation window, so final runtime verification used the deployed container to build representative batch/case webhook payloads with the live remote config and confirmed `camera_ip = 192.168.3.4` plus all new downstream fields were present.
+
+## Current Task: Deploy To 192.168.5.103
+
+- [x] Inspect the existing deployment layout and active containers on `xiaozheng@192.168.5.103`.
+- [x] Verify the exact webhook route on that host before writing config.
+- [x] Sync the current project code to the remote deployment directory without overwriting the live RTSP and calibration config.
+- [x] Configure the remote webhook settings for the local `video-recognition` receiver.
+- [x] Rebuild and restart the remote API/runtime containers, then verify health and outbound webhook configuration.
+
+### Deployment Findings
+
+- Existing deployment path on `192.168.5.103` is `/home/xiaozheng/cold_display_guard`, not `~/apps/cold-display-guard/app`.
+- The host already runs `cold-display-guard-api`, `cold-display-guard-runtime`, and `cold-display-guard-web` on ports `19080` and `23000`.
+- The same host also runs `video-recognition`, and a direct probe to `http://127.0.0.1:8080/api/webhook/cold-display-guard` returned `200 OK`, so this is the verified webhook target for this environment.
+
+### Deployment Verification
+
+- From inside the running `cold-display-guard-api` container on `192.168.5.103`:
+  - `http://host.docker.internal:8080/api/webhook/cold-display-guard` failed DNS resolution.
+  - `http://172.17.0.1:8080/api/webhook/cold-display-guard` returned `200 OK`.
+  - `http://192.168.5.103:8080/api/webhook/cold-display-guard` returned `200 OK`.
+- The configured webhook target was set to `http://192.168.5.103:8080/api/webhook/cold-display-guard` for both `event_url` and `case_url`.
+- Remote config was enriched to include:
+  - `case_sink`
+  - `alarm_snapshot_upload`
+  - `webhook_retry_sink`
+  - `webhook_delivery_sink`
+  - `webhooks`
+- Code sync used `rsync` with `config/example.toml` excluded so the live RTSP URL and calibration polygons were preserved.
+- Remote rebuild/restart completed for `cold-display-guard-api` and `cold-display-guard-runtime`.
+- Verified after restart:
+  - `GET http://127.0.0.1:19080/api/manage/health` returned `status=ok`
+  - `GET http://127.0.0.1:19080/api/manage/config` showed `webhooks.enabled=true`
+  - `event_url` and `case_url` both active on `http://192.168.5.103:8080/api/webhook/cold-display-guard`
+  - `alarm_snapshot_upload.enabled=true`
+
+## Current Task: Alarm Snapshot Calibration Overlay
+
+**Goal:** Webhook-linked uploaded alarm snapshots should visually include the calibrated cold display zones and trash confirmation ROI from the current config.
+
+**Design:** Keep the existing runtime flow intact: capture current RTSP frame, process events, then upload an alarm snapshot only for warning/alarm events. Before JPEG encoding, build overlay regions from `[[zones]]` plus `[trash].roi`, clamp normalized polygon coordinates to the image bounds, draw a semi-transparent fill and visible outline directly onto a copied `Frame.rgb`, and pass that annotated frame to the existing encoder/uploader. Do not change `BatchEngine`, Webhook payload shape, OTA upload protocol, or management snapshot capture.
+
+- [x] Review task-relevant lessons and current dirty worktree.
+- [x] Inspect `alarm_snapshots.py`, `main.py`, config polygon shape, and existing tests.
+- [x] Write a failing unit test proving alert snapshot upload encodes an annotated frame when zones/trash ROI are configured.
+- [x] Write focused unit tests for polygon overlay behavior using a tiny RGB frame.
+- [x] Run targeted tests and confirm the new tests fail for the expected missing overlay behavior.
+- [x] Implement the smallest standard-library overlay helper in `src/cold_display_guard/alarm_snapshots.py`.
+- [x] Wire `capture_alert_snapshot` to apply configured overlays before JPEG encoding.
+- [x] Run targeted snapshot/runtime tests.
+- [x] Run the full Python test suite.
+
+### Review
+
+- Added `apply_calibration_overlay` in `src/cold_display_guard/alarm_snapshots.py` to draw configured food-zone polygons in yellow and the trash ROI in red onto a copied frame before JPEG encoding and OTA upload.
+- The overlay clamps normalized coordinates to image bounds, draws semi-transparent fills plus outlines, and leaves the original `Frame.rgb` unchanged for downstream runtime processing.
+- `capture_alert_snapshot` now encodes the annotated frame when warning/alarm events trigger snapshot upload; non-alert events and disabled upload behavior are unchanged.
+- Targeted verification passed:
+  - `PYTHONPATH=src python3 -m unittest tests/test_alarm_snapshots.py -v`
+  - `PYTHONPATH=src python3 -m unittest tests/test_main.py -v`
+- Full verification passed:
+  - `PYTHONPATH=src python3 -m unittest discover -s tests -v`
+
+## Current Task: Deploy Overlay Update To 10.8.0.23
+
+**Goal:** Deploy the alarm snapshot calibration overlay change to `xiaozheng@10.8.0.23` without overwriting live RTSP/calibration config or unrelated local changes.
+
+**Plan:** Inspect the remote deployment layout first, confirm which containers are active, sync only the runtime source file required for the overlay change, rebuild/restart the API/runtime services that use the Python image, and verify both service health and the deployed source code.
+
+- [x] Inspect remote deployment directory, Docker/Compose files, and active containers on `xiaozheng@10.8.0.23`.
+- [x] Confirm the remote config file remains present and is not overwritten.
+- [x] Sync `src/cold_display_guard/alarm_snapshots.py` to the remote deployment path.
+- [x] Rebuild and restart only the affected `cold-display-guard-api` and `cold-display-guard-runtime` services when Compose is available.
+- [x] Verify management API health after restart.
+- [x] Verify the deployed remote source contains `apply_calibration_overlay`.
+
+### Deployment Review
+
+- Remote deployment path confirmed as `/home/xiaozheng/cold_display_guard`.
+- Active services before deployment: `cold-display-guard-api`, `cold-display-guard-runtime`, and `cold-display-guard-web`.
+- Remote live `config/example.toml` was checked before and after deployment and was not overwritten.
+- Synced only `src/cold_display_guard/alarm_snapshots.py` to avoid deploying unrelated local `web/nginx.conf` changes.
+- Created a timestamped backup of the previous remote `alarm_snapshots.py` beside the source file before syncing.
+- Rebuilt `cold-display-guard:dev` with `docker compose --env-file deploy/cold-display-guard.env -f deploy/docker-compose.yml build cold-display-guard-api`.
+- Restarted only `cold-display-guard-api` and `cold-display-guard-runtime` with Compose; `cold-display-guard-web` remained untouched.
+- Verification passed:
+  - `curl http://127.0.0.1:19080/api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - `docker exec cold-display-guard-api python3 -c ...` confirmed `apply_calibration_overlay` exists in the running image with signature `(frame, config) -> Frame`.
+  - API and runtime logs show normal startup after restart.
+
+## Current Task: Update Timing Parameters On 10.8.0.23
+
+**Goal:** Adjust the live timing settings on `xiaozheng@10.8.0.23` per operator request.
+
+**Applied mapping:** The current application has no separate pre-warning threshold. It supports `max_dwell_seconds` for the time alarm/overdue threshold and `trash_confirmation_seconds` for the disposal confirmation window before warning escalation. Applied `max_dwell_seconds = 120` and `trash_confirmation_seconds = 30`.
+
+- [x] Back up `/home/xiaozheng/cold_display_guard/config/example.toml`.
+- [x] Update `[thresholds].max_dwell_seconds` from `300` to `120`.
+- [x] Update `[thresholds].trash_confirmation_seconds` from `120` to `30`.
+- [x] Restart `cold-display-guard-api` and `cold-display-guard-runtime`.
+- [x] Verify `/api/manage/health`.
+- [x] Verify `/api/manage/config` returns `{"max_dwell_seconds": 120, "trash_confirmation_seconds": 30}`.
+
+### Timing Update Review
+
+- Remote config was edited in place after creating a timestamped backup.
+- `cold-display-guard-api` and `cold-display-guard-runtime` were explicitly restarted with Docker Compose.
+- `cold-display-guard-web` was not restarted.
+- Verification passed:
+  - `GET http://127.0.0.1:19080/api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - `GET http://127.0.0.1:19080/api/manage/config` returned `max_dwell_seconds = 120` and `trash_confirmation_seconds = 30`.
+  - Container status showed `cold-display-guard-api` healthy and `cold-display-guard-runtime` running after restart.
+- Note: requested `预警时长 = 1min` is not independently configurable in the current codebase; supporting distinct pre-warning at 60 seconds and overdue at 120 seconds would require a code change.
+
+## Current Task: Pre-Warning Alarm Flow And Full Webhook/MQTT Chain
+
+**Goal:** Implement the requested camera-side timing flow, deploy it to `xiaozheng@10.8.0.23`, and verify the Webhook -> `video_recognition_local` -> MQTT -> `store_data_platform` chain.
+
+**Design:** Keep all timing decisions inside `cold_display_guard.BatchEngine`. Add separate thresholds for pre-warning, alarm, and alarm-removal timeout; emit explicit lifecycle events so downstream services do not infer camera-side timers. Keep `video_recognition_local` as a transparent Webhook/MQTT bridge, and update `store_data_platform` only where event names map to notifications, case types, and CRM penalty submission.
+
+- [x] Review task-relevant instructions, lessons, and dirty worktree.
+- [x] Inspect the current cold-display engine, case store, webhook payload, and tests.
+- [x] Inspect `video_recognition_local` cold-display Webhook receiver and MQTT publisher.
+- [x] Inspect `store_data_platform` cold-display MQTT consumer, notification mapping, and CRM submission trigger.
+- [x] Inspect `xiaozheng@10.8.0.23` active containers and deployment paths.
+- [x] Add failing cold-display engine/case/config/webhook tests for `time_pre_warning`, `pre_warning_handled`, `time_alarm`, and `alarm_removal_timeout`.
+- [x] Implement the camera-side state machine and config fields.
+- [x] Add/adjust `video_recognition_local` passthrough tests for the new event names.
+- [x] Add/adjust `store_data_platform` tests and mappings for new event semantics.
+- [x] Run local targeted and full relevant verification.
+- [x] Deploy changed services to `xiaozheng@10.8.0.23` without overwriting live RTSP/calibration secrets.
+- [x] Update the remote timing config to `pre_warning_seconds=60`, `max_dwell_seconds=120`, `alarm_removal_seconds=30`, `trash_confirmation_seconds=30`.
+- [x] Verify remote Webhook target reachability from the cold-display container to local `video-recognition`.
+- [x] Observe cold-display, video-recognition, MQTT, and platform logs; record the result.
+
+### Current Findings
+
+- `cold_display_guard` currently has only `max_dwell_seconds` and `trash_confirmation_seconds`; it cannot independently represent 1-minute pre-warning, 2-minute alarm, and 30-second alarm-removal timeout.
+- `video_recognition_local` receives `/api/webhook/cold-display-guard` payloads as generic JSON and forwards them to MQTT; new event names should remain transparent, but tests should lock this behavior.
+- `store_data_platform` currently treats `time_alarm` and `batch_pending_disposal` as warning notifications, and only `warning_escalated` triggers CRM penalty submission. This must change so `time_pre_warning` is the warning, `time_alarm` is the alert reminder, and `alarm_removal_timeout` triggers CRM submission.
+- On `10.8.0.23`, active containers include `cold-display-guard-*`, `video-recognition`, and `mosquitto`; `video-recognition` runs with host networking, while `cold-display-guard-api` runs on its Compose network.
+
+### Local Verification
+
+- Cold-display full Python suite passed: `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`98` tests).
+- `video_recognition_local` cold-display focused tests passed: `go test ./internal/server ./internal/mqtt ./cmd -run 'TestColdDisplayGuard|Test.*ColdDisplayGuard' -count=1`.
+- `store_data_platform` display-cabinet service focused tests passed: `go test ./store_data/service -run 'Test.*StoreDisplayCabinet|TestResolveStoreDisplayCabinet.*|TestShouldSubmitStoreDisplayCabinetPenalty|TestBuildStoreDisplayCabinet.*' -count=1`.
+
+### Deployment Review
+
+- Synced only these cold-display source files to `xiaozheng@10.8.0.23:/home/xiaozheng/cold_display_guard/src/cold_display_guard/`: `models.py`, `config.py`, `engine.py`, `cases.py`, `webhooks.py`.
+- Backed up the remote source files and live `config/example.toml` before deployment.
+- Updated the live remote thresholds to `pre_warning_seconds=60`, `max_dwell_seconds=120`, `alarm_removal_seconds=30`, and `trash_confirmation_seconds=30`.
+- Updated the live remote Webhook target from the unreachable old host to `http://10.8.0.23:8080/api/webhook/cold-display-guard`.
+- Rebuilt `cold-display-guard:dev` and restarted only `cold-display-guard-api` and `cold-display-guard-runtime`.
+- Remote verification passed:
+  - `GET /api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - `GET /api/manage/config` returned the four expected threshold values and the new Webhook target.
+  - Container-side synthetic engine run emitted `batch_started`, `time_pre_warning`, `time_alarm`, `alarm_removal_timeout`, then `batch_pending_disposal` plus `batch_discarded`.
+  - Natural runtime log emitted `alarm_removal_timeout` for `batch_000881` at `2026-06-15T11:52:20+08:00`.
+  - Webhook delivery for that event returned HTTP `200` from `video-recognition`.
+  - `video_recognition_local` result JSONL recorded both `alarm_removal_timeout` batch and case events.
+  - MQTT probe confirmed `video-recognition` published to `video/cold-display-guard/result/cold-display-guard` with `device_identifier=cold-display-guard`.
+- `store_data_platform` is not deployed on `10.8.0.23` under that repository name or as an identifiable container; platform handling changes were completed and verified in the local repository.
+- The cold-display retry queue has no pending entries; old `192.168.5.103` failures are already dead-letter history.
+
+## Current Task: Alarm Snapshot Labels And Zone Colors
+
+**Goal:** Uploaded alarm screenshots should show each calibrated region name directly on the image, and different cold-display zones should use different overlay colors.
+
+**Design:** Extend the existing standard-library overlay path. Keep drawing configured polygons before JPEG upload, but carry a display label for each region, choose a stable color from a fixed palette by zone order, and draw a small high-contrast text label inside the polygon. Keep trash ROI red and labeled separately.
+
+- [x] Inspect the current calibration overlay helper and tests.
+- [x] Add failing tests for per-zone colors and visible region labels.
+- [x] Implement labels and stable zone color palette.
+- [x] Run snapshot tests and full Python tests.
+- [x] Deploy the overlay update to `xiaozheng@10.8.0.23`.
+- [x] Verify remote API/runtime health and deployed overlay helper.
+
+### Review
+
+- `apply_calibration_overlay` now assigns each cold-display zone a stable color from a fixed palette and keeps the trash ROI red.
+- Each overlay region now carries a label and draws a small high-contrast label box directly on the frame before JPEG encoding/upload.
+- The built-in label renderer covers common现场 labels such as `区域 1` through digits and `垃圾区`, plus basic ASCII for custom numeric/English labels.
+- Verification passed:
+  - `PYTHONPATH=src python3 -m unittest tests/test_alarm_snapshots.py -v`
+  - `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`99` tests)
+- Deployed `src/cold_display_guard/alarm_snapshots.py` to `xiaozheng@10.8.0.23` after backing up the previous remote file.
+- Rebuilt `cold-display-guard:dev` and restarted `cold-display-guard-api` plus `cold-display-guard-runtime`.
+- Remote verification passed:
+  - `GET /api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - Container-side overlay smoke test confirmed two zones render different RGB values and label text pixels are present.
+
+## Current Task: Alarm Snapshot Chinese Label Rendering Fix
+
+**Goal:** Fix unreadable/garbled Chinese region names on uploaded alarm screenshots while keeping per-zone colors and fallback labeling robust.
+
+**Design:** Use a real CJK font renderer for Chinese labels in the alarm snapshot overlay path. Install Noto CJK fonts in the runtime image, render labels through ffmpeg `drawtext` when the font is available, and fall back to readable ASCII labels if the font renderer is unavailable.
+
+- [x] Reproduce and identify the likely root cause: remote container only matched DejaVu for `zh-cn`, so Chinese labels had no real CJK font path.
+- [x] Add regression tests for Docker CJK font installation and readable ASCII fallback labels.
+- [x] Update `Dockerfile` to install `fonts-noto-cjk`.
+- [x] Update `alarm_snapshots.py` to prefer CJK font rendering and use `R1`/`TRASH` fallback text when needed.
+- [x] Run focused and full local Python verification.
+- [x] Deploy `Dockerfile` and `alarm_snapshots.py` to `xiaozheng@10.8.0.23` without overwriting live config.
+- [x] Rebuild/restart `cold-display-guard-api` and `cold-display-guard-runtime`.
+- [x] Verify remote API/runtime health, CJK font availability, overlay smoke behavior, and runtime logs.
+
+### Review
+
+- Root cause was the screenshot overlay path not having a real Chinese font renderer in the deployed image; the container matched DejaVu before this fix.
+- The rebuilt remote container now reports `NotoSansCJK-Regular.ttc: "Noto Sans CJK SC" "Regular"` for `fc-match :lang=zh-cn`.
+- Remote overlay smoke test confirmed `find_cjk_font_file()` returns `/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc`, Chinese labels change the frame, bright label pixels are present, and different regions retain distinct colors.
+- Local verification passed:
+  - `PYTHONPATH=src python3 -m unittest tests/test_alarm_snapshots.py -v`
+  - `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`101` tests)
+- Remote verification passed:
+  - `GET /api/manage/health` returned `status=ok`, `runtime_status=running`, and version `dev`.
+  - `cold-display-guard-api` is healthy and `cold-display-guard-runtime` is running after restart.
+  - Runtime logs show normal startup after the restart.
+
+## Current Task: Investigate False Normal Consumption Events On 10.8.0.23
+
+**Goal:** Determine why the live system records a normal consumption event about every two minutes with a dwell time near 13 seconds even when no one touched the cold display cabinet.
+
+**Debug plan:** Inspect remote runtime/event/case/diagnostic logs first, correlate `batch_started` and `batch_consumed` pairs by zone and dwell time, then trace the vision metrics for those timestamps to identify whether the source is occupancy flicker, runtime restart state restoration, config thresholds, or downstream display interpretation.
+
+- [ ] Inspect recent remote events and confirm the exact event names, zones, dwell seconds, and cadence.
+- [ ] Inspect runtime diagnostics around those timestamps for occupancy and vision metric flicker.
+- [ ] Inspect live config and runtime logs for sampling/stabilization settings and restarts.
+- [x] Form and test a root-cause hypothesis before changing code or live thresholds.
+- [x] Record findings, fix if needed, and verify with logs/tests.
+
+### Findings And Fix
+
+- The repeated records were real `batch_started` -> `batch_consumed` events from the camera-side engine, not a downstream display issue.
+- Before the fix, recent events showed repeated zone 1 batches ending after 13-33 seconds, matching the two-frame confirmation cadence at the current sampling rate.
+- Root cause had two parts:
+  - Zone 1 was genuinely occupied, but its vision signal hovered around the old relative dark threshold, so short raw-occupancy dips were interpreted as item removal.
+  - Zone 2 was occupied before or during baseline learning, so its relative difference from baseline stayed near zero and it was not detected as occupied.
+- Added `occupancy_absolute_dark_fraction` in `src/cold_display_guard/vision.py`, defaulting to `0.0` so existing configs are unchanged unless they opt in.
+- Updated the live config on `xiaozheng@10.8.0.23`:
+  - `occupancy_dark_fraction = 0.12`
+  - `occupancy_absolute_dark_fraction = 0.085`
+  - `empty_confirm_frames = 6`
+- Rebuilt and restarted `cold-display-guard-api` and `cold-display-guard-runtime`.
+- Verification:
+  - Local full Python suite passed: `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`102` tests).
+  - Remote health returned `status=ok` and `runtime_status=running`.
+  - Remote container config shows the new thresholds.
+  - After deployment, latest diagnostics stabilized at `zone_counts = {"1": 1, "2": 1, "6": 1}`.
+- During a two-minute observation window after `13:25`, no new `batch_consumed` events were emitted; only expected pre-warning/alarm lifecycle events appeared for the occupied zones.
+
+## Current Task: Reduce Alarm Snapshot Label Visual Obstruction
+
+**Goal:** Region labels on uploaded alarm screenshots should be smaller and more transparent so operators can inspect the food/display image underneath.
+
+**Design:** Keep the existing label content, placement, CJK font rendering, and per-zone colors. Only reduce the visual weight of the label layer by lowering font size, black label-box opacity, border width, and fallback label-box opacity.
+
+- [x] Inspect current alarm snapshot label rendering style.
+- [x] Add a regression test for smaller ffmpeg drawtext label style.
+- [x] Reduce drawtext font size and label-box opacity.
+- [x] Keep fallback label renderer visually consistent with the ffmpeg path.
+- [x] Run full local verification.
+- [x] Deploy the updated snapshot overlay style to `xiaozheng@10.8.0.23`.
+- [x] Verify remote runtime health and deployed label style.
+
+### Notes
+
+- Targeted snapshot test passed: `PYTHONPATH=src python3 -m unittest tests/test_alarm_snapshots.py -v`.
+- Full local verification passed: `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`103` tests).
+- Remote verification passed:
+  - `GET /api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - Running container uses `fontsize=13`, `boxcolor=black@0.34`, and `boxborderw=2` for region labels.
+  - `cold-display-guard-runtime` logs show normal startup after restart.
+
+## Current Task: Limit Alert Snapshot Overlay To Event Zones
+
+**Goal:** Uploaded warning/alarm screenshots should only draw the cold-display region polygons and names for the zones that actually triggered the warning/alarm event. Other configured zones and the trash ROI should not be drawn on those uploaded screenshots.
+
+**Plan:** Keep the full calibration overlay helper available for tests and general use, but pass alert event zone IDs from `capture_alert_snapshot` into the overlay loader and disable trash ROI drawing for alert uploads.
+
+- [x] Add a regression test proving alert snapshot upload only annotates the triggering event zone.
+- [x] Filter snapshot overlay regions by event `zone_id` during alert upload.
+- [x] Preserve full overlay behavior when `apply_calibration_overlay` is called without filters.
+- [x] Run full local Python verification.
+- [x] Deploy `alarm_snapshots.py` to `xiaozheng@10.8.0.23`.
+- [x] Verify remote API/runtime health and deployed filtered-overlay behavior.
+
+### Review
+
+- Local verification passed:
+  - `PYTHONPATH=src python3 -m unittest tests/test_alarm_snapshots.py -v`
+  - `PYTHONPATH=src python3 -m unittest discover -s tests -v` (`104` tests)
+- Deployed only `src/cold_display_guard/alarm_snapshots.py` to `xiaozheng@10.8.0.23` after backing up the previous remote file; live config was not overwritten.
+- Rebuilt `cold-display-guard:dev` and restarted `cold-display-guard-api` plus `cold-display-guard-runtime`.
+- Remote verification passed:
+  - `GET /api/manage/health` returned `status=ok` and `runtime_status=running`.
+  - Container-side smoke test for a zone-1 alert returned `zone1_changed=True`, `zone2_unchanged=True`, and `trash_unchanged=True`.
+  - API/runtime logs show normal startup after restart.
+
+## Current Task: Check Webhook Duplicate Delivery
+
+**Goal:** Verify whether `cold_display_guard` is sending duplicate Webhook requests to `video-recognition` on `xiaozheng@10.8.0.23`.
+
+**Investigation:** Compare the sending code path, remote webhook delivery audit, retry queue state, cold-display event/case logs, `video-recognition` HTTP logs, and the receiver-side JSONL payloads.
+
+- [x] Inspect sender code path for direct event/case delivery and retry drain behavior.
+- [x] Confirm remote Webhook config uses the same URL for `event_url` and `case_url`.
+- [x] Check sender delivery audit for duplicate receiver `task_id` values.
+- [x] Check retry queue for pending successful redelivery risk.
+- [x] Check receiver-side cold-display JSONL for duplicate payloads and duplicate business keys.
+- [x] Trace the only coarse duplicate-looking case around `batch_000898`.
+
+### Review
+
+- Current remote config sends both `batch_event` and `case_event` to `http://10.8.0.23:8080/api/webhook/cold-display-guard`, so one business transition can produce two HTTP POSTs to the same endpoint with different `kind` values.
+- Sender audit `logs/webhook_delivery.jsonl` contains `3056` records total; recent valid delivery has `321` direct `ok` records and `0` retry `ok` records.
+- Receiver-returned `task_id` values are unique: `321` unique task IDs and `0` duplicate task IDs.
+- Retry queue has `547` latest retry items, all `dead_letter`; there are no pending retries.
+- Receiver-side `video-recognition` cold-display files for `2026-06-15` contain `181` business payloads; exact payload duplicates are `0`, and fine-grained business key duplicates are `0`.
+- Sender `events.jsonl` contains `3325` events; duplicate `(batch_id, event, ts, zone_id)` keys are `0`.
+- The only coarse duplicate-looking receiver entry was `batch_000898` at `13:20:26`: the same frame emitted `time_pre_warning` and `pre_warning_handled`, which produced separate `case_event` actions `created` and `handled`. This is not the same Webhook request repeated.

tests/test_alarm_snapshots.py

@@ -0,0 +1,315 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from cold_display_guard import alarm_snapshots
+from cold_display_guard.alarm_snapshots import (
+    capture_alert_snapshot,
+    fallback_label_text,
+    load_alarm_snapshot_settings,
+    upload_snapshot_bytes,
+)
+from cold_display_guard.vision import Frame
+
+
+UTC = timezone.utc
+
+
+class AlarmSnapshotTests(unittest.TestCase):
+    def test_load_alarm_snapshot_settings_from_config(self) -> None:
+        settings = load_alarm_snapshot_settings(
+            {
+                "alarm_snapshot_upload": {
+                    "enabled": True,
+                    "service_url": "https://ota.zhengxinshipin.com",
+                    "secret": "change-me-in-production",
+                    "object_key_prefix": "alarms/cold-display",
+                    "connect_timeout_seconds": 4,
+                    "read_timeout_seconds": 9,
+                    "encode_timeout_seconds": 7,
+                }
+            }
+        )
+
+        self.assertTrue(settings.enabled)
+        self.assertEqual(settings.service_url, "https://ota.zhengxinshipin.com")
+        self.assertEqual(settings.secret, "change-me-in-production")
+        self.assertEqual(settings.object_key_prefix, "alarms/cold-display")
+        self.assertEqual(settings.connect_timeout_seconds, 4)
+        self.assertEqual(settings.read_timeout_seconds, 9)
+        self.assertEqual(settings.encode_timeout_seconds, 7)
+
+    def test_upload_snapshot_bytes_uses_documented_chunk_upload_flow(self) -> None:
+        json_calls: list[tuple[str, dict[str, object]]] = []
+        chunk_calls: list[tuple[str, dict[str, str], bytes, dict[str, str]]] = []
+
+        def fake_post_json(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, dict[str, object]]:
+            json_calls.append((url, payload))
+            if url.endswith("/token/generate"):
+                return 200, {"token": "token-1", "expires_at": 1770003600}
+            if url.endswith("/upload/init"):
+                return 200, {"upload_id": "upload-1"}
+            if url.endswith("/upload/complete"):
+                return 200, {"upload_id": "upload-1", "object_key": "uploads/alarms/a.jpg", "file_size": 3, "file_md5": "900150983cd24fb0d6963f7d28e17f72"}
+            raise AssertionError(url)
+
+        def fake_post_multipart(
+            url: str,
+            fields: dict[str, str],
+            file_field: str,
+            file_name: str,
+            file_bytes: bytes,
+            timeout: tuple[float, float],
+        ) -> tuple[int, dict[str, object]]:
+            chunk_calls.append((url, fields, file_bytes, {"file_field": file_field, "file_name": file_name}))
+            return 200, {"upload_id": "upload-1", "index": 0, "size": len(file_bytes), "received_chunks": 1, "total_chunks": 1}
+
+        result = upload_snapshot_bytes(
+            b"abc",
+            file_name="alarm.jpg",
+            object_key_hint="alarms/a.jpg",
+            settings=load_alarm_snapshot_settings({}),
+            post_json_request=fake_post_json,
+            post_multipart_request=fake_post_multipart,
+        )
+
+        self.assertEqual(result["status"], "uploaded")
+        self.assertEqual(result["object_key"], "uploads/alarms/a.jpg")
+        self.assertEqual(json_calls[0][0], "https://ota.zhengxinshipin.com/token/generate")
+        self.assertEqual(json_calls[1][0], "https://ota.zhengxinshipin.com/upload/init")
+        self.assertEqual(json_calls[2][0], "https://ota.zhengxinshipin.com/upload/complete")
+        self.assertIn("token=token-1", chunk_calls[0][0])
+        self.assertIn("upload_id=upload-1", chunk_calls[0][0])
+        self.assertEqual(chunk_calls[0][1]["chunk_md5"], "900150983cd24fb0d6963f7d28e17f72")
+        self.assertEqual(chunk_calls[0][3]["file_field"], "chunk")
+
+    def test_capture_alert_snapshot_skips_non_alert_events(self) -> None:
+        result = capture_alert_snapshot(
+            Frame(width=1, height=1, rgb=b"\x00\x00\x00"),
+            [{"event": "batch_started", "severity": "info", "batch_id": "batch_1"}],
+            {},
+            now=datetime(2026, 6, 9, 9, 0, tzinfo=UTC),
+        )
+
+        self.assertIsNone(result)
+
+    def test_capture_alert_snapshot_uploads_current_frame_for_alert_events(self) -> None:
+        encode_calls: list[Frame] = []
+        upload_calls: list[tuple[bytes, str, str]] = []
+
+        def fake_encode(frame: Frame, timeout_seconds: float) -> bytes:
+            encode_calls.append(frame)
+            return b"jpeg-bytes"
+
+        def fake_upload(
+            image_bytes: bytes,
+            *,
+            file_name: str,
+            object_key_hint: str,
+            settings,
+            post_json_request=None,
+            post_multipart_request=None,
+        ) -> dict[str, object]:
+            upload_calls.append((image_bytes, file_name, object_key_hint))
+            return {"status": "uploaded", "object_key": "uploads/alarms/test.jpg", "file_name": file_name}
+
+        result = capture_alert_snapshot(
+            Frame(width=1, height=1, rgb=b"\x01\x02\x03"),
+            [{"event": "time_alarm", "severity": "alarm", "batch_id": "batch_1", "camera_id": "cam_1", "zone_id": "1", "ts": "2026-06-09T09:00:00+00:00"}],
+            {"alarm_snapshot_upload": {"enabled": True, "object_key_prefix": "alarms/cold-display"}},
+            now=datetime(2026, 6, 9, 9, 0, tzinfo=UTC),
+            jpeg_encoder=fake_encode,
+            uploader=fake_upload,
+        )
+
+        self.assertEqual(len(encode_calls), 1)
+        self.assertEqual(upload_calls[0][0], b"jpeg-bytes")
+        self.assertEqual(result["status"], "uploaded")
+        self.assertEqual(result["object_key"], "uploads/alarms/test.jpg")
+        self.assertEqual(result["batch_ids"], ["batch_1"])
+
+    def test_calibration_overlay_draws_zones_and_trash_roi_without_mutating_source(self) -> None:
+        apply_overlay = getattr(alarm_snapshots, "apply_calibration_overlay", None)
+        self.assertTrue(callable(apply_overlay), "apply_calibration_overlay should be available")
+        frame = Frame(width=5, height=5, rgb=b"\x00\x00\x00" * 25)
+
+        annotated = apply_overlay(
+            frame,
+            {
+                "zones": [
+                    {
+                        "id": "1",
+                        "label": "区域 1",
+                        "polygon": [[0.2, 0.2], [0.8, 0.2], [0.8, 0.8], [0.2, 0.8]],
+                    }
+                ],
+                "trash": {"roi": [[0.0, 0.0], [0.4, 0.0], [0.0, 0.4]]},
+            },
+        )
+
+        self.assertEqual(frame.rgb, b"\x00\x00\x00" * 25)
+        self.assertNotEqual(annotated.rgb, frame.rgb)
+        self.assertNotEqual(annotated.pixel(1, 1), (0, 0, 0))
+        self.assertNotEqual(annotated.pixel(2, 2), (0, 0, 0))
+        self.assertNotEqual(annotated.pixel(0, 0), (0, 0, 0))
+
+    def test_calibration_overlay_uses_distinct_zone_colors_and_draws_labels(self) -> None:
+        frame = Frame(width=40, height=20, rgb=b"\x00\x00\x00" * 800)
+
+        annotated = alarm_snapshots.apply_calibration_overlay(
+            frame,
+            {
+                "zones": [
+                    {
+                        "id": "1",
+                        "label": "区域 1",
+                        "polygon": [[0.05, 0.10], [0.40, 0.10], [0.40, 0.90], [0.05, 0.90]],
+                    },
+                    {
+                        "id": "2",
+                        "label": "区域 2",
+                        "polygon": [[0.55, 0.10], [0.90, 0.10], [0.90, 0.90], [0.55, 0.90]],
+                    },
+                ]
+            },
+        )
+
+        self.assertNotEqual(annotated.pixel(10, 15), annotated.pixel(30, 15))
+        label_pixels = [annotated.pixel(x, y) for y in range(2, 10) for x in range(2, 18)]
+        self.assertTrue(any(max(pixel) >= 220 for pixel in label_pixels), "expected bright label text pixels")
+
+    def test_chinese_label_fallback_uses_readable_ascii_when_font_renderer_is_unavailable(self) -> None:
+        self.assertEqual(fallback_label_text("区域 1"), "R1")
+        self.assertEqual(fallback_label_text("区域 12"), "R12")
+        self.assertEqual(fallback_label_text("垃圾区"), "TRASH")
+
+    def test_docker_image_installs_cjk_fonts_for_alarm_snapshot_labels(self) -> None:
+        dockerfile = (Path(__file__).resolve().parents[1] / "Dockerfile").read_text(encoding="utf-8")
+        self.assertIn("fonts-noto-cjk", dockerfile)
+
+    def test_drawtext_label_style_stays_small_and_translucent(self) -> None:
+        filter_text = alarm_snapshots.build_drawtext_filter(
+            [
+                alarm_snapshots.OverlayLabel(
+                    text="区域 1",
+                    fallback_text="R1",
+                    x=10,
+                    y=20,
+                    accent_rgb=(255, 196, 0),
+                )
+            ],
+            Path("/tmp/NotoSansCJK-Regular.ttc"),
+            height=360,
+        )
+
+        self.assertIn("fontsize=13", filter_text)
+        self.assertIn("boxcolor=black@0.34", filter_text)
+        self.assertIn("boxborderw=2", filter_text)
+
+    def test_capture_alert_snapshot_encodes_frame_with_configured_calibration_overlay(self) -> None:
+        encoded_frames: list[Frame] = []
+
+        def fake_encode(frame: Frame, timeout_seconds: float) -> bytes:
+            encoded_frames.append(frame)
+            return b"jpeg-bytes"
+
+        def fake_upload(
+            image_bytes: bytes,
+            *,
+            file_name: str,
+            object_key_hint: str,
+            settings,
+            post_json_request=None,
+            post_multipart_request=None,
+        ) -> dict[str, object]:
+            return {"status": "uploaded", "object_key": "uploads/alarms/overlay.jpg", "file_name": file_name}
+
+        source_frame = Frame(width=5, height=5, rgb=b"\x00\x00\x00" * 25)
+        result = capture_alert_snapshot(
+            source_frame,
+            [{"event": "time_alarm", "severity": "alarm", "batch_id": "batch_1", "camera_id": "cam_1", "zone_id": "1", "ts": "2026-06-09T09:00:00+00:00"}],
+            {
+                "alarm_snapshot_upload": {"enabled": True},
+                "zones": [
+                    {
+                        "id": "1",
+                        "label": "区域 1",
+                        "polygon": [[0.2, 0.2], [0.8, 0.2], [0.8, 0.8], [0.2, 0.8]],
+                    }
+                ],
+                "trash": {"roi": [[0.0, 0.0], [0.4, 0.0], [0.0, 0.4]]},
+            },
+            now=datetime(2026, 6, 9, 9, 0, tzinfo=UTC),
+            jpeg_encoder=fake_encode,
+            uploader=fake_upload,
+        )
+
+        self.assertEqual(result["status"], "uploaded")
+        self.assertEqual(source_frame.rgb, b"\x00\x00\x00" * 25)
+        self.assertEqual(len(encoded_frames), 1)
+        self.assertNotEqual(encoded_frames[0].rgb, source_frame.rgb)
+        self.assertNotEqual(encoded_frames[0].pixel(1, 1), (0, 0, 0))
+
+    def test_capture_alert_snapshot_only_draws_alert_event_zones(self) -> None:
+        encoded_frames: list[Frame] = []
+
+        def fake_encode(frame: Frame, timeout_seconds: float) -> bytes:
+            encoded_frames.append(frame)
+            return b"jpeg-bytes"
+
+        def fake_upload(
+            image_bytes: bytes,
+            *,
+            file_name: str,
+            object_key_hint: str,
+            settings,
+            post_json_request=None,
+            post_multipart_request=None,
+        ) -> dict[str, object]:
+            return {"status": "uploaded", "object_key": "uploads/alarms/zone-only.jpg", "file_name": file_name}
+
+        source_frame = Frame(width=30, height=20, rgb=b"\x00\x00\x00" * 600)
+        result = capture_alert_snapshot(
+            source_frame,
+            [
+                {
+                    "event": "time_alarm",
+                    "severity": "alarm",
+                    "batch_id": "batch_1",
+                    "camera_id": "cam_1",
+                    "zone_id": "1",
+                    "ts": "2026-06-09T09:00:00+00:00",
+                }
+            ],
+            {
+                "alarm_snapshot_upload": {"enabled": True},
+                "zones": [
+                    {
+                        "id": "1",
+                        "label": "区域 1",
+                        "polygon": [[0.00, 0.00], [0.45, 0.00], [0.45, 1.00], [0.00, 1.00]],
+                    },
+                    {
+                        "id": "2",
+                        "label": "区域 2",
+                        "polygon": [[0.55, 0.00], [1.00, 0.00], [1.00, 1.00], [0.55, 1.00]],
+                    },
+                ],
+                "trash": {"roi": [[0.45, 0.50], [0.55, 0.50], [0.55, 1.00], [0.45, 1.00]]},
+            },
+            now=datetime(2026, 6, 9, 9, 0, tzinfo=UTC),
+            jpeg_encoder=fake_encode,
+            uploader=fake_upload,
+        )
+
+        self.assertEqual(result["status"], "uploaded")
+        self.assertEqual(len(encoded_frames), 1)
+        self.assertNotEqual(encoded_frames[0].pixel(5, 10), (0, 0, 0))
+        self.assertEqual(encoded_frames[0].pixel(25, 10), (0, 0, 0))
+        self.assertEqual(encoded_frames[0].pixel(15, 15), (0, 0, 0))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_cases.py

@@ -48,6 +48,55 @@ class CaseStoreTests(unittest.TestCase):
         self.assertEqual(snapshots[0]["case_status"], "open")
         self.assertEqual(snapshots[0]["source_event"], "time_alarm")
 
+    def test_time_pre_warning_creates_open_pre_warning_case(self) -> None:
+        store = CaseStore()
+
+        snapshots = store.apply_batch_events(
+            [event("time_pre_warning", self.t0, severity="warning", state="pre_warning")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_type"], "pre_warning")
+        self.assertEqual(snapshots[0]["case_status"], "open")
+        self.assertEqual(snapshots[0]["source_event"], "time_pre_warning")
+
+    def test_pre_warning_handled_auto_closes_open_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_pre_warning", self.t0, severity="warning", state="pre_warning")])
+
+        snapshots = store.apply_batch_events(
+            [event("pre_warning_handled", self.t0.replace(minute=1), severity="info", state="handled")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_status"], "handled")
+        self.assertEqual(snapshots[0]["handled_source"], "auto_removed_before_alarm")
+
+    def test_time_alarm_upgrades_pre_warning_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_pre_warning", self.t0, severity="warning", state="pre_warning")])
+
+        snapshots = store.apply_batch_events([event("time_alarm", self.t0.replace(minute=2), severity="alarm", state="alerted")])
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_type"], "time_alarm")
+        self.assertEqual(snapshots[0]["case_status"], "open")
+        self.assertEqual(snapshots[0]["source_event"], "time_alarm")
+
+    def test_alarm_removal_timeout_upgrades_same_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_pre_warning", self.t0, severity="warning", state="pre_warning")])
+        store.apply_batch_events([event("time_alarm", self.t0.replace(minute=2), severity="alarm", state="alerted")])
+
+        snapshots = store.apply_batch_events(
+            [event("alarm_removal_timeout", self.t0.replace(minute=3), severity="alarm", state="alarm_removal_timeout")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_type"], "alarm_removal_timeout")
+        self.assertEqual(snapshots[0]["case_status"], "open")
+        self.assertEqual(snapshots[0]["source_event"], "alarm_removal_timeout")
+
     def test_pending_disposal_upgrades_existing_case(self) -> None:
         store = CaseStore()
         store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])

tests/test_config.py

@@ -16,7 +16,9 @@ class ConfigTests(unittest.TestCase):
 camera_id = "cam_a"
 
 [thresholds]
+pre_warning_seconds = 20
 max_dwell_seconds = 30
+alarm_removal_seconds = 2
 trash_confirmation_seconds = 4
 
 [layout]
@@ -29,7 +31,9 @@ cols = 2
             settings = load_settings(path)
 
         self.assertEqual(settings.camera_id, "cam_a")
+        self.assertEqual(settings.pre_warning_seconds, 20)
         self.assertEqual(settings.max_dwell_seconds, 30)
+        self.assertEqual(settings.alarm_removal_seconds, 2)
         self.assertEqual(settings.trash_confirmation_seconds, 4)
         self.assertEqual(settings.zone_ids, ("r1c1", "r1c2"))
 
@@ -118,6 +122,8 @@ zone_ids = ["1", "2", "3"]
             text = path.read_text(encoding="utf-8")
 
         self.assertIn("zone_count = 2", text)
+        self.assertIn("pre_warning_seconds = 0", text)
+        self.assertIn("alarm_removal_seconds = 0", text)
         self.assertIn('label = "区域 1"', text)
         self.assertIn("[trash]", text)
         self.assertNotIn('"trash"', text.split("[layout]", maxsplit=1)[1].split("[[zones]]", maxsplit=1)[0])
@@ -128,10 +134,17 @@ zone_ids = ["1", "2", "3"]
             save_config_document(
                 path,
                 {
+                    "alarm_snapshot_upload": {
+                        "enabled": True,
+                        "service_url": "https://ota.zhengxinshipin.com",
+                        "secret": "change-me-in-production",
+                        "object_key_prefix": "cold-display/alarms",
+                    },
                     "webhooks": {
                         "enabled": True,
                         "event_url": "https://example.com/events",
                         "case_url": "https://example.com/cases",
+                        "source_id": "cold-display-guard",
                         "callback_token": "secret",
                         "connect_timeout_seconds": 3,
                         "read_timeout_seconds": 5,
@@ -146,9 +159,14 @@ zone_ids = ["1", "2", "3"]
             )
             text = path.read_text(encoding="utf-8")
 
+        self.assertIn("[alarm_snapshot_upload]", text)
+        self.assertIn('service_url = "https://ota.zhengxinshipin.com"', text)
+        self.assertIn('secret = "change-me-in-production"', text)
+        self.assertIn('object_key_prefix = "cold-display/alarms"', text)
         self.assertIn("[webhooks]", text)
         self.assertIn('event_url = "https://example.com/events"', text)
         self.assertIn('case_url = "https://example.com/cases"', text)
+        self.assertIn('source_id = "cold-display-guard"', text)
         self.assertIn('callback_token = "secret"', text)
         self.assertIn("retry_max_attempts = 4", text)
         self.assertIn("retry_backoff_seconds = 30", text)

tests/test_engine.py

@@ -140,6 +140,96 @@ class BatchEngineTests(unittest.TestCase):
         self.assertEqual(alarm_events[0]["current_count"], 1)
         self.assertIn("alerted_at", alarm_events[0])
 
+    def test_pre_warning_emits_once_before_alarm_threshold(self) -> None:
+        settings = EngineSettings(
+            camera_id="test_cam",
+            pre_warning_seconds=60,
+            max_dwell_seconds=120,
+            alarm_removal_seconds=30,
+            trash_confirmation_seconds=30,
+            zone_ids=("1",),
+        )
+        engine = BatchEngine(settings)
+        engine.process(obs(self.t0, {"1": 1}))
+
+        pre_warning_events = engine.process(obs(self.t0 + timedelta(seconds=60), {"1": 1}))
+        repeated_events = engine.process(obs(self.t0 + timedelta(seconds=90), {"1": 1}))
+
+        self.assertEqual([event["event"] for event in pre_warning_events], ["time_pre_warning"])
+        self.assertEqual(pre_warning_events[0]["severity"], "warning")
+        self.assertEqual(pre_warning_events[0]["state"], "pre_warning")
+        self.assertEqual(pre_warning_events[0]["pre_warning_seconds"], 60)
+        self.assertEqual(pre_warning_events[0]["pre_warned_at"], (self.t0 + timedelta(seconds=60)).isoformat())
+        self.assertEqual(pre_warning_events[0]["current_count"], 1)
+        self.assertEqual(repeated_events, [])
+
+    def test_pre_warning_removed_before_alarm_is_auto_handled(self) -> None:
+        settings = EngineSettings(
+            camera_id="test_cam",
+            pre_warning_seconds=60,
+            max_dwell_seconds=120,
+            alarm_removal_seconds=30,
+            trash_confirmation_seconds=30,
+            zone_ids=("1",),
+        )
+        engine = BatchEngine(settings)
+        engine.process(obs(self.t0, {"1": 1}))
+        engine.process(obs(self.t0 + timedelta(seconds=60), {"1": 1}))
+
+        events = engine.process(obs(self.t0 + timedelta(seconds=90), {"1": 0}))
+
+        self.assertEqual([event["event"] for event in events], ["pre_warning_handled"])
+        self.assertEqual(events[0]["severity"], "info")
+        self.assertEqual(events[0]["state"], "handled")
+        self.assertEqual(events[0]["handled_source"], "auto_removed_before_alarm")
+        self.assertEqual(events[0]["ended_at"], (self.t0 + timedelta(seconds=90)).isoformat())
+
+    def test_alarm_removal_timeout_emits_once_before_late_removal(self) -> None:
+        settings = EngineSettings(
+            camera_id="test_cam",
+            pre_warning_seconds=60,
+            max_dwell_seconds=120,
+            alarm_removal_seconds=30,
+            trash_confirmation_seconds=30,
+            zone_ids=("1",),
+        )
+        engine = BatchEngine(settings)
+        engine.process(obs(self.t0, {"1": 1}))
+        engine.process(obs(self.t0 + timedelta(seconds=60), {"1": 1}))
+        alarm_events = engine.process(obs(self.t0 + timedelta(seconds=120), {"1": 1}))
+
+        timeout_events = engine.process(obs(self.t0 + timedelta(seconds=151), {"1": 1}))
+        repeated_events = engine.process(obs(self.t0 + timedelta(seconds=160), {"1": 1}))
+        removal_events = engine.process(obs(self.t0 + timedelta(seconds=170), {"1": 0}, trash=True))
+
+        self.assertEqual([event["event"] for event in alarm_events], ["time_alarm"])
+        self.assertEqual(alarm_events[0]["alarm_removal_deadline"], (self.t0 + timedelta(seconds=150)).isoformat())
+        self.assertEqual([event["event"] for event in timeout_events], ["alarm_removal_timeout"])
+        self.assertEqual(timeout_events[0]["severity"], "alarm")
+        self.assertEqual(timeout_events[0]["state"], "alarm_removal_timeout")
+        self.assertEqual(timeout_events[0]["reason"], "alarmed_batch_not_removed_after_alarm_window")
+        self.assertEqual(repeated_events, [])
+        self.assertEqual([event["event"] for event in removal_events], ["batch_pending_disposal", "batch_discarded"])
+
+    def test_alarmed_batch_removed_within_alarm_window_does_not_emit_removal_timeout(self) -> None:
+        settings = EngineSettings(
+            camera_id="test_cam",
+            pre_warning_seconds=60,
+            max_dwell_seconds=120,
+            alarm_removal_seconds=30,
+            trash_confirmation_seconds=30,
+            zone_ids=("1",),
+        )
+        engine = BatchEngine(settings)
+        engine.process(obs(self.t0, {"1": 1}))
+        engine.process(obs(self.t0 + timedelta(seconds=60), {"1": 1}))
+        engine.process(obs(self.t0 + timedelta(seconds=120), {"1": 1}))
+
+        events = engine.process(obs(self.t0 + timedelta(seconds=150), {"1": 0}, trash=True))
+
+        self.assertEqual([event["event"] for event in events], ["batch_pending_disposal", "batch_discarded"])
+        self.assertTrue(all(event["event"] != "alarm_removal_timeout" for event in events))
+
     def test_alarmed_batch_removed_without_trash_deposit_escalates_warning(self) -> None:
         settings = EngineSettings(
             camera_id="test_cam",
@@ -260,6 +350,37 @@ class BatchEngineTests(unittest.TestCase):
         self.assertEqual(removal_events[0]["batch_id"], "batch_000124")
         self.assertEqual(removal_events[0]["dwell_seconds"], 1400)
 
+    def test_restore_keeps_alarm_removal_timeout_deadline_after_runtime_restart(self) -> None:
+        settings = EngineSettings(
+            camera_id="test_cam",
+            pre_warning_seconds=60,
+            max_dwell_seconds=120,
+            alarm_removal_seconds=30,
+            trash_confirmation_seconds=30,
+            zone_ids=("1",),
+        )
+        engine = BatchEngine(settings)
+        engine.restore_from_events(
+            [
+                {
+                    "event": "time_alarm",
+                    "zone_id": "1",
+                    "batch_id": "batch_000124",
+                    "started_at": self.t0.isoformat(),
+                    "pre_warned_at": (self.t0 + timedelta(seconds=60)).isoformat(),
+                    "alerted_at": (self.t0 + timedelta(seconds=120)).isoformat(),
+                    "current_count": 1,
+                    "state": "alerted",
+                },
+            ],
+            active_zone_counts={"1": 1},
+        )
+
+        timeout_events = engine.process(obs(self.t0 + timedelta(seconds=151), {"1": 1}))
+
+        self.assertEqual([event["event"] for event in timeout_events], ["alarm_removal_timeout"])
+        self.assertEqual(timeout_events[0]["batch_id"], "batch_000124")
+
     def test_restore_skips_active_false_positive_when_latest_zone_count_is_empty(self) -> None:
         settings = EngineSettings(
             camera_id="test_cam",

tests/test_main.py

@@ -6,14 +6,16 @@ import unittest
 from datetime import datetime, timezone
 from pathlib import Path
 
-from cold_display_guard.cases import CaseStore
+from cold_display_guard.cases import CaseStore, append_case_snapshots, load_case_snapshots
 from cold_display_guard.main import (
     case_sink_path,
+    capture_runtime_alarm_snapshot,
     deliver_runtime_webhooks,
     persist_case_updates,
     restore_runtime_state,
     webhook_retry_sink_path,
 )
+from cold_display_guard.vision import Frame
 from cold_display_guard.webhooks import load_retry_snapshots
 
 
@@ -65,6 +67,61 @@ class RuntimeRestoreTests(unittest.TestCase):
         self.assertEqual(written[0]["case_type"], "time_alarm")
         self.assertEqual(written[0]["case_status"], "open")
 
+    def test_persist_case_updates_preserves_api_handled_snapshot(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "cases.jsonl"
+            runtime_store = CaseStore()
+            created = persist_case_updates(
+                runtime_store,
+                path,
+                [
+                    {
+                        "event": "time_alarm",
+                        "ts": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                        "batch_id": "batch_000001",
+                        "camera_id": "cam_01",
+                        "zone_id": "1",
+                        "zone_label": "区域 1",
+                        "severity": "alarm",
+                        "state": "alerted",
+                    }
+                ],
+            )[0]
+            api_store = CaseStore(load_case_snapshots(path))
+            append_case_snapshots(
+                path,
+                [
+                    api_store.mark_handled(
+                        str(created["case_id"]),
+                        handled_at=datetime(2026, 6, 9, 9, 5, tzinfo=UTC),
+                        handled_by="alice",
+                        handled_source="manual",
+                    )
+                ],
+            )
+
+            snapshots = persist_case_updates(
+                runtime_store,
+                path,
+                [
+                    {
+                        "event": "batch_pending_disposal",
+                        "ts": datetime(2026, 6, 9, 9, 6, tzinfo=UTC).isoformat(),
+                        "batch_id": "batch_000001",
+                        "camera_id": "cam_01",
+                        "zone_id": "1",
+                        "zone_label": "区域 1",
+                        "severity": "warning",
+                        "state": "pending_disposal",
+                    }
+                ],
+            )
+            latest = CaseStore(load_case_snapshots(path)).latest_cases()[0]
+
+        self.assertEqual(snapshots, [])
+        self.assertEqual(latest["case_status"], "handled")
+        self.assertEqual(latest["handled_source"], "manual")
+
     def test_deliver_runtime_webhooks_sends_event_and_case_payloads(self) -> None:
         deliveries: list[tuple[str, dict[str, object]]] = []
 
@@ -114,6 +171,29 @@ class RuntimeRestoreTests(unittest.TestCase):
         self.assertEqual(deliveries[0][1]["kind"], "batch_event")
         self.assertEqual(deliveries[1][1]["kind"], "case_event")
 
+    def test_capture_runtime_alarm_snapshot_uses_current_frame_for_alert_events(self) -> None:
+        frame = Frame(width=1, height=1, rgb=b"\x01\x02\x03")
+        result = capture_runtime_alarm_snapshot(
+            frame,
+            [
+                {
+                    "event": "time_alarm",
+                    "severity": "alarm",
+                    "batch_id": "batch_000001",
+                    "camera_id": "cam_01",
+                    "zone_id": "1",
+                    "ts": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                }
+            ],
+            {"alarm_snapshot_upload": {"enabled": True}},
+            now=datetime(2026, 6, 9, 9, 0, tzinfo=UTC),
+            jpeg_encoder=lambda frame, timeout_seconds: b"jpeg",
+            uploader=lambda image_bytes, **kwargs: {"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "file_name": "a.jpg"},
+        )
+
+        self.assertEqual(result["status"], "uploaded")
+        self.assertEqual(result["object_key"], "uploads/alarms/a.jpg")
+
     def test_deliver_runtime_webhooks_enqueues_failure_and_drains_due_retry(self) -> None:
         attempts = {"count": 0}
 
@@ -169,6 +249,53 @@ class RuntimeRestoreTests(unittest.TestCase):
         self.assertEqual(retries[-1]["status"], "delivered")
         self.assertEqual(retries[-1]["attempt_count"], 2)
 
+    def test_deliver_runtime_webhooks_includes_snapshot_path_in_alert_payloads(self) -> None:
+        deliveries: list[dict[str, object]] = []
+
+        def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:
+            deliveries.append(payload)
+            return 200, "ok"
+
+        deliver_runtime_webhooks(
+            [
+                {
+                    "event": "time_alarm",
+                    "ts": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                    "batch_id": "batch_000001",
+                    "camera_id": "cam_01",
+                    "zone_id": "1",
+                    "zone_label": "区域 1",
+                    "severity": "alarm",
+                    "state": "alerted",
+                }
+            ],
+            [
+                {
+                    "case_id": "case_batch_000001",
+                    "batch_id": "batch_000001",
+                    "case_type": "time_alarm",
+                    "case_status": "open",
+                    "source_event": "time_alarm",
+                    "handled_source": "",
+                    "created_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                    "updated_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                }
+            ],
+            {
+                "webhooks": {
+                    "enabled": True,
+                    "event_url": "https://example.com/events",
+                    "case_url": "https://example.com/cases",
+                }
+            },
+            Path(tempfile.mkdtemp()) / "webhook_delivery.jsonl",
+            http_post=fake_post,
+            snapshot_upload={"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "batch_ids": ["batch_000001"]},
+        )
+
+        self.assertEqual(deliveries[0]["snapshot_object_key"], "uploads/alarms/a.jpg")
+        self.assertEqual(deliveries[1]["snapshot_object_key"], "uploads/alarms/a.jpg")
+
     def test_restore_runtime_state_uses_stable_occupancy_when_raw_metrics_flicker(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             diagnostics_path = Path(tmpdir) / "runtime_diagnostics.jsonl"

tests/test_manage_api.py

@@ -392,6 +392,11 @@ class ManageApiTests(unittest.TestCase):
                 {
                     "case_sink": {"path": "logs/cases.jsonl"},
                     "webhook_retry_sink": {"path": "logs/webhook_retry.jsonl"},
+                    "alarm_snapshot_upload": {
+                        "enabled": True,
+                        "service_url": "https://ota.zhengxinshipin.com",
+                        "secret": "change-me-in-production",
+                    },
                     "webhooks": {
                         "enabled": True,
                         "event_url": "https://example.com/events",
@@ -406,6 +411,9 @@ class ManageApiTests(unittest.TestCase):
 
         self.assertEqual(payload["case_sink"]["path"], str((root / "logs" / "cases.jsonl").resolve()))
         self.assertEqual(payload["webhook_retry_sink"]["path"], str((root / "logs" / "webhook_retry.jsonl").resolve()))
+        self.assertTrue(payload["alarm_snapshot_upload"]["enabled"])
+        self.assertEqual(payload["alarm_snapshot_upload"]["service_url"], "https://ota.zhengxinshipin.com")
+        self.assertNotIn("secret", payload["alarm_snapshot_upload"])
         self.assertTrue(payload["webhooks"]["enabled"])
         self.assertEqual(payload["webhooks"]["retry_max_attempts"], 4)
         self.assertNotIn("callback_token", payload["webhooks"])

tests/test_vision.py

@@ -10,6 +10,7 @@ from cold_display_guard.vision import (
     RuntimeVisionSettings,
     ZoneOccupancyDetector,
     load_runtime_vision_settings,
+    metrics_indicate_occupied,
     point_in_polygon,
 )
 
@@ -157,6 +158,7 @@ class VisionTests(unittest.TestCase):
 
         self.assertEqual(settings.sample_stride_pixels, 4)
         self.assertEqual(settings.occupancy_mean_delta, 55.0)
+        self.assertEqual(settings.occupancy_absolute_dark_fraction, 0.0)
         self.assertEqual(settings.occupancy_confirm_frames, 2)
         self.assertEqual(settings.empty_confirm_frames, 2)
         self.assertEqual(settings.trash_motion_delta, 18.0)
@@ -164,6 +166,25 @@ class VisionTests(unittest.TestCase):
         self.assertEqual(settings.trash_sustained_motion_frames, 2)
         self.assertEqual(settings.trash_motion_cooldown_seconds, 3)
 
+    def test_absolute_dark_fraction_can_detect_food_already_present_in_baseline(self) -> None:
+        settings = RuntimeVisionSettings(
+            occupancy_mean_delta=55,
+            occupancy_texture_delta=18,
+            occupancy_dark_fraction=0.06,
+            occupancy_absolute_dark_fraction=0.085,
+        )
+
+        occupied = metrics_indicate_occupied(
+            settings,
+            mean_delta=5.0,
+            texture_delta=0.5,
+            dark_fraction=0.09,
+            baseline_dark_fraction=0.10,
+            bright_fraction=0.0,
+        )
+
+        self.assertTrue(occupied)
+
     def test_detector_can_seed_previous_baseline_and_occupancy(self) -> None:
         detector = ZoneOccupancyDetector(
             [Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],

tests/test_webhooks.py

@@ -28,6 +28,7 @@ class WebhookTests(unittest.TestCase):
                     "enabled": True,
                     "event_url": "https://example.com/events",
                     "case_url": "https://example.com/cases",
+                    "source_id": "cold-display-guard",
                     "callback_token": "secret",
                     "connect_timeout_seconds": 4,
                     "read_timeout_seconds": 6,
@@ -42,6 +43,7 @@ class WebhookTests(unittest.TestCase):
         self.assertTrue(settings.enabled)
         self.assertEqual(settings.event_url, "https://example.com/events")
         self.assertEqual(settings.case_url, "https://example.com/cases")
+        self.assertEqual(settings.source_id, "cold-display-guard")
         self.assertEqual(settings.callback_token, "secret")
         self.assertEqual(settings.connect_timeout_seconds, 4)
         self.assertEqual(settings.read_timeout_seconds, 6)
@@ -61,14 +63,123 @@ class WebhookTests(unittest.TestCase):
                 "zone_label": "区域 1",
                 "severity": "alarm",
                 "state": "alerted",
-            }
+                "started_at": datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat(),
+                "dwell_seconds": 1200,
+                "alerted_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+            },
+            camera_ip="192.168.3.4",
         )
 
         self.assertEqual(payload["kind"], "batch_event")
         self.assertEqual(payload["event"], "time_alarm")
+        self.assertEqual(payload["event_code"], "batch_000001")
+        self.assertEqual(payload["camera_ip"], "192.168.3.4")
         self.assertEqual(payload["zone_label"], "区域 1")
+        self.assertEqual(payload["started_at"], datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat())
+        self.assertEqual(payload["dwell_seconds"], 1200)
+        self.assertFalse(payload["is_discarded"])
+        self.assertEqual(payload["alerted_at"], datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat())
+
+    def test_build_batch_event_payload_preserves_pre_warning_and_alarm_times(self) -> None:
+        pre_warned_at = datetime(2026, 6, 9, 8, 59, tzinfo=UTC).isoformat()
+        alarm_at = datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat()
+
+        pre_warning_payload = build_batch_event_payload(
+            {
+                "event": "time_pre_warning",
+                "ts": pre_warned_at,
+                "batch_id": "batch_000001",
+                "camera_id": "cam_01",
+                "zone_id": "1",
+                "zone_label": "区域 1",
+                "severity": "warning",
+                "state": "pre_warning",
+                "started_at": datetime(2026, 6, 9, 8, 58, tzinfo=UTC).isoformat(),
+                "pre_warned_at": pre_warned_at,
+            }
+        )
+        alarm_payload = build_batch_event_payload(
+            {
+                "event": "time_alarm",
+                "ts": alarm_at,
+                "batch_id": "batch_000001",
+                "camera_id": "cam_01",
+                "zone_id": "1",
+                "zone_label": "区域 1",
+                "severity": "alarm",
+                "state": "alerted",
+                "started_at": datetime(2026, 6, 9, 8, 58, tzinfo=UTC).isoformat(),
+                "pre_warned_at": pre_warned_at,
+                "alerted_at": alarm_at,
+            }
+        )
+
+        self.assertEqual(pre_warning_payload["pre_warned_at"], pre_warned_at)
+        self.assertEqual(pre_warning_payload["created_at"], pre_warned_at)
+        self.assertEqual(pre_warning_payload["alarm_at"], "")
+        self.assertEqual(alarm_payload["pre_warned_at"], pre_warned_at)
+        self.assertEqual(alarm_payload["alarm_at"], alarm_at)
+
+    def test_build_batch_event_payload_includes_uploaded_snapshot_path(self) -> None:
+        payload = build_batch_event_payload(
+            {
+                "event": "time_alarm",
+                "ts": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                "batch_id": "batch_000001",
+                "camera_id": "cam_01",
+                "zone_id": "1",
+                "zone_label": "区域 1",
+                "severity": "alarm",
+                "state": "alerted",
+            },
+            snapshot_upload={"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "batch_ids": ["batch_000001"]},
+        )
+
+        self.assertEqual(payload["snapshot_upload_status"], "uploaded")
+        self.assertEqual(payload["snapshot_object_key"], "uploads/alarms/a.jpg")
 
     def test_build_case_event_payload_wraps_case_snapshot(self) -> None:
+        payload = build_case_event_payload(
+            {
+                "case_id": "case_batch_000001",
+                "case_type": "warning_escalated",
+                "case_status": "handled",
+                "batch_id": "batch_000001",
+                "camera_id": "cam_01",
+                "zone_id": "1",
+                "zone_label": "区域 1",
+                "source_event": "warning_escalated",
+                "created_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                "handled_at": datetime(2026, 6, 9, 9, 6, tzinfo=UTC).isoformat(),
+                "handled_source": "auto_closed",
+                "updated_at": datetime(2026, 6, 9, 9, 5, tzinfo=UTC).isoformat(),
+                "payload": {
+                    "event": {
+                        "started_at": datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat(),
+                        "ended_at": datetime(2026, 6, 9, 9, 4, tzinfo=UTC).isoformat(),
+                        "dwell_seconds": 1440,
+                        "alerted_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
+                    }
+                },
+            },
+            camera_ip="192.168.3.4",
+        )
+
+        self.assertEqual(payload["kind"], "case_event")
+        self.assertEqual(payload["action"], "handled")
+        self.assertEqual(payload["case_id"], "case_batch_000001")
+        self.assertEqual(payload["event_code"], "batch_000001")
+        self.assertEqual(payload["camera_id"], "cam_01")
+        self.assertEqual(payload["camera_ip"], "192.168.3.4")
+        self.assertEqual(payload["zone_label"], "区域 1")
+        self.assertEqual(payload["started_at"], datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat())
+        self.assertEqual(payload["ended_at"], datetime(2026, 6, 9, 9, 4, tzinfo=UTC).isoformat())
+        self.assertEqual(payload["dwell_seconds"], 1440)
+        self.assertTrue(payload["is_discarded"])
+        self.assertEqual(payload["discarded_at"], datetime(2026, 6, 9, 9, 6, tzinfo=UTC).isoformat())
+        self.assertEqual(payload["alerted_at"], datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat())
+
+    def test_build_case_event_payload_includes_uploaded_snapshot_path(self) -> None:
         payload = build_case_event_payload(
             {
                 "case_id": "case_batch_000001",
@@ -78,12 +189,12 @@ class WebhookTests(unittest.TestCase):
                 "source_event": "warning_escalated",
                 "handled_source": "",
                 "updated_at": datetime(2026, 6, 9, 9, 5, tzinfo=UTC).isoformat(),
-            }
+            },
+            snapshot_upload={"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "batch_ids": ["batch_000001"]},
         )
 
-        self.assertEqual(payload["kind"], "case_event")
-        self.assertEqual(payload["action"], "updated")
-        self.assertEqual(payload["case_id"], "case_batch_000001")
+        self.assertEqual(payload["snapshot_upload_status"], "uploaded")
+        self.assertEqual(payload["snapshot_object_key"], "uploads/alarms/a.jpg")
 
     def test_send_batch_event_webhooks_delivers_payload(self) -> None:
         deliveries: list[tuple[str, dict[str, object], tuple[float, float]]] = []
@@ -111,9 +222,11 @@ class WebhookTests(unittest.TestCase):
                     "webhooks": {
                         "enabled": True,
                         "event_url": "https://example.com/events",
+                        "source_id": "cold-display-guard",
                         "connect_timeout_seconds": 4,
                         "read_timeout_seconds": 6,
-                    }
+                    },
+                    "stream": {"rtsp_url": "rtsp://admin:secret@192.168.3.4:554/h264/ch1/main/av_stream"},
                 },
                 audit_path,
                 http_post=fake_post,
@@ -121,6 +234,8 @@ class WebhookTests(unittest.TestCase):
 
         self.assertEqual(deliveries[0][0], "https://example.com/events")
         self.assertEqual(deliveries[0][1]["kind"], "batch_event")
+        self.assertEqual(deliveries[0][1]["camera_ip"], "192.168.3.4")
+        self.assertEqual(deliveries[0][1]["source_id"], "cold-display-guard")
         self.assertEqual(deliveries[0][2], (4.0, 6.0))
 
     def test_send_case_webhooks_delivers_payload(self) -> None:
@@ -148,7 +263,9 @@ class WebhookTests(unittest.TestCase):
                     "webhooks": {
                         "enabled": True,
                         "case_url": "https://example.com/cases",
-                    }
+                        "source_id": "cold-display-guard",
+                    },
+                    "stream": {"rtsp_url": "rtsp://admin:secret@192.168.3.4:554/h264/ch1/main/av_stream"},
                 },
                 audit_path,
                 http_post=fake_post,
@@ -157,6 +274,8 @@ class WebhookTests(unittest.TestCase):
         self.assertEqual(deliveries[0][0], "https://example.com/cases")
         self.assertEqual(deliveries[0][1]["kind"], "case_event")
         self.assertEqual(deliveries[0][1]["action"], "handled")
+        self.assertEqual(deliveries[0][1]["camera_ip"], "192.168.3.4")
+        self.assertEqual(deliveries[0][1]["source_id"], "cold-display-guard")
 
     def test_failed_delivery_is_logged_without_raising(self) -> None:
         def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:

web/nginx.conf

@@ -2,11 +2,14 @@ server {
     listen 80;
     server_name _;
 
+    resolver 127.0.0.11 ipv6=off valid=10s;
+
     root /usr/share/nginx/html;
     index index.html;
 
     location /api/ {
-        proxy_pass http://cold-display-guard-api:19080;
+        set $api_upstream http://cold-display-guard-api:19080;
+        proxy_pass $api_upstream;
         proxy_http_version 1.1;
         proxy_set_header Host $host;
         proxy_set_header X-Real-IP $remote_addr;