diff --git a/README_zh.md b/README_zh.md
index 983ff4c..dcdf907 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -97,6 +97,7 @@ http://127.0.0.1:23000
 - 标定数字食品区域和垃圾桶 ROI
 - 直接保存标定结果到项目配置文件
 - 查看事件汇总、区域序号、停留时间、报警和警告事件
+- 查看本地处置单状态，并手工标记为已处理
 
 项目仍保留 `tools/calibrator` 作为轻量单页标定工具，但正式使用建议走 `23000` 管理页。
 
@@ -117,6 +118,12 @@ http://127.0.0.1:19080
 - `PUT /api/manage/calibration`
 - `GET /api/manage/summary`
 - `GET /api/manage/events`
+- `GET /api/manage/cases`
+- `GET /api/manage/cases/summary`
+- `POST /api/manage/cases/{case_id}/handle`
+- `POST /api/manage/webhooks/case-update`
+
+`/api/manage/webhooks/case-update` 需要请求头 `X-Webhook-Token`，并且请求体里的 `status` 目前固定为 `handled`。
 
 ## 运行识别计时进程
 
@@ -133,7 +140,9 @@ scripts/run_runtime.sh
 3. 按标定区域做占用变化检测。
 4. 判断垃圾桶区域是否有明显投放动作。
 5. 调用批次计时状态机。
-6. 写入 `logs/events.jsonl`，管理页会读取这个文件。
+6. 将 `time_alarm`、`batch_pending_disposal`、`warning_escalated` 映射到本地处置单状态。
+7. 写入 `logs/events.jsonl`、`logs/cases.jsonl`、`logs/runtime_diagnostics.jsonl`。
+8. 按配置向外部系统推送事件 webhook 和处置单 webhook。
 
 当前视觉版本是可运行的启发式版本：
 
@@ -168,8 +177,24 @@ trash_sustained_motion_delta = 8.0
 trash_sustained_motion_frames = 2
 trash_motion_cooldown_seconds = 3
 diagnostics_path = "logs/runtime_diagnostics.jsonl"
+
+[case_sink]
+path = "logs/cases.jsonl"
+
+[webhooks]
+enabled = true
+event_url = "https://example.com/runtime-events"
+case_url = "https://example.com/case-events"
+callback_token = "shared-secret"
+connect_timeout_seconds = 3
+read_timeout_seconds = 5
 ```
 
+运行时会额外记录：
+
+- `logs/cases.jsonl`：本地处置单状态变更
+- `logs/webhook_delivery.jsonl`：Webhook 投递结果审计
+
 ## 本地测试
 
 ```bash
diff --git a/docs/superpowers/plans/2026-06-09-webhook-case-management-implementation.md b/docs/superpowers/plans/2026-06-09-webhook-case-management-implementation.md
new file mode 100644
index 0000000..5c08ce8
--- /dev/null
+++ b/docs/superpowers/plans/2026-06-09-webhook-case-management-implementation.md
@@ -0,0 +1,112 @@
+# Webhook Case Management 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:** Build local case management plus outbound/inbound webhook support on top of the existing runtime batch-event flow.
+
+**Architecture:** Keep `BatchEngine` as the factual event source, then add a separate case-state module that consumes selected events and persists case snapshots. Add a webhook delivery module for both batch events and case events, expose management APIs for case listing and handling, and render the resulting case workflow in the existing management console without mixing facts and workflow state.
+
+**Tech Stack:** Python 3.12 via pyenv, Python standard library HTTP/JSON/TOML stack, JSONL files, unittest, Vite + vanilla JavaScript, Node test runner.
+
+---
+
+## File Map
+
+- Create: `src/cold_display_guard/cases.py`
+- Create: `src/cold_display_guard/webhooks.py`
+- Modify: `src/cold_display_guard/config.py`
+- Modify: `src/cold_display_guard/main.py`
+- Modify: `src/cold_display_guard/manage_api.py`
+- Modify: `web/src/main.js`
+- Modify: `web/src/zone-state.js`
+- Create: `tests/test_cases.py`
+- Create: `tests/test_webhooks.py`
+- Modify: `tests/test_manage_api.py`
+- Modify: `tests/test_main.py`
+- Modify: `web/test/zone-state.test.js`
+
+### Task 1: Backend Case State Layer
+
+**Files:**
+- Create: `src/cold_display_guard/cases.py`
+- Create: `tests/test_cases.py`
+- Modify: `src/cold_display_guard/main.py`
+- Modify: `tests/test_main.py`
+
+- [ ] Write failing tests for case creation, case escalation, manual/callback/auto close, and restore behavior in `tests/test_cases.py`.
+- [ ] Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_cases.py -v`
+Expected: failing assertions or import errors for missing case helpers.
+- [ ] Implement minimal case dataclasses, JSONL load/save helpers, event-to-case transitions, and restore logic in `src/cold_display_guard/cases.py`.
+- [ ] Wire runtime event processing in `src/cold_display_guard/main.py` so emitted batch events produce persisted case snapshots.
+- [ ] Run:
+  - `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`
+Expected: PASS.
+
+### Task 2: Webhook Configuration And Delivery
+
+**Files:**
+- Create: `src/cold_display_guard/webhooks.py`
+- Create: `tests/test_webhooks.py`
+- Modify: `src/cold_display_guard/config.py`
+- Modify: `src/cold_display_guard/main.py`
+
+- [ ] Write failing tests for webhook config parsing, batch event payload delivery, case event payload delivery, and delivery-failure logging in `tests/test_webhooks.py`.
+- [ ] Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_webhooks.py -v`
+Expected: FAIL because webhook helpers/config support do not exist yet.
+- [ ] Implement webhook settings parsing/saving in `src/cold_display_guard/config.py` and synchronous delivery plus audit logging in `src/cold_display_guard/webhooks.py`.
+- [ ] Integrate webhook sending into `src/cold_display_guard/main.py` after local event and case persistence.
+- [ ] Run:
+  - `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`
+Expected: PASS.
+
+### Task 3: Management API For Cases And Callback Handling
+
+**Files:**
+- Modify: `src/cold_display_guard/manage_api.py`
+- Modify: `tests/test_manage_api.py`
+- Modify: `src/cold_display_guard/config.py`
+
+- [ ] Write failing API tests for `/api/manage/cases`, `/api/manage/cases/summary`, `/api/manage/cases/{case_id}/handle`, and `/api/manage/webhooks/case-update` in `tests/test_manage_api.py`.
+- [ ] Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
+Expected: FAIL because the new endpoints and case summary behavior are missing.
+- [ ] Implement case listing, case summary, manual handle, and token-protected callback handling in `src/cold_display_guard/manage_api.py`.
+- [ ] Ensure config payloads expose webhook settings and case/log sink paths without leaking secrets unnecessarily.
+- [ ] Run: `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
+Expected: PASS.
+
+### Task 4: Frontend Case Management UI
+
+**Files:**
+- Modify: `web/src/main.js`
+- Modify: `web/src/zone-state.js`
+- Modify: `web/test/zone-state.test.js`
+
+- [ ] Write failing frontend tests for case summary mapping, case table rendering helpers, event/case separation, and manual handle request shaping in `web/test/zone-state.test.js`.
+- [ ] Run: `node --test web/test/zone-state.test.js`
+Expected: FAIL because case helpers and UI state handling do not exist yet.
+- [ ] Implement frontend model helpers and UI rendering for case summaries, case rows, and manual handle actions while preserving the existing runtime event table semantics.
+- [ ] Run:
+  - `node --test web/test/zone-state.test.js`
+  - `cd web && pnpm build`
+Expected: PASS.
+
+### Task 5: Full Verification And Documentation Alignment
+
+**Files:**
+- Modify: `README_zh.md`
+- Modify: `tasks/todo.md`
+
+- [ ] Update documentation for new webhook config, case logs, and management endpoints if implementation changed the documented surface area.
+- [ ] Run targeted verification:
+  - `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_webhooks.py -v`
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest tests/test_manage_api.py -v`
+  - `node --test web/test/zone-state.test.js`
+Expected: PASS.
+- [ ] Run full verification:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest discover -s tests -v`
+  - `cd web && pnpm build`
+Expected: PASS.
+- [ ] Record final verification outcomes and any environmental caveats in `tasks/todo.md`.
diff --git a/docs/superpowers/specs/2026-06-09-webhook-case-management-design.md b/docs/superpowers/specs/2026-06-09-webhook-case-management-design.md
new file mode 100644
index 0000000..aef1458
--- /dev/null
+++ b/docs/superpowers/specs/2026-06-09-webhook-case-management-design.md
@@ -0,0 +1,216 @@
+# Webhook Case Management Design
+
+**Goal:** Add outbound webhooks plus a local case-management layer so the project can both push runtime facts to external systems and independently track pending/handled cases in the local management console.
+
+**Architecture:** Keep the existing runtime event stream as the source of operational facts. Add a separate case-state layer that consumes selected runtime events, persists case state transitions, exposes management APIs, and emits case webhooks without mutating the underlying batch facts. Integrate manual handling and external callback handling through the same case-state model.
+
+**Tech Stack:** Python 3.11+ standard library backend, JSONL persistence, Vite + vanilla JavaScript frontend, existing unittest and Node test suites.
+
+---
+
+## Scope
+
+This design extends the current project in four focused areas:
+
+1. Add outbound webhook delivery for runtime batch events.
+2. Add a local case model for operator workflow.
+3. Add management APIs for listing, summarizing, manually handling, and externally updating cases.
+4. Add frontend views and actions for local case operations.
+
+The runtime batch engine remains the producer of factual detection events. Case handling is a downstream interpretation layer.
+
+## Current Constraints
+
+- The current runtime writes facts to `logs/events.jsonl` and diagnostics to `logs/runtime_diagnostics.jsonl`.
+- The management API is a small standard-library HTTP server and should stay that way.
+- The frontend already renders runtime metrics and runtime events and should continue to do so.
+- The user-selected workflow requires both manual handling and external callback handling.
+- The user-selected workflow requires both event webhooks and case webhooks.
+- The events that should enter the local pending-case flow are `time_alarm`, `batch_pending_disposal`, and `warning_escalated`.
+
+## Design Summary
+
+The system is split into three cooperating layers:
+
+1. **Batch event layer**
+   Produces facts such as `batch_started`, `time_alarm`, `batch_pending_disposal`, `batch_discarded`, and `warning_escalated`. These remain append-only runtime facts.
+
+2. **Case state layer**
+   Consumes selected batch events and maintains a separate per-batch local case state. The case layer owns pending/handled workflow and does not rewrite prior runtime facts.
+
+3. **Integration layer**
+   Delivers outbound event and case webhooks, accepts external case callbacks, and records webhook delivery attempts for audit and debugging.
+
+## Persistence Model
+
+- `logs/events.jsonl`
+  Existing runtime fact log. No schema removals.
+- `logs/cases.jsonl`
+  New append-only case transition log. Each line records a case snapshot after a state change.
+- `logs/webhook_delivery.jsonl`
+  New append-only webhook delivery audit log. Each line records an attempted outbound delivery result.
+
+`events.jsonl` remains the source of factual batch history. `cases.jsonl` is the source of case workflow state. `webhook_delivery.jsonl` is operational telemetry only.
+
+## Case Model
+
+Each batch can own at most one local case. A case is created or updated from selected batch events and then independently handled by a local operator or external callback.
+
+### Case fields
+
+- `case_id`
+- `batch_id`
+- `camera_id`
+- `zone_id`
+- `zone_label`
+- `case_type`
+- `case_status`
+- `source_event`
+- `created_at`
+- `updated_at`
+- `handled_at`
+- `handled_by`
+- `handled_source`
+- `last_event_ts`
+- `payload`
+
+### Case type values
+
+- `time_alarm`
+- `pending_disposal`
+- `warning_escalated`
+
+### Case status values
+
+- `open`
+- `handled`
+
+### Handled source values
+
+- `manual`
+- `webhook_callback`
+- `auto_closed`
+
+## Case State Flow
+
+1. `time_alarm`
+   Create a case if one does not exist for the batch. If a case already exists, keep it open and refresh timestamps.
+
+2. `batch_pending_disposal`
+   Create a case if one does not exist. If one exists, update it in place and upgrade `case_type` to `pending_disposal`.
+
+3. `warning_escalated`
+   Update the same case in place and upgrade `case_type` to `warning_escalated`.
+
+4. Manual handling
+   Mark the case as `handled`, set `handled_source=manual`, record `handled_by`, and append the new snapshot to `cases.jsonl`.
+
+5. External callback handling
+   Mark the case as `handled`, set `handled_source=webhook_callback`, optionally record `handled_by` and `source_ref`, and append the new snapshot to `cases.jsonl`.
+
+6. `batch_discarded`
+   If the related case is still `open`, close it automatically with `handled_source=auto_closed`.
+
+Handled cases must not reopen when stale older events are replayed or re-read. Only new event processing in forward time may mutate an existing case. Restore logic must preserve handled status across runtime/API restarts.
+
+## Backend Components
+
+- Create `src/cold_display_guard/cases.py` for case transition logic, persistence, restore, and summary helpers.
+- Create `src/cold_display_guard/webhooks.py` for webhook config parsing, payload building, synchronous delivery, and delivery audit logging.
+- Extend `src/cold_display_guard/config.py` for webhook configuration and case/log sink paths.
+- Extend `src/cold_display_guard/main.py` to feed runtime events into case persistence and webhook delivery.
+- Extend `src/cold_display_guard/manage_api.py` to expose case listing, case summary, manual handling, and token-protected callback handling.
+
+## API Design
+
+All new endpoints stay under `/api/manage/*`.
+
+- `GET /api/manage/cases`
+  Query: `status=open|handled` optional, `limit` optional.
+- `GET /api/manage/cases/summary`
+  Returns case counts and latest update time.
+- `POST /api/manage/cases/{case_id}/handle`
+  Body: `handled_by` required, `note` optional.
+- `POST /api/manage/webhooks/case-update`
+  Body: `case_id` required, `status` required and must equal `handled`, `handled_by` optional, `source_ref` optional.
+
+The callback endpoint must require the configured shared token in the `X-Webhook-Token` header and must reject unauthenticated updates.
+
+## Webhook Configuration
+
+```toml
+[webhooks]
+enabled = true
+event_url = "https://example.com/runtime-events"
+case_url = "https://example.com/case-events"
+callback_token = "shared-secret"
+connect_timeout_seconds = 3
+read_timeout_seconds = 5
+```
+
+## Outbound Webhook Delivery
+
+Event webhook payload core fields:
+
+- `kind = "batch_event"`
+- `event`
+- `ts`
+- `batch_id`
+- `camera_id`
+- `zone_id`
+- `zone_label`
+- `severity`
+- `state`
+
+Case webhook payload core fields:
+
+- `kind = "case_event"`
+- `action = "created" | "updated" | "handled"`
+- `case_id`
+- `case_type`
+- `case_status`
+- `batch_id`
+- `source_event`
+- `handled_source`
+- `updated_at`
+
+Delivery rules:
+
+- Local runtime facts and case state must be persisted before webhook failure can affect control flow.
+- Webhook failure must append a line to `logs/webhook_delivery.jsonl`.
+- Webhook failure must not stop local event persistence or local case persistence.
+- This batch does not add a retry queue.
+
+## Frontend Changes
+
+- Keep the current runtime event table for factual runtime events only.
+- Add a separate case table with:
+  - `case_id`
+  - `case_type`
+  - `case_status`
+  - `zone_label`
+  - `batch_id`
+  - `created_at`
+  - `updated_at`
+  - `handled_source`
+- Add manual-handle UI for `open` cases with `handled_by` required and `note` optional.
+- Add summary cards for:
+  - `open_case_count`
+  - `handled_case_count`
+  - `time_alarm_case_count`
+  - `pending_disposal_case_count`
+  - `warning_escalated_case_count`
+
+## Testing Plan
+
+- Preserve existing batch engine behavior tests.
+- Add case tests for create, escalate, manual handle, callback handle, auto-close, and non-reopen behavior.
+- Add webhook tests for payloads, delivery success, and failure audit logging.
+- Add API tests for new case and callback endpoints.
+- Add frontend tests for case rendering, case summary mapping, and manual-handle request flow.
+
+Verification commands:
+
+- `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest discover -s tests -v`
+- `node --test web/test/zone-state.test.js`
+- `cd web && pnpm build`
diff --git a/src/cold_display_guard/cases.py b/src/cold_display_guard/cases.py
new file mode 100644
index 0000000..34aef7f
--- /dev/null
+++ b/src/cold_display_guard/cases.py
@@ -0,0 +1,238 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+
+EVENT_CASE_TYPES = {
+    "time_alarm": "time_alarm",
+    "batch_pending_disposal": "pending_disposal",
+    "warning_escalated": "warning_escalated",
+}
+
+CASE_PRIORITY = {
+    "time_alarm": 1,
+    "pending_disposal": 2,
+    "warning_escalated": 3,
+}
+
+
+@dataclass(slots=True)
+class CaseSnapshot:
+    case_id: str
+    batch_id: str
+    camera_id: str
+    zone_id: str
+    zone_label: str
+    case_type: str
+    case_status: str
+    source_event: str
+    created_at: datetime
+    updated_at: datetime
+    handled_at: datetime | None = None
+    handled_by: str = ""
+    handled_source: str = ""
+    last_event_ts: datetime | None = None
+    payload: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        payload = {
+            "case_id": self.case_id,
+            "batch_id": self.batch_id,
+            "camera_id": self.camera_id,
+            "zone_id": self.zone_id,
+            "zone_label": self.zone_label,
+            "case_type": self.case_type,
+            "case_status": self.case_status,
+            "source_event": self.source_event,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+            "handled_by": self.handled_by,
+            "handled_source": self.handled_source,
+            "payload": self.payload,
+        }
+        if self.handled_at is not None:
+            payload["handled_at"] = self.handled_at.isoformat()
+        if self.last_event_ts is not None:
+            payload["last_event_ts"] = self.last_event_ts.isoformat()
+        return payload
+
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> "CaseSnapshot":
+        return cls(
+            case_id=str(payload.get("case_id", "")),
+            batch_id=str(payload.get("batch_id", "")),
+            camera_id=str(payload.get("camera_id", "")),
+            zone_id=str(payload.get("zone_id", "")),
+            zone_label=str(payload.get("zone_label", "")),
+            case_type=str(payload.get("case_type", "")),
+            case_status=str(payload.get("case_status", "")),
+            source_event=str(payload.get("source_event", "")),
+            created_at=parse_datetime(payload.get("created_at")) or datetime.min,
+            updated_at=parse_datetime(payload.get("updated_at")) or datetime.min,
+            handled_at=parse_datetime(payload.get("handled_at")),
+            handled_by=str(payload.get("handled_by", "")),
+            handled_source=str(payload.get("handled_source", "")),
+            last_event_ts=parse_datetime(payload.get("last_event_ts")),
+            payload=dict(payload.get("payload", {}) or {}),
+        )
+
+
+class CaseStore:
+    def __init__(self, snapshots: list[dict[str, Any]] | None = None) -> None:
+        self._cases: dict[str, CaseSnapshot] = {}
+        for payload in snapshots or []:
+            snapshot = CaseSnapshot.from_dict(payload)
+            if not snapshot.case_id:
+                continue
+            existing = self._cases.get(snapshot.case_id)
+            if existing is None or snapshot.updated_at >= existing.updated_at:
+                self._cases[snapshot.case_id] = snapshot
+
+    def latest_cases(self) -> list[dict[str, Any]]:
+        snapshots = sorted(self._cases.values(), key=lambda item: item.updated_at, reverse=True)
+        return [snapshot.to_dict() for snapshot in snapshots]
+
+    def apply_batch_events(self, events: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        snapshots: list[dict[str, Any]] = []
+        for event in events:
+            snapshot = self._apply_batch_event(event)
+            if snapshot is not None:
+                snapshots.append(snapshot.to_dict())
+        return snapshots
+
+    def mark_handled(
+        self,
+        case_id: str,
+        *,
+        handled_at: datetime,
+        handled_by: str = "",
+        handled_source: str,
+        note: str = "",
+        source_ref: str = "",
+    ) -> dict[str, Any]:
+        snapshot = self._cases[case_id]
+        snapshot.case_status = "handled"
+        snapshot.updated_at = handled_at
+        snapshot.handled_at = handled_at
+        snapshot.handled_by = handled_by
+        snapshot.handled_source = handled_source
+        payload = dict(snapshot.payload)
+        if note:
+            payload["note"] = note
+        if source_ref:
+            payload["source_ref"] = source_ref
+        snapshot.payload = payload
+        return snapshot.to_dict()
+
+    def _apply_batch_event(self, event: dict[str, Any]) -> CaseSnapshot | None:
+        event_name = str(event.get("event", ""))
+        when = parse_datetime(event.get("ts"))
+        if when is None:
+            return None
+        batch_id = str(event.get("batch_id", "")).strip()
+        if not batch_id:
+            return None
+
+        case_id = build_case_id(batch_id)
+        existing = self._cases.get(case_id)
+        if event_name == "batch_discarded":
+            if existing is None or existing.case_status == "handled":
+                return None
+            return self._close_case(existing, when, handled_source="auto_closed")
+
+        case_type = EVENT_CASE_TYPES.get(event_name)
+        if case_type is None:
+            return None
+        if existing is not None:
+            if existing.last_event_ts is not None and when <= existing.last_event_ts:
+                return None
+            if existing.case_status == "handled":
+                return None
+            existing.case_type = higher_priority_case_type(existing.case_type, case_type)
+            existing.case_status = "open"
+            existing.source_event = event_name
+            existing.updated_at = when
+            existing.last_event_ts = when
+            existing.payload = {"event": dict(event)}
+            return existing
+
+        snapshot = CaseSnapshot(
+            case_id=case_id,
+            batch_id=batch_id,
+            camera_id=str(event.get("camera_id", "")),
+            zone_id=str(event.get("zone_id", "")),
+            zone_label=str(event.get("zone_label", "")),
+            case_type=case_type,
+            case_status="open",
+            source_event=event_name,
+            created_at=when,
+            updated_at=when,
+            last_event_ts=when,
+            payload={"event": dict(event)},
+        )
+        self._cases[case_id] = snapshot
+        return snapshot
+
+    def _close_case(self, snapshot: CaseSnapshot, handled_at: datetime, *, handled_source: str) -> CaseSnapshot:
+        snapshot.case_status = "handled"
+        snapshot.updated_at = handled_at
+        snapshot.handled_at = handled_at
+        snapshot.handled_source = handled_source
+        return snapshot
+
+
+def build_case_id(batch_id: str) -> str:
+    return f"case_{batch_id}"
+
+
+def higher_priority_case_type(current: str, incoming: str) -> str:
+    if CASE_PRIORITY.get(incoming, 0) >= CASE_PRIORITY.get(current, 0):
+        return incoming
+    return current
+
+
+def append_case_snapshots(path: Path, payloads: list[dict[str, Any]]) -> None:
+    if not payloads:
+        return
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        if path.exists() and path.stat().st_size > 0 and not file_ends_with_newline(path):
+            handle.write("\n")
+        for payload in payloads:
+            handle.write(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+            handle.write("\n")
+
+
+def load_case_snapshots(path: Path) -> list[dict[str, Any]]:
+    if not path.exists():
+        return []
+    items: list[dict[str, Any]] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        try:
+            payload = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(payload, dict):
+            items.append(payload)
+    return items
+
+
+def parse_datetime(value: Any) -> datetime | None:
+    if isinstance(value, datetime):
+        return value
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(str(value))
+    except ValueError:
+        return None
+
+
+def file_ends_with_newline(path: Path) -> bool:
+    with path.open("rb") as handle:
+        handle.seek(-1, 2)
+        return handle.read(1) == b"\n"
diff --git a/src/cold_display_guard/config.py b/src/cold_display_guard/config.py
index 47af412..4b69931 100644
--- a/src/cold_display_guard/config.py
+++ b/src/cold_display_guard/config.py
@@ -192,6 +192,34 @@ def format_config_document(data: dict[str, Any]) -> str:
     lines.append("[event_sink]")
     lines.append(f'path = "{_escape(str(event_sink.get("path", "logs/events.jsonl")))}"')
     lines.append("")
+
+    case_sink = data.get("case_sink", {})
+    if case_sink:
+        lines.append("[case_sink]")
+        lines.append(f'path = "{_escape(str(case_sink.get("path", "logs/cases.jsonl")))}"')
+        lines.append("")
+
+    webhooks = data.get("webhooks", {})
+    if webhooks:
+        lines.append("[webhooks]")
+        for key in (
+            "callback_token",
+            "case_url",
+            "connect_timeout_seconds",
+            "enabled",
+            "event_url",
+            "read_timeout_seconds",
+        ):
+            if key not in webhooks:
+                continue
+            value = webhooks[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("")
     return "\n".join(lines)
 
 
diff --git a/src/cold_display_guard/main.py b/src/cold_display_guard/main.py
index 1a5af49..7da781a 100644
--- a/src/cold_display_guard/main.py
+++ b/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.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
 from cold_display_guard.frame_source import FrameCaptureError, RTSPFrameSource
@@ -18,6 +19,7 @@ from cold_display_guard.vision import (
     load_runtime_vision_settings,
     metrics_indicate_occupied,
 )
+from cold_display_guard.webhooks import send_batch_event_webhooks, send_case_webhooks
 
 
 def main() -> int:
@@ -51,6 +53,11 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
 
     timezone = ZoneInfo(str(config.get("timezone", "Asia/Shanghai")))
     event_path = resolve_project_path(project_root, str(config.get("event_sink", {}).get("path", "logs/events.jsonl")))
+    case_path = case_sink_path(project_root, config)
+    webhook_delivery_path = resolve_project_path(
+        project_root,
+        str(config.get("webhook_delivery_sink", {}).get("path", "logs/webhook_delivery.jsonl")),
+    )
     diagnostics_path = resolve_project_path(project_root, str(runtime.get("diagnostics_path", "logs/runtime_diagnostics.jsonl")))
     sample_interval_seconds = max(0.1, float(runtime.get("sample_interval_seconds", 5.0)))
     frame_width = max(64, int(runtime.get("frame_width", 640)))
@@ -66,6 +73,7 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
     vision_settings = load_runtime_vision_settings(config)
     detector = ZoneOccupancyDetector(regions, trash_region, vision_settings)
     engine = BatchEngine(settings)
+    case_store = load_case_store(case_path)
     baseline_seed, active_zone_counts = restore_runtime_state(diagnostics_path, config)
     if baseline_seed:
         detector.seed_baseline(baseline_seed)
@@ -74,10 +82,13 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
         engine.restore_from_events(load_jsonl_tail(event_path, 2000), active_zone_counts=active_zone_counts)
 
     event_path.parent.mkdir(parents=True, exist_ok=True)
+    case_path.parent.mkdir(parents=True, exist_ok=True)
+    webhook_delivery_path.parent.mkdir(parents=True, exist_ok=True)
     diagnostics_path.parent.mkdir(parents=True, exist_ok=True)
     print(f"Cold Display Guard runtime started")
     print(f"Config: {resolved_config}")
     print(f"Events: {event_path}")
+    print(f"Cases: {case_path}")
     print(f"Diagnostics: {diagnostics_path}")
 
     iteration = 0
@@ -90,6 +101,8 @@ def run(config_path: str | Path, once: bool = False, max_iterations: int = 0) ->
             observation = Observation(ts=when, zone_counts=zone_counts, trash_deposit_count=trash_deposit_count)
             events = engine.process(observation)
             append_jsonl(event_path, events)
+            case_snapshots = persist_case_updates(case_store, case_path, events)
+            deliver_runtime_webhooks(events, case_snapshots, config, webhook_delivery_path)
             append_jsonl(
                 diagnostics_path,
                 [
@@ -122,6 +135,11 @@ def resolve_project_path(project_root: Path, raw_path: str) -> Path:
     return path.resolve()
 
 
+def case_sink_path(project_root: Path, config: dict) -> Path:
+    raw_path = str(config.get("case_sink", {}).get("path", "logs/cases.jsonl"))
+    return resolve_project_path(project_root, raw_path)
+
+
 def append_jsonl(path: Path, payloads: list[dict]) -> None:
     if not payloads:
         return
@@ -131,6 +149,28 @@ def append_jsonl(path: Path, payloads: list[dict]) -> None:
             handle.write("\n")
 
 
+def load_case_store(path: Path) -> CaseStore:
+    return CaseStore(load_case_snapshots(path))
+
+
+def persist_case_updates(case_store: CaseStore, path: Path, events: list[dict[str, object]]) -> list[dict[str, object]]:
+    snapshots = case_store.apply_batch_events(events)
+    append_case_snapshots(path, snapshots)
+    return snapshots
+
+
+def deliver_runtime_webhooks(
+    events: list[dict[str, object]],
+    case_snapshots: list[dict[str, object]],
+    config: dict[str, object],
+    audit_path: Path,
+    *,
+    http_post=None,
+) -> None:
+    send_batch_event_webhooks(events, config, audit_path, http_post=http_post)
+    send_case_webhooks(case_snapshots, config, audit_path, http_post=http_post)
+
+
 def restore_runtime_state(diagnostics_path: Path, config: dict) -> tuple[dict[str, RegionMetrics], dict[str, int]]:
     latest = load_jsonl_tail(diagnostics_path, 1)
     if not latest:
diff --git a/src/cold_display_guard/manage_api.py b/src/cold_display_guard/manage_api.py
index 3b266a2..f5e818b 100644
--- a/src/cold_display_guard/manage_api.py
+++ b/src/cold_display_guard/manage_api.py
@@ -11,6 +11,7 @@ from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, urlparse
 
+from cold_display_guard.cases import CaseStore, append_case_snapshots, load_case_snapshots
 from cold_display_guard.config import (
     load_config_document,
     merge_calibration,
@@ -19,6 +20,7 @@ from cold_display_guard.config import (
     save_config_document,
 )
 from cold_display_guard.vision import load_runtime_vision_settings, metrics_indicate_occupied
+from cold_display_guard.webhooks import send_case_webhooks
 
 
 PROJECT_TYPE = "cold_display_guard"
@@ -66,6 +68,15 @@ def create_handler(ctx: ManageContext) -> type[BaseHTTPRequestHandler]:
                 limit = bounded_int(query.get("limit", ["200"])[0], 1, MAX_EVENT_LINES)
                 self._send_json({"items": load_events(ctx, limit), "limit": limit})
                 return
+            if parsed.path == "/api/manage/cases":
+                query = parse_qs(parsed.query)
+                limit = bounded_int(query.get("limit", ["200"])[0], 1, MAX_EVENT_LINES)
+                status = str(query.get("status", [""])[0]).strip().lower()
+                self._send_json({"items": load_cases(ctx, limit=limit, status=status), "limit": limit})
+                return
+            if parsed.path == "/api/manage/cases/summary":
+                self._send_json(build_case_summary(ctx))
+                return
             if parsed.path == "/api/manage/diagnostics":
                 query = parse_qs(parsed.query)
                 limit = bounded_int(query.get("limit", ["50"])[0], 1, MAX_EVENT_LINES)
@@ -88,6 +99,13 @@ def create_handler(ctx: ManageContext) -> type[BaseHTTPRequestHandler]:
             if parsed.path == "/api/manage/snapshot":
                 self._capture_snapshot()
                 return
+            if parsed.path.startswith("/api/manage/cases/") and parsed.path.endswith("/handle"):
+                case_id = parsed.path.removeprefix("/api/manage/cases/").removesuffix("/handle").strip("/")
+                self._handle_case(case_id)
+                return
+            if parsed.path == "/api/manage/webhooks/case-update":
+                self._handle_case_callback()
+                return
             self.send_error(HTTPStatus.NOT_FOUND)
 
         def log_message(self, format: str, *args: object) -> None:
@@ -154,6 +172,54 @@ def create_handler(ctx: ManageContext) -> type[BaseHTTPRequestHandler]:
             self.end_headers()
             self.wfile.write(image)
 
+        def _handle_case(self, case_id: str) -> None:
+            payload = self._read_json()
+            handled_by = str(payload.get("handled_by", "")).strip()
+            if not case_id:
+                self._send_json({"error": "case_id is required"}, HTTPStatus.BAD_REQUEST)
+                return
+            if not handled_by:
+                self._send_json({"error": "handled_by is required"}, HTTPStatus.BAD_REQUEST)
+                return
+            snapshot = handle_case_update(
+                ctx,
+                case_id,
+                handled_by=handled_by,
+                handled_source="manual",
+                note=str(payload.get("note", "")).strip(),
+            )
+            if snapshot is None:
+                self._send_json({"error": "case not found"}, HTTPStatus.NOT_FOUND)
+                return
+            self._send_json(snapshot)
+
+        def _handle_case_callback(self) -> None:
+            payload = self._read_json()
+            config = load_config_document(ctx.config_path)
+            token = str(config.get("webhooks", {}).get("callback_token", ""))
+            if not token or self.headers.get("X-Webhook-Token") != token:
+                self._send_json({"error": "forbidden"}, HTTPStatus.FORBIDDEN)
+                return
+            case_id = str(payload.get("case_id", "")).strip()
+            status = str(payload.get("status", "")).strip().lower()
+            if not case_id:
+                self._send_json({"error": "case_id is required"}, HTTPStatus.BAD_REQUEST)
+                return
+            if status != "handled":
+                self._send_json({"error": "status must be handled"}, HTTPStatus.BAD_REQUEST)
+                return
+            snapshot = handle_case_update(
+                ctx,
+                case_id,
+                handled_by=str(payload.get("handled_by", "")).strip(),
+                handled_source="webhook_callback",
+                source_ref=str(payload.get("source_ref", "")).strip(),
+            )
+            if snapshot is None:
+                self._send_json({"error": "case not found"}, HTTPStatus.NOT_FOUND)
+                return
+            self._send_json(snapshot)
+
         def _read_json(self) -> dict[str, Any]:
             length = int(self.headers.get("Content-Length", "0"))
             if length == 0:
@@ -229,6 +295,9 @@ def main() -> int:
 def config_payload(ctx: ManageContext) -> dict[str, Any]:
     data = load_config_document(ctx.config_path)
     event_path = event_sink_path(ctx, data)
+    case_path = case_sink_path(ctx, data)
+    webhooks = dict(data.get("webhooks", {}) or {})
+    webhooks.pop("callback_token", None)
     return {
         "project_type": PROJECT_TYPE,
         "config_path": str(ctx.config_path),
@@ -243,6 +312,8 @@ def config_payload(ctx: ManageContext) -> dict[str, Any]:
         "zones": data.get("zones", []),
         "trash": data.get("trash", {}),
         "event_sink": {"path": str(event_path)},
+        "case_sink": {"path": str(case_path)},
+        "webhooks": webhooks,
     }
 
 
@@ -307,6 +378,40 @@ def load_events(ctx: ManageContext, limit: int) -> list[dict[str, Any]]:
     return load_jsonl_tail(path, limit)
 
 
+def load_cases(ctx: ManageContext, limit: int, status: str = "") -> list[dict[str, Any]]:
+    store = CaseStore(load_case_snapshots(case_sink_path(ctx)))
+    cases = store.latest_cases()
+    if status:
+        cases = [item for item in cases if str(item.get("case_status", "")).lower() == status]
+    return cases[:limit]
+
+
+def build_case_summary(ctx: ManageContext) -> dict[str, Any]:
+    cases = load_cases(ctx, limit=MAX_EVENT_LINES)
+    summary = {
+        "open_case_count": 0,
+        "handled_case_count": 0,
+        "time_alarm_case_count": 0,
+        "pending_disposal_case_count": 0,
+        "warning_escalated_case_count": 0,
+        "latest_case_update_time": "",
+    }
+    for case in cases:
+        status = str(case.get("case_status", ""))
+        case_type = str(case.get("case_type", ""))
+        if status == "open":
+            summary["open_case_count"] += 1
+        elif status == "handled":
+            summary["handled_case_count"] += 1
+        key = f"{case_type}_case_count"
+        if key in summary:
+            summary[key] += 1
+        updated_at = str(case.get("updated_at", ""))
+        if updated_at and updated_at > str(summary["latest_case_update_time"]):
+            summary["latest_case_update_time"] = updated_at
+    return summary
+
+
 def load_diagnostics(ctx: ManageContext, limit: int) -> list[dict[str, Any]]:
     path = diagnostics_path(ctx)
     return load_jsonl_tail(path, limit)
@@ -335,6 +440,26 @@ def event_sink_path(ctx: ManageContext, data: dict[str, Any] | None = None) -> P
     return path.resolve()
 
 
+def case_sink_path(ctx: ManageContext, data: dict[str, Any] | None = None) -> Path:
+    if data is None:
+        data = load_config_document(ctx.config_path)
+    raw_path = str(data.get("case_sink", {}).get("path", "logs/cases.jsonl"))
+    path = Path(raw_path).expanduser()
+    if not path.is_absolute():
+        path = ctx.project_root / path
+    return path.resolve()
+
+
+def webhook_delivery_path(ctx: ManageContext, data: dict[str, Any] | None = None) -> Path:
+    if data is None:
+        data = load_config_document(ctx.config_path)
+    raw_path = str(data.get("webhook_delivery_sink", {}).get("path", "logs/webhook_delivery.jsonl"))
+    path = Path(raw_path).expanduser()
+    if not path.is_absolute():
+        path = ctx.project_root / path
+    return path.resolve()
+
+
 def diagnostics_path(ctx: ManageContext, data: dict[str, Any] | None = None) -> Path:
     if data is None:
         data = load_config_document(ctx.config_path)
@@ -345,6 +470,34 @@ def diagnostics_path(ctx: ManageContext, data: dict[str, Any] | None = None) ->
     return path.resolve()
 
 
+def handle_case_update(
+    ctx: ManageContext,
+    case_id: str,
+    *,
+    handled_by: str,
+    handled_source: str,
+    note: str = "",
+    source_ref: str = "",
+) -> dict[str, Any] | None:
+    config = load_config_document(ctx.config_path)
+    path = case_sink_path(ctx, config)
+    store = CaseStore(load_case_snapshots(path))
+    matching = {item["case_id"] for item in store.latest_cases()}
+    if case_id not in matching:
+        return None
+    snapshot = store.mark_handled(
+        case_id,
+        handled_at=datetime.now(timezone.utc),
+        handled_by=handled_by,
+        handled_source=handled_source,
+        note=note,
+        source_ref=source_ref,
+    )
+    append_case_snapshots(path, [snapshot])
+    send_case_webhooks([snapshot], config, webhook_delivery_path(ctx, config))
+    return snapshot
+
+
 def latest_zone_counts(diagnostics: list[dict[str, Any]], config: dict[str, Any] | None = None) -> dict[str, int]:
     for item in reversed(diagnostics):
         stable_counts = stable_zone_counts_from_diagnostics(item)
diff --git a/src/cold_display_guard/webhooks.py b/src/cold_display_guard/webhooks.py
new file mode 100644
index 0000000..0dcbc00
--- /dev/null
+++ b/src/cold_display_guard/webhooks.py
@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+from urllib import request
+
+
+@dataclass(frozen=True, slots=True)
+class WebhookSettings:
+    enabled: bool = False
+    event_url: str = ""
+    case_url: str = ""
+    callback_token: str = ""
+    connect_timeout_seconds: float = 3.0
+    read_timeout_seconds: float = 5.0
+
+
+HttpPost = Callable[[str, dict[str, object], tuple[float, float]], tuple[int, str]]
+
+
+def load_webhook_settings(config: dict[str, Any]) -> WebhookSettings:
+    payload = config.get("webhooks", {})
+    if not isinstance(payload, dict):
+        payload = {}
+    return WebhookSettings(
+        enabled=bool(payload.get("enabled", False)),
+        event_url=str(payload.get("event_url", "")),
+        case_url=str(payload.get("case_url", "")),
+        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)),
+    )
+
+
+def build_batch_event_payload(event: dict[str, object]) -> dict[str, object]:
+    return {
+        "kind": "batch_event",
+        "event": event.get("event", ""),
+        "ts": event.get("ts", ""),
+        "batch_id": event.get("batch_id", ""),
+        "camera_id": event.get("camera_id", ""),
+        "zone_id": event.get("zone_id", ""),
+        "zone_label": event.get("zone_label", ""),
+        "severity": event.get("severity", ""),
+        "state": event.get("state", ""),
+    }
+
+
+def build_case_event_payload(snapshot: dict[str, object]) -> dict[str, object]:
+    return {
+        "kind": "case_event",
+        "action": infer_case_action(snapshot),
+        "case_id": snapshot.get("case_id", ""),
+        "case_type": snapshot.get("case_type", ""),
+        "case_status": snapshot.get("case_status", ""),
+        "batch_id": snapshot.get("batch_id", ""),
+        "source_event": snapshot.get("source_event", ""),
+        "handled_source": snapshot.get("handled_source", ""),
+        "updated_at": snapshot.get("updated_at", ""),
+    }
+
+
+def infer_case_action(snapshot: dict[str, object]) -> str:
+    if str(snapshot.get("case_status", "")) == "handled":
+        return "handled"
+    created_at = str(snapshot.get("created_at", ""))
+    updated_at = str(snapshot.get("updated_at", ""))
+    return "created" if created_at and created_at == updated_at else "updated"
+
+
+def send_batch_event_webhooks(
+    events: list[dict[str, object]],
+    config: dict[str, Any],
+    audit_path: Path,
+    *,
+    http_post: HttpPost | None = None,
+) -> list[dict[str, object]]:
+    settings = load_webhook_settings(config)
+    if not settings.enabled or not settings.event_url:
+        return []
+    deliveries: list[dict[str, object]] = []
+    for event in events:
+        payload = build_batch_event_payload(event)
+        deliveries.append(
+            deliver_webhook(
+                settings.event_url,
+                payload,
+                audit_path,
+                target="batch_event",
+                settings=settings,
+                http_post=http_post,
+            )
+        )
+    return deliveries
+
+
+def send_case_webhooks(
+    snapshots: list[dict[str, object]],
+    config: dict[str, Any],
+    audit_path: Path,
+    *,
+    http_post: HttpPost | None = None,
+) -> list[dict[str, object]]:
+    settings = load_webhook_settings(config)
+    if not settings.enabled or not settings.case_url:
+        return []
+    deliveries: list[dict[str, object]] = []
+    for snapshot in snapshots:
+        payload = build_case_event_payload(snapshot)
+        deliveries.append(
+            deliver_webhook(
+                settings.case_url,
+                payload,
+                audit_path,
+                target="case_event",
+                settings=settings,
+                http_post=http_post,
+            )
+        )
+    return deliveries
+
+
+def deliver_webhook(
+    url: str,
+    payload: dict[str, object],
+    audit_path: Path,
+    *,
+    target: str,
+    settings: WebhookSettings,
+    http_post: HttpPost | None = None,
+) -> dict[str, object]:
+    post = http_post or post_json
+    timeout = (settings.connect_timeout_seconds, settings.read_timeout_seconds)
+    try:
+        status_code, response_text = post(url, payload, timeout)
+        record = {
+            "ts": datetime.now(timezone.utc).isoformat(),
+            "target": target,
+            "url": url,
+            "status": "ok",
+            "status_code": status_code,
+            "message": response_text,
+        }
+    except OSError as exc:
+        record = {
+            "ts": datetime.now(timezone.utc).isoformat(),
+            "target": target,
+            "url": url,
+            "status": "error",
+            "message": str(exc),
+        }
+    append_delivery_record(audit_path, record)
+    return record
+
+
+def post_json(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:
+    data = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
+    req = request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
+    with request.urlopen(req, timeout=sum(timeout)) as response:
+        return response.getcode(), response.read().decode("utf-8")
+
+
+def append_delivery_record(path: Path, payload: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+        handle.write("\n")
diff --git a/tasks/todo.md b/tasks/todo.md
index c234842..be6026f 100644
--- a/tasks/todo.md
+++ b/tasks/todo.md
@@ -1,10 +1,51 @@
 # Task Todo
 
-- [x] Review the current `AGENTS.md` structure and check for existing `tasks/` tracking files.
-- [x] Check in on the plan before implementation and keep this change scoped to the requested documentation update.
-- [x] Merge the requested workflow orchestration, task management, and execution principles into `AGENTS.md`.
-- [x] Verify the updated `AGENTS.md` content against the requested rules and record the review result here.
+- [x] Review the current project instructions and check for task-relevant lessons.
+- [x] Check repository status before writing the implementation plan.
+- [x] Inspect existing engine, CLI, docs, and frontend event handling for disposal-tracking impact.
+- [x] Write the design spec for webhook case management in an isolated worktree.
+- [x] Confirm the design with the user before implementation.
 
-## Review
+## Design Review
 
-- Verified by re-reading `AGENTS.md` and checking the inserted sections for plan-first workflow, `tasks/todo.md` tracking, `tasks/lessons.md` review/update rules, subagent guidance, verification requirements, and simplicity/minimal-impact principles.
+- Spec path: `docs/superpowers/specs/2026-06-09-webhook-case-management-design.md`
+- Scope fixed to local case management plus outbound and inbound webhook integration.
+- Confirmed behaviors:
+  - manual handling and external callback handling are both supported
+  - cases are created from `time_alarm`, `batch_pending_disposal`, and `warning_escalated`
+  - both batch-event webhooks and case-state webhooks are required
+  - callback `status` is exactly `handled`
+  - callback-applied case handling must emit a `case_event` webhook
+
+## 2026-06-09 Implementation Plan
+
+- [x] Create isolated worktree for implementation on branch `feat/webhook-case-management`.
+- [x] Re-check runtime baseline in the worktree and note the local Python environment requirement.
+- [x] Write the detailed implementation plan to `docs/superpowers/plans/2026-06-09-webhook-case-management-implementation.md`.
+- [x] Execute backend case-state TDD cycle.
+- [x] Execute webhook integration TDD cycle.
+- [x] Execute management API TDD cycle.
+- [x] Execute frontend case-management TDD cycle.
+- [x] Run full verification and record outcomes.
+
+## 2026-06-09 Implementation Review
+
+- Worktree path: `/Users/glo/.config/superpowers/worktrees/cold_display_guard/webhook-case-management`
+- Baseline note: the default `python3` in this shell resolves to macOS system Python 3.9 and cannot import the repo's `dataclass(..., slots=True)` code. Python verification in this worktree must run through `eval "$(/opt/homebrew/bin/pyenv init -)" && python ...`, which resolves to Python 3.12.11.
+- Frontend baseline check in the worktree passed with `node --test web/test/zone-state.test.js`.
+- Implemented:
+  - `src/cold_display_guard/cases.py` for case lifecycle and JSONL persistence
+  - `src/cold_display_guard/webhooks.py` for outbound event/case webhook delivery and audit logging
+  - runtime integration in `src/cold_display_guard/main.py`
+  - case listing/summary/manual-handle/callback routes in `src/cold_display_guard/manage_api.py`
+  - frontend case summary and manual-handle flow in `web/src/main.js` and `web/src/zone-state.js`
+- Targeted verification passed during implementation:
+  - `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_webhooks.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`
+  - `node --test web/test/zone-state.test.js`
+- Final verification passed:
+  - `eval "$(/opt/homebrew/bin/pyenv init -)" && PYTHONPATH=src python -m unittest discover -s tests -v`
+  - `cd web && pnpm build`
+- Frontend build note: the isolated worktree needed `cd web && pnpm install --frozen-lockfile` before `pnpm build` because `node_modules` are not shared into new worktrees.
diff --git a/tests/test_cases.py b/tests/test_cases.py
new file mode 100644
index 0000000..94ac3b7
--- /dev/null
+++ b/tests/test_cases.py
@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from cold_display_guard.cases import CaseStore, append_case_snapshots, load_case_snapshots
+
+
+UTC = timezone.utc
+
+
+def event(
+    event_name: str,
+    when: datetime,
+    *,
+    batch_id: str = "batch_000001",
+    zone_id: str = "1",
+    zone_label: str = "区域 1",
+    camera_id: str = "cam_01",
+    severity: str = "info",
+    state: str = "active",
+) -> dict[str, object]:
+    return {
+        "event": event_name,
+        "ts": when.isoformat(),
+        "batch_id": batch_id,
+        "camera_id": camera_id,
+        "zone_id": zone_id,
+        "zone_label": zone_label,
+        "severity": severity,
+        "state": state,
+    }
+
+
+class CaseStoreTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.t0 = datetime(2026, 6, 9, 9, 0, tzinfo=UTC)
+
+    def test_time_alarm_creates_open_case(self) -> None:
+        store = CaseStore()
+
+        snapshots = store.apply_batch_events([event("time_alarm", self.t0, 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_pending_disposal_upgrades_existing_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])
+
+        snapshots = store.apply_batch_events(
+            [event("batch_pending_disposal", self.t0.replace(minute=1), severity="warning", state="pending_disposal")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_type"], "pending_disposal")
+        self.assertEqual(snapshots[0]["case_status"], "open")
+        self.assertEqual(snapshots[0]["source_event"], "batch_pending_disposal")
+
+    def test_warning_escalated_upgrades_same_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])
+        store.apply_batch_events(
+            [event("batch_pending_disposal", self.t0.replace(minute=1), severity="warning", state="pending_disposal")]
+        )
+
+        snapshots = store.apply_batch_events(
+            [event("warning_escalated", self.t0.replace(minute=2), severity="warning", state="warning")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_type"], "warning_escalated")
+        self.assertEqual(snapshots[0]["case_status"], "open")
+        self.assertEqual(snapshots[0]["source_event"], "warning_escalated")
+
+    def test_batch_discarded_auto_closes_open_case(self) -> None:
+        store = CaseStore()
+        store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])
+
+        snapshots = store.apply_batch_events(
+            [event("batch_discarded", self.t0.replace(minute=3), severity="info", state="discarded")]
+        )
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(snapshots[0]["case_status"], "handled")
+        self.assertEqual(snapshots[0]["handled_source"], "auto_closed")
+
+    def test_manual_handle_closes_case(self) -> None:
+        store = CaseStore()
+        created = store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])[0]
+
+        snapshot = store.mark_handled(
+            str(created["case_id"]),
+            handled_at=self.t0.replace(minute=4),
+            handled_by="alice",
+            handled_source="manual",
+            note="checked",
+        )
+
+        self.assertEqual(snapshot["case_status"], "handled")
+        self.assertEqual(snapshot["handled_source"], "manual")
+        self.assertEqual(snapshot["handled_by"], "alice")
+        self.assertEqual(snapshot["payload"]["note"], "checked")
+
+    def test_callback_handle_closes_case(self) -> None:
+        store = CaseStore()
+        created = store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])[0]
+
+        snapshot = store.mark_handled(
+            str(created["case_id"]),
+            handled_at=self.t0.replace(minute=5),
+            handled_by="crm-bot",
+            handled_source="webhook_callback",
+            source_ref="crm-123",
+        )
+
+        self.assertEqual(snapshot["case_status"], "handled")
+        self.assertEqual(snapshot["handled_source"], "webhook_callback")
+        self.assertEqual(snapshot["payload"]["source_ref"], "crm-123")
+
+    def test_handled_case_does_not_reopen_on_stale_event(self) -> None:
+        store = CaseStore()
+        created = store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])[0]
+        store.mark_handled(
+            str(created["case_id"]),
+            handled_at=self.t0.replace(minute=5),
+            handled_by="alice",
+            handled_source="manual",
+        )
+
+        snapshots = store.apply_batch_events(
+            [event("batch_pending_disposal", self.t0.replace(minute=1), severity="warning", state="pending_disposal")]
+        )
+
+        self.assertEqual(snapshots, [])
+        case = store.latest_cases()[0]
+        self.assertEqual(case["case_status"], "handled")
+        self.assertEqual(case["handled_source"], "manual")
+
+    def test_case_snapshots_round_trip_through_jsonl(self) -> None:
+        store = CaseStore()
+        snapshots = store.apply_batch_events([event("time_alarm", self.t0, severity="alarm", state="alerted")])
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "cases.jsonl"
+            append_case_snapshots(path, snapshots)
+            loaded = load_case_snapshots(path)
+
+        self.assertEqual(len(loaded), 1)
+        self.assertEqual(loaded[0]["case_type"], "time_alarm")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_config.py b/tests/test_config.py
index ed5f90b..472a75d 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -122,6 +122,32 @@ zone_ids = ["1", "2", "3"]
         self.assertIn("[trash]", text)
         self.assertNotIn('"trash"', text.split("[layout]", maxsplit=1)[1].split("[[zones]]", maxsplit=1)[0])
 
+    def test_save_config_document_writes_webhooks_and_case_sink(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "config.toml"
+            save_config_document(
+                path,
+                {
+                    "webhooks": {
+                        "enabled": True,
+                        "event_url": "https://example.com/events",
+                        "case_url": "https://example.com/cases",
+                        "callback_token": "secret",
+                        "connect_timeout_seconds": 3,
+                        "read_timeout_seconds": 5,
+                    },
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                },
+            )
+            text = path.read_text(encoding="utf-8")
+
+        self.assertIn("[webhooks]", text)
+        self.assertIn('event_url = "https://example.com/events"', text)
+        self.assertIn('case_url = "https://example.com/cases"', text)
+        self.assertIn('callback_token = "secret"', text)
+        self.assertIn("[case_sink]", text)
+        self.assertIn('path = "logs/cases.jsonl"', text)
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_main.py b/tests/test_main.py
index 22ec2f9..111bdb7 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -3,12 +3,102 @@ from __future__ import annotations
 import json
 import tempfile
 import unittest
+from datetime import datetime, timezone
 from pathlib import Path
 
-from cold_display_guard.main import restore_runtime_state
+from cold_display_guard.cases import CaseStore
+from cold_display_guard.main import case_sink_path, deliver_runtime_webhooks, persist_case_updates, restore_runtime_state
+
+
+UTC = timezone.utc
 
 
 class RuntimeRestoreTests(unittest.TestCase):
+    def test_case_sink_path_uses_default_logs_location(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+
+            path = case_sink_path(root, {})
+
+        self.assertEqual(path, (root / "logs" / "cases.jsonl").resolve())
+
+    def test_persist_case_updates_writes_case_snapshots(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "cases.jsonl"
+            store = CaseStore()
+
+            snapshots = persist_case_updates(
+                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",
+                    }
+                ],
+            )
+
+            written = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+
+        self.assertEqual(len(snapshots), 1)
+        self.assertEqual(written[0]["case_type"], "time_alarm")
+        self.assertEqual(written[0]["case_status"], "open")
+
+    def test_deliver_runtime_webhooks_sends_event_and_case_payloads(self) -> None:
+        deliveries: list[tuple[str, dict[str, object]]] = []
+
+        def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:
+            deliveries.append((url, payload))
+            return 200, "ok"
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            audit_path = Path(tmpdir) / "webhook_delivery.jsonl"
+            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",
+                    }
+                },
+                audit_path,
+                http_post=fake_post,
+            )
+
+        self.assertEqual(len(deliveries), 2)
+        self.assertEqual(deliveries[0][1]["kind"], "batch_event")
+        self.assertEqual(deliveries[1][1]["kind"], "case_event")
+
     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"
diff --git a/tests/test_manage_api.py b/tests/test_manage_api.py
index cc4cab6..1ee1d76 100644
--- a/tests/test_manage_api.py
+++ b/tests/test_manage_api.py
@@ -1,15 +1,47 @@
 from __future__ import annotations
 
+import http.client
 import json
 import tempfile
+import threading
 import unittest
+from http.server import ThreadingHTTPServer
 from pathlib import Path
 
 from cold_display_guard.config import load_config_document, merge_calibration, save_config_document
-from cold_display_guard.manage_api import ManageContext, build_summary
+from cold_display_guard.manage_api import ManageContext, build_summary, config_payload, create_handler
 
 
 class ManageApiTests(unittest.TestCase):
+    def _serve_once(self, ctx: ManageContext) -> tuple[ThreadingHTTPServer, threading.Thread]:
+        server = ThreadingHTTPServer(("127.0.0.1", 0), create_handler(ctx))
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        return server, thread
+
+    def _request(
+        self,
+        server: ThreadingHTTPServer,
+        method: str,
+        path: str,
+        body: dict | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> tuple[int, dict]:
+        conn = http.client.HTTPConnection("127.0.0.1", server.server_address[1], timeout=5)
+        payload = None if body is None else json.dumps(body)
+        final_headers = {"Content-Type": "application/json"}
+        final_headers.update(headers or {})
+        conn.request(method, path, body=payload, headers=final_headers)
+        response = conn.getresponse()
+        raw = response.read().decode("utf-8")
+        conn.close()
+        return response.status, json.loads(raw or "{}")
+
+    def _stop_server(self, server: ThreadingHTTPServer, thread: threading.Thread) -> None:
+        server.shutdown()
+        thread.join()
+        server.server_close()
+
     def test_merge_calibration_updates_zones_and_trash(self) -> None:
         data = {
             "camera_id": "cam",
@@ -350,6 +382,242 @@ class ManageApiTests(unittest.TestCase):
 
         self.assertEqual(summary["metrics"]["latest_zone_counts"], {"1": 1, "2": 0})
 
+    def test_config_payload_exposes_case_sink_and_webhooks_without_callback_token(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                    "webhooks": {
+                        "enabled": True,
+                        "event_url": "https://example.com/events",
+                        "case_url": "https://example.com/cases",
+                        "callback_token": "secret",
+                    },
+                },
+            )
+
+            payload = config_payload(ManageContext(config_path=config_path, project_root=root))
+
+        self.assertEqual(payload["case_sink"]["path"], str((root / "logs" / "cases.jsonl").resolve()))
+        self.assertTrue(payload["webhooks"]["enabled"])
+        self.assertNotIn("callback_token", payload["webhooks"])
+
+    def test_cases_endpoint_returns_latest_snapshots(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                    "layout": {"zone_ids": ["1"]},
+                },
+            )
+            cases_path = root / "logs" / "cases.jsonl"
+            cases_path.parent.mkdir()
+            cases_path.write_text(
+                json.dumps(
+                    {
+                        "case_id": "case_batch_000001",
+                        "batch_id": "batch_000001",
+                        "camera_id": "cam_01",
+                        "zone_id": "1",
+                        "zone_label": "区域 1",
+                        "case_type": "time_alarm",
+                        "case_status": "open",
+                        "source_event": "time_alarm",
+                        "created_at": "2026-06-09T09:00:00+08:00",
+                        "updated_at": "2026-06-09T09:00:00+08:00",
+                        "payload": {},
+                    }
+                ),
+                encoding="utf-8",
+            )
+            ctx = ManageContext(config_path=config_path, project_root=root)
+            server, thread = self._serve_once(ctx)
+            try:
+                status, payload = self._request(server, "GET", "/api/manage/cases?status=open")
+            finally:
+                self._stop_server(server, thread)
+
+        self.assertEqual(status, 200)
+        self.assertEqual(len(payload["items"]), 1)
+        self.assertEqual(payload["items"][0]["case_id"], "case_batch_000001")
+
+    def test_case_summary_endpoint_counts_open_and_handled(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                    "layout": {"zone_ids": ["1"]},
+                },
+            )
+            cases_path = root / "logs" / "cases.jsonl"
+            cases_path.parent.mkdir()
+            cases_path.write_text(
+                "\n".join(
+                    [
+                        json.dumps(
+                            {
+                                "case_id": "case_batch_000001",
+                                "batch_id": "batch_000001",
+                                "camera_id": "cam_01",
+                                "zone_id": "1",
+                                "zone_label": "区域 1",
+                                "case_type": "time_alarm",
+                                "case_status": "open",
+                                "source_event": "time_alarm",
+                                "created_at": "2026-06-09T09:00:00+08:00",
+                                "updated_at": "2026-06-09T09:00:00+08:00",
+                                "payload": {},
+                            }
+                        ),
+                        json.dumps(
+                            {
+                                "case_id": "case_batch_000002",
+                                "batch_id": "batch_000002",
+                                "camera_id": "cam_01",
+                                "zone_id": "1",
+                                "zone_label": "区域 1",
+                                "case_type": "warning_escalated",
+                                "case_status": "handled",
+                                "source_event": "warning_escalated",
+                                "created_at": "2026-06-09T09:01:00+08:00",
+                                "updated_at": "2026-06-09T09:05:00+08:00",
+                                "handled_source": "manual",
+                                "payload": {},
+                            }
+                        ),
+                    ]
+                ),
+                encoding="utf-8",
+            )
+            ctx = ManageContext(config_path=config_path, project_root=root)
+            server, thread = self._serve_once(ctx)
+            try:
+                status, payload = self._request(server, "GET", "/api/manage/cases/summary")
+            finally:
+                self._stop_server(server, thread)
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["open_case_count"], 1)
+        self.assertEqual(payload["handled_case_count"], 1)
+        self.assertEqual(payload["warning_escalated_case_count"], 1)
+
+    def test_manual_handle_endpoint_appends_handled_snapshot(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                    "layout": {"zone_ids": ["1"]},
+                },
+            )
+            cases_path = root / "logs" / "cases.jsonl"
+            cases_path.parent.mkdir()
+            cases_path.write_text(
+                json.dumps(
+                    {
+                        "case_id": "case_batch_000001",
+                        "batch_id": "batch_000001",
+                        "camera_id": "cam_01",
+                        "zone_id": "1",
+                        "zone_label": "区域 1",
+                        "case_type": "time_alarm",
+                        "case_status": "open",
+                        "source_event": "time_alarm",
+                        "created_at": "2026-06-09T09:00:00+08:00",
+                        "updated_at": "2026-06-09T09:00:00+08:00",
+                        "payload": {},
+                    }
+                ),
+                encoding="utf-8",
+            )
+            ctx = ManageContext(config_path=config_path, project_root=root)
+            server, thread = self._serve_once(ctx)
+            try:
+                status, payload = self._request(
+                    server,
+                    "POST",
+                    "/api/manage/cases/case_batch_000001/handle",
+                    body={"handled_by": "alice", "note": "checked"},
+                )
+            finally:
+                self._stop_server(server, thread)
+
+            lines = [json.loads(line) for line in cases_path.read_text(encoding="utf-8").splitlines()]
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["case_status"], "handled")
+        self.assertEqual(lines[-1]["handled_source"], "manual")
+        self.assertEqual(lines[-1]["payload"]["note"], "checked")
+
+    def test_callback_endpoint_requires_token_and_handles_case(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "case_sink": {"path": "logs/cases.jsonl"},
+                    "webhooks": {"callback_token": "secret"},
+                    "layout": {"zone_ids": ["1"]},
+                },
+            )
+            cases_path = root / "logs" / "cases.jsonl"
+            cases_path.parent.mkdir()
+            cases_path.write_text(
+                json.dumps(
+                    {
+                        "case_id": "case_batch_000001",
+                        "batch_id": "batch_000001",
+                        "camera_id": "cam_01",
+                        "zone_id": "1",
+                        "zone_label": "区域 1",
+                        "case_type": "time_alarm",
+                        "case_status": "open",
+                        "source_event": "time_alarm",
+                        "created_at": "2026-06-09T09:00:00+08:00",
+                        "updated_at": "2026-06-09T09:00:00+08:00",
+                        "payload": {},
+                    }
+                ),
+                encoding="utf-8",
+            )
+            ctx = ManageContext(config_path=config_path, project_root=root)
+            server, thread = self._serve_once(ctx)
+            try:
+                unauthorized_status, _ = self._request(
+                    server,
+                    "POST",
+                    "/api/manage/webhooks/case-update",
+                    body={"case_id": "case_batch_000001", "status": "handled"},
+                )
+                status, payload = self._request(
+                    server,
+                    "POST",
+                    "/api/manage/webhooks/case-update",
+                    body={"case_id": "case_batch_000001", "status": "handled", "handled_by": "crm-bot"},
+                    headers={"X-Webhook-Token": "secret"},
+                )
+            finally:
+                self._stop_server(server, thread)
+
+            lines = [json.loads(line) for line in cases_path.read_text(encoding="utf-8").splitlines()]
+
+        self.assertEqual(unauthorized_status, 403)
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["handled_source"], "webhook_callback")
+        self.assertEqual(lines[-1]["handled_source"], "webhook_callback")
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py
new file mode 100644
index 0000000..28363da
--- /dev/null
+++ b/tests/test_webhooks.py
@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from cold_display_guard.webhooks import (
+    build_batch_event_payload,
+    build_case_event_payload,
+    load_webhook_settings,
+    send_batch_event_webhooks,
+    send_case_webhooks,
+)
+
+
+UTC = timezone.utc
+
+
+class WebhookTests(unittest.TestCase):
+    def test_load_webhook_settings_from_config(self) -> None:
+        settings = load_webhook_settings(
+            {
+                "webhooks": {
+                    "enabled": True,
+                    "event_url": "https://example.com/events",
+                    "case_url": "https://example.com/cases",
+                    "callback_token": "secret",
+                    "connect_timeout_seconds": 4,
+                    "read_timeout_seconds": 6,
+                }
+            }
+        )
+
+        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.callback_token, "secret")
+        self.assertEqual(settings.connect_timeout_seconds, 4)
+        self.assertEqual(settings.read_timeout_seconds, 6)
+
+    def test_build_batch_event_payload_wraps_runtime_event(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",
+            }
+        )
+
+        self.assertEqual(payload["kind"], "batch_event")
+        self.assertEqual(payload["event"], "time_alarm")
+        self.assertEqual(payload["zone_label"], "区域 1")
+
+    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": "open",
+                "batch_id": "batch_000001",
+                "source_event": "warning_escalated",
+                "handled_source": "",
+                "updated_at": datetime(2026, 6, 9, 9, 5, tzinfo=UTC).isoformat(),
+            }
+        )
+
+        self.assertEqual(payload["kind"], "case_event")
+        self.assertEqual(payload["action"], "updated")
+        self.assertEqual(payload["case_id"], "case_batch_000001")
+
+    def test_send_batch_event_webhooks_delivers_payload(self) -> None:
+        deliveries: list[tuple[str, dict[str, object], tuple[float, float]]] = []
+
+        def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:
+            deliveries.append((url, payload, timeout))
+            return 202, "ok"
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            audit_path = Path(tmpdir) / "webhook_delivery.jsonl"
+            send_batch_event_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",
+                    }
+                ],
+                {
+                    "webhooks": {
+                        "enabled": True,
+                        "event_url": "https://example.com/events",
+                        "connect_timeout_seconds": 4,
+                        "read_timeout_seconds": 6,
+                    }
+                },
+                audit_path,
+                http_post=fake_post,
+            )
+
+        self.assertEqual(deliveries[0][0], "https://example.com/events")
+        self.assertEqual(deliveries[0][1]["kind"], "batch_event")
+        self.assertEqual(deliveries[0][2], (4.0, 6.0))
+
+    def test_send_case_webhooks_delivers_payload(self) -> None:
+        deliveries: list[tuple[str, dict[str, object]]] = []
+
+        def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]:
+            deliveries.append((url, payload))
+            return 200, "ok"
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            audit_path = Path(tmpdir) / "webhook_delivery.jsonl"
+            send_case_webhooks(
+                [
+                    {
+                        "case_id": "case_batch_000001",
+                        "case_type": "time_alarm",
+                        "case_status": "handled",
+                        "batch_id": "batch_000001",
+                        "source_event": "time_alarm",
+                        "handled_source": "manual",
+                        "updated_at": datetime(2026, 6, 9, 9, 10, tzinfo=UTC).isoformat(),
+                    }
+                ],
+                {
+                    "webhooks": {
+                        "enabled": True,
+                        "case_url": "https://example.com/cases",
+                    }
+                },
+                audit_path,
+                http_post=fake_post,
+            )
+
+        self.assertEqual(deliveries[0][0], "https://example.com/cases")
+        self.assertEqual(deliveries[0][1]["kind"], "case_event")
+        self.assertEqual(deliveries[0][1]["action"], "handled")
+
+    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]:
+            raise OSError("network down")
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            audit_path = Path(tmpdir) / "webhook_delivery.jsonl"
+            send_batch_event_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",
+                    }
+                ],
+                {
+                    "webhooks": {
+                        "enabled": True,
+                        "event_url": "https://example.com/events",
+                    }
+                },
+                audit_path,
+                http_post=fake_post,
+            )
+            logged = [json.loads(line) for line in audit_path.read_text(encoding="utf-8").splitlines()]
+
+        self.assertEqual(logged[0]["status"], "error")
+        self.assertEqual(logged[0]["target"], "batch_event")
+        self.assertIn("network down", logged[0]["message"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/web/src/main.js b/web/src/main.js
index bda4a8f..f7e01e5 100644
--- a/web/src/main.js
+++ b/web/src/main.js
@@ -3,6 +3,8 @@ import {
   TRASH_REGION_ID,
   alarmMinutesToSeconds,
   buildCalibrationPayload,
+  buildCaseDisplayModel,
+  buildManualHandlePayload,
   buildPolygonMap,
   buildRuntimeDisplayModel,
   clampZoneCount,
@@ -31,6 +33,8 @@ const state = {
   config: null,
   summary: null,
   events: [],
+  cases: [],
+  caseSummary: null,
   activeTab: "events",
   activeRegion: "1",
   foodZones: defaultFoodZones,
@@ -147,6 +151,11 @@ app.innerHTML = `
           <div class="panel-title">最近事件</div>
           <div id="eventsTable" class="events-table"></div>
         </section>
+        <section class="panel case-panel">
+          <div class="panel-meta">CASE WORKFLOW</div>
+          <div class="panel-title">处置单</div>
+          <div id="casesTable" class="events-table"></div>
+        </section>
       </section>
 
       <section id="settingsView" class="view hidden">
@@ -216,6 +225,7 @@ const els = {
   runtimeProgress: document.querySelector("#runtimeProgress"),
   metrics: document.querySelector("#metrics"),
   eventsTable: document.querySelector("#eventsTable"),
+  casesTable: document.querySelector("#casesTable"),
   statusPill: document.querySelector("#statusPill"),
   activeRegionBadge: document.querySelector("#activeRegionBadge"),
 };
@@ -245,6 +255,7 @@ function wireEvents() {
   document.querySelector("#clearRegion").addEventListener("click", clearRegion);
   document.querySelector("#loadConfigPolygons").addEventListener("click", loadPolygonsFromConfig);
   els.canvas.addEventListener("click", addPoint);
+  els.casesTable.addEventListener("click", handleCaseTableClick);
   window.addEventListener("resize", drawCanvas);
   els.foodZoneCount.addEventListener("input", () => updateFoodZoneCount(els.foodZoneCount.value));
   [els.rtspUrl, els.settingsRtspUrl, els.cameraId, els.timezone, els.maxDwell, els.trashWindow].forEach((input) => {
@@ -309,9 +320,11 @@ async function refreshRuntimeDataSilently() {
 }
 
 async function loadRuntimeData() {
-  const [summaryResult, eventsResult] = await Promise.allSettled([
+  const [summaryResult, eventsResult, casesResult, caseSummaryResult] = await Promise.allSettled([
     apiJson("/api/manage/summary"),
     apiJson("/api/manage/events?limit=1000"),
+    apiJson("/api/manage/cases?limit=1000"),
+    apiJson("/api/manage/cases/summary"),
   ]);
   const errors = [];
   if (summaryResult.status === "fulfilled") {
@@ -326,6 +339,18 @@ async function loadRuntimeData() {
     state.events = [];
     errors.push(`events ${errorMessage(eventsResult.reason)}`);
   }
+  if (casesResult.status === "fulfilled") {
+    state.cases = casesResult.value.items || [];
+  } else {
+    state.cases = [];
+    errors.push(`cases ${errorMessage(casesResult.reason)}`);
+  }
+  if (caseSummaryResult.status === "fulfilled") {
+    state.caseSummary = caseSummaryResult.value;
+  } else {
+    state.caseSummary = null;
+    errors.push(`case summary ${errorMessage(caseSummaryResult.reason)}`);
+  }
   state.runtimeDemoReason = errors.length ? errors.join("；") : "";
 }
 
@@ -515,12 +540,21 @@ function buildRuntimeModel() {
   });
 }
 
+function buildCaseModel() {
+  return buildCaseDisplayModel({
+    summary: state.caseSummary,
+    cases: state.cases,
+  });
+}
+
 function renderRuntimeSections() {
   const runtimeModel = buildRuntimeModel();
+  const caseModel = buildCaseModel();
   renderRuntimeOverview(runtimeModel);
-  renderMetrics(runtimeModel);
+  renderMetrics(runtimeModel, caseModel);
   renderRuntimeProgress(runtimeModel);
   renderEvents(runtimeModel);
+  renderCases(caseModel);
 }
 
 function renderRegionList() {
@@ -739,12 +773,13 @@ function renderRuntimeOverview(model) {
   `;
 }
 
-function renderMetrics(model) {
+function renderMetrics(model, caseModel) {
   const metrics = model.summary?.metrics || {};
   const alertCount = metrics.alert_count ?? 0;
   const warningCount = metrics.warning_count ?? 0;
   const violationCount = metrics.violation_count ?? 0;
   const baselineReady = Boolean(metrics.baseline_ready);
+  const caseMetrics = caseModel?.metrics || {};
   const metricLabel = (label) => label;
   const cards = [
     {label: metricLabel("事件总数"), value: metrics.event_count ?? 0, tone: "neutral"},
@@ -754,6 +789,11 @@ function renderMetrics(model) {
     {label: metricLabel("诊断帧数"), value: metrics.diagnostics_count ?? 0, tone: "neutral"},
     {label: metricLabel("基线状态"), value: baselineReady ? "ready" : "learning", tone: baselineReady ? "good" : "warning"},
     {label: metricLabel("最新报警"), value: metrics.latest_alert_time || "-", tone: metrics.latest_alert_time ? "danger" : "neutral"},
+    {label: metricLabel("待处理处置单"), value: caseMetrics.openCaseCount ?? 0, tone: (caseMetrics.openCaseCount ?? 0) > 0 ? "warning" : "good"},
+    {label: metricLabel("已处理处置单"), value: caseMetrics.handledCaseCount ?? 0, tone: "good"},
+    {label: metricLabel("超时报警单"), value: caseMetrics.timeAlarmCaseCount ?? 0, tone: "neutral"},
+    {label: metricLabel("待丢弃确认单"), value: caseMetrics.pendingDisposalCaseCount ?? 0, tone: "neutral"},
+    {label: metricLabel("升级警告单"), value: caseMetrics.warningEscalatedCaseCount ?? 0, tone: (caseMetrics.warningEscalatedCaseCount ?? 0) > 0 ? "danger" : "good"},
     {label: metricLabel("事件文件"), value: metrics.events_path || "-", tone: "path"},
   ];
   const zoneCounts = metrics.latest_zone_counts || {};
@@ -834,6 +874,36 @@ function renderEvents(model) {
   `;
 }
 
+function renderCases(model) {
+  if (!model.rows.length) {
+    els.casesTable.innerHTML = `<div class="empty">还没有处置单数据</div>`;
+    return;
+  }
+  els.casesTable.innerHTML = `
+    <table>
+      <thead><tr><th>处置单</th><th>类型</th><th>状态</th><th>区域</th><th>批次</th><th>更新时间</th><th>处理来源</th><th>操作</th></tr></thead>
+      <tbody>
+        ${model.rows
+          .map((row) => `
+            <tr class="event-row ${row.tone}">
+              <td>${escapeHtml(row.caseId)}</td>
+              <td>${escapeHtml(row.typeLabel)}</td>
+              <td><span class="event-severity ${row.tone}">${escapeHtml(row.statusLabel)}</span></td>
+              <td>${escapeHtml(row.zone_label || "")}</td>
+              <td>${escapeHtml(row.batch_id || "")}</td>
+              <td>${escapeHtml(row.updated_at || "")}</td>
+              <td>${escapeHtml(row.handledSourceLabel || "-")}</td>
+              <td>${row.case_status === "open"
+                ? `<button type="button" class="secondary-action" data-handle-case="${escapeHtml(row.caseId)}">标记已处理</button>`
+                : `<span class="event-source real">已完成</span>`}</td>
+            </tr>
+          `)
+          .join("")}
+      </tbody>
+    </table>
+  `;
+}
+
 function formatDuration(seconds) {
   const value = Number(seconds);
   if (!Number.isFinite(value) || value <= 0) {
@@ -898,6 +968,39 @@ async function apiJson(path, options = {}) {
   return payload;
 }
 
+function handleCaseTableClick(event) {
+  const button = event.target.closest("[data-handle-case]");
+  if (!button) {
+    return;
+  }
+  handleCase(button.dataset.handleCase);
+}
+
+async function handleCase(caseId) {
+  const handledBy = window.prompt("请输入处理人");
+  if (handledBy === null) {
+    return;
+  }
+  const trimmedHandledBy = handledBy.trim();
+  if (!trimmedHandledBy) {
+    setStatus("处理人不能为空");
+    return;
+  }
+  const note = window.prompt("请输入处理备注（可选）") || "";
+  try {
+    setStatus("正在更新处置单状态...");
+    await apiJson(`/api/manage/cases/${encodeURIComponent(caseId)}/handle`, {
+      method: "POST",
+      body: buildManualHandlePayload(trimmedHandledBy, note),
+    });
+    await loadRuntimeData();
+    renderRuntimeSections();
+    setStatus("处置单已标记为已处理");
+  } catch (error) {
+    setStatus(`更新处置单失败：${error.message}`);
+  }
+}
+
 function setStatus(message) {
   state.status = message;
   els.statusText.textContent = message;
diff --git a/web/src/zone-state.js b/web/src/zone-state.js
index f494ee5..d1c34e0 100644
--- a/web/src/zone-state.js
+++ b/web/src/zone-state.js
@@ -158,6 +158,38 @@ export function buildRuntimeDisplayModel({
   };
 }
 
+export function buildCaseDisplayModel({summary = null, cases = []} = {}) {
+  const metrics = {
+    openCaseCount: Number(summary?.open_case_count || 0),
+    handledCaseCount: Number(summary?.handled_case_count || 0),
+    timeAlarmCaseCount: Number(summary?.time_alarm_case_count || 0),
+    pendingDisposalCaseCount: Number(summary?.pending_disposal_case_count || 0),
+    warningEscalatedCaseCount: Number(summary?.warning_escalated_case_count || 0),
+  };
+  const rows = (Array.isArray(cases) ? cases : [])
+    .map((item) => ({
+      ...item,
+      caseId: String(item.case_id || ""),
+      typeLabel: caseTypeLabel(item.case_type),
+      statusLabel: String(item.case_status || "") === "handled" ? "已处理" : "待处理",
+      tone: String(item.case_status || "") === "handled" ? "good" : "warning",
+      handledSourceLabel: caseHandledSourceLabel(item.handled_source),
+    }))
+    .sort((left, right) => timestampMillis(right.updated_at) - timestampMillis(left.updated_at));
+  return {metrics, rows};
+}
+
+export function buildManualHandlePayload(handledBy, note = "") {
+  const payload = {
+    handled_by: String(handledBy || "").trim(),
+  };
+  const trimmedNote = String(note || "").trim();
+  if (trimmedNote) {
+    payload.note = trimmedNote;
+  }
+  return payload;
+}
+
 export function getRegionColor(id) {
   if (id === TRASH_REGION_ID) {
     return "#111827";
@@ -221,6 +253,32 @@ function containsDemoMarker(value) {
   return text.includes("demo") || text.includes("演示");
 }
 
+function caseTypeLabel(caseType) {
+  if (caseType === "warning_escalated") {
+    return "升级警告";
+  }
+  if (caseType === "pending_disposal") {
+    return "待丢弃确认";
+  }
+  if (caseType === "time_alarm") {
+    return "超时报警";
+  }
+  return String(caseType || "");
+}
+
+function caseHandledSourceLabel(source) {
+  if (source === "manual") {
+    return "人工处理";
+  }
+  if (source === "webhook_callback") {
+    return "回调处理";
+  }
+  if (source === "auto_closed") {
+    return "自动关闭";
+  }
+  return "";
+}
+
 function createEmptyRuntimeSummary(thresholdSeconds) {
   return {
     result_type: "cold_display_guard",
diff --git a/web/test/zone-state.test.js b/web/test/zone-state.test.js
index b3fc67c..0237d2c 100644
--- a/web/test/zone-state.test.js
+++ b/web/test/zone-state.test.js
@@ -5,6 +5,8 @@ import {
   TRASH_REGION_ID,
   alarmMinutesToSeconds,
   buildCalibrationPayload,
+  buildCaseDisplayModel,
+  buildManualHandlePayload,
   buildPolygonMap,
   buildRuntimeDisplayModel,
   classifyEvent,
@@ -628,3 +630,58 @@ test("buildRuntimeDisplayModel uses config threshold when event omits threshold"
     source: "real",
   }]);
 });
+
+test("buildCaseDisplayModel normalizes case rows and summary metrics", () => {
+  const model = buildCaseDisplayModel({
+    summary: {
+      open_case_count: 1,
+      handled_case_count: 2,
+      time_alarm_case_count: 1,
+      pending_disposal_case_count: 1,
+      warning_escalated_case_count: 1,
+    },
+    cases: [
+      {
+        case_id: "case_batch_000001",
+        case_type: "warning_escalated",
+        case_status: "open",
+        zone_label: "区域 1",
+        batch_id: "batch_000001",
+        updated_at: "2026-06-09T09:10:00+08:00",
+        handled_source: "",
+      },
+      {
+        case_id: "case_batch_000002",
+        case_type: "time_alarm",
+        case_status: "handled",
+        zone_label: "区域 2",
+        batch_id: "batch_000002",
+        updated_at: "2026-06-09T09:12:00+08:00",
+        handled_source: "manual",
+      },
+    ],
+  });
+
+  assert.deepEqual(model.metrics, {
+    openCaseCount: 1,
+    handledCaseCount: 2,
+    timeAlarmCaseCount: 1,
+    pendingDisposalCaseCount: 1,
+    warningEscalatedCaseCount: 1,
+  });
+  assert.equal(model.rows[0].caseId, "case_batch_000002");
+  assert.equal(model.rows[0].statusLabel, "已处理");
+  assert.equal(model.rows[0].tone, "good");
+  assert.equal(model.rows[1].typeLabel, "升级警告");
+  assert.equal(model.rows[1].statusLabel, "待处理");
+});
+
+test("buildManualHandlePayload trims handled_by and keeps optional note", () => {
+  assert.deepEqual(buildManualHandlePayload(" alice ", " checked "), {
+    handled_by: "alice",
+    note: "checked",
+  });
+  assert.deepEqual(buildManualHandlePayload("bob", ""), {
+    handled_by: "bob",
+  });
+});