diff --git a/.trellis/spec/hyb-oa/deployment-trialrun.md b/.trellis/spec/hyb-oa/deployment-trialrun.md
index 86ff396a..8c39bfdd 100644
--- a/.trellis/spec/hyb-oa/deployment-trialrun.md
+++ b/.trellis/spec/hyb-oa/deployment-trialrun.md
@@ -1443,18 +1443,42 @@ PGPASSWORD="$POSTGRES_PASSWORD" docker exec -e PGPASSWORD db pg_dump ...
 - Units: `hyb-oa-public-app-health.service` and
   `hyb-oa-public-app-health.timer`; the timer uses `OnCalendar=*:0/5`.
 - Local contract: `bash scripts/qa/run-app-health-monitor-contract.sh`.
+- Application statuses: `FAIL_APP_HEALTH`, `PASS_APP_HEALTH_READY` and
+  `FAIL_APP_HEALTH_FLAPPING`.
+- Probe-state schema v2 phases: `healthy`, `failure_pending`, `alerted`,
+  `recovery_pending`, `recovery_watch` and `flapping`.
 
 ### 3. Contracts
 
 - Run the probe from the existing ARM host against an exact HTTPS `/ready` URL.
   Success requires trusted TLS, HTTP 2xx, bounded JSON and `data.ok === true`.
-- Count any stable failure category toward the same consecutive-failure
-  threshold. The first failure is pending; the second evaluates a failure
-  alert. Once alerting, a changed category may send a new alert, identical
-  failures are deduplicated, and one later success sends one recovery notice.
-- Persist only schema version, count, stable category and the alert state in a
-  root-owned `0700` directory with atomic `0600` files. Never persist or send
-  the URL, resolved address, response payload, credential or raw exception.
+- Classify DNS resolution and TCP connection failures separately before the
+  shared fallback catches `OSError`; keep TLS, timeout, HTTP, response-size and
+  response-contract categories stable. Count any stable failure category toward
+  the same consecutive-failure threshold. The first failure is pending and the
+  second evaluates a failure alert; identical failures remain deduplicated.
+- After an alert, require two consecutive healthy samples before emitting one
+  recovery. A single healthy sample followed by another failure stays in the
+  existing incident and sends no recovery noise.
+- Keep the recovered category under a 60-minute observation window. The same
+  category reaching the two-failure threshold in that window emits one
+  `FAIL_APP_HEALTH_FLAPPING` reminder. Later switches stay local until health
+  has remained continuous for 30 minutes, then the probe exits flapping
+  silently. A different category in the recovery window remains a new normal
+  incident and still uses the two-failure threshold.
+- Persist only schema version, phase, bounded counts, stable categories and
+  necessary Unix timestamps in a root-owned `0700` directory with atomic
+  `0600` files. Schema v1 migrates conservatively without backfilling a
+  notification. Never persist or send the URL, resolved address, response
+  payload, credential or raw exception.
+- The probe returns a notification candidate on repeated samples while the
+  corresponding state is active; the shared helper owns final deduplication.
+  This lets a failed send retry on a later sample without repeating a
+  successfully sent message. Healthy flapping samples may retry the stable
+  flapping reminder but never produce a recovery candidate.
+- Application messages use the explicit titles `OA 公网健康告警`,
+  `OA 公网健康恢复` and `OA 公网健康链路反复抖动`. Shared backup/dead-man
+  statuses retain their generic operations titles.
 - Application-health probe state and alert state are separate from backup and
   dead-man state. The installer must deploy the helper revision that passed the
   backup contract; an older installed helper may reject the new status tokens.
@@ -1474,11 +1498,19 @@ PGPASSWORD="$POSTGRES_PASSWORD" docker exec -e PGPASSWORD db pg_dump ...
 
 - Non-HTTPS, userinfo, query, fragment or non-`/ready` URL -> reject before a
   request with `ready_url_invalid`.
-- DNS/connect, TLS, timeout, HTTP, oversized response or invalid JSON/contract
-  -> stable redacted category and non-zero oneshot exit.
+- DNS, TCP connection, TLS, timeout, HTTP, oversized response or invalid
+  JSON/contract -> distinct stable redacted category and non-zero oneshot exit.
 - Invalid owner/mode/link/state or alert-helper failure -> fail closed with a
   fixed local error; do not include the offending value.
 - A first isolated failure followed by health -> no failure or recovery message.
+- One healthy sample after an alerted failure -> `recovery_pending`, no recovery
+  message; the second consecutive sample -> one recovery.
+- Same-category recurrence before 60 minutes -> one flapping reminder; at
+  exactly 60 minutes -> a new normal incident, not flapping.
+- Flapping health shorter than 30 minutes -> remain flapping; at 30 continuous
+  minutes -> return to `healthy` without a message.
+- Invalid/unknown schema, out-of-range counter/timestamp or inconsistent phase
+  fields -> `local_state_invalid`, with no raw state in output.
 - Any overridden managed target or service name -> installer stops before
   `systemctl`, file replacement or removal.
 - SSH succeeds but `sudo -n` requests a password -> stop before upload or
@@ -1491,8 +1523,9 @@ PGPASSWORD="$POSTGRES_PASSWORD" docker exec -e PGPASSWORD db pg_dump ...
 
 ### 5. Good/Base/Bad Cases
 
-- Good: ARM observes two consecutive unavailable checks, sends one redacted
-  category alert, suppresses repeats and sends one recovery after health returns.
+- Good: ARM observes two consecutive unavailable checks, sends one explicit
+  OA-public-health alert, suppresses repeats, confirms recovery with two healthy
+  samples, and emits only one flapping reminder for a same-category recurrence.
 - Base: repository contracts pass while units remain disabled; report the code
   as ready for an approved rollout, not as an active monitor.
 - Bad: run the timer only on the OA host, send raw `/ready` errors, install and
@@ -1501,9 +1534,15 @@ PGPASSWORD="$POSTGRES_PASSWORD" docker exec -e PGPASSWORD db pg_dump ...
 ### 6. Tests Required
 
 - The local HTTPS stub must assert healthy baseline, pending first failure,
-  second-failure alert, identical deduplication, changed-category alert, one
-  recovery, transient suppression, TLS/connect/timeout/HTTP/size/JSON failures,
-  corrupt private state, URL rejection and absence of sensitive output.
+  second-failure alert, identical deduplication, changed-category alert,
+  two-sample recovery, transient suppression, same-category flapping,
+  different-category recurrence, silent 30-minute exit, exact 30/60-minute
+  boundaries, schema-v1 migration, separate DNS/TCP classification,
+  TLS/timeout/HTTP/size/JSON failures, corrupt private state, URL rejection and
+  absence of sensitive output.
+- Captured application messages must assert the three explicit titles and
+  stable status/category only. The automatic-backup contract must prove shared
+  helper titles, recovery and deduplication remain unchanged.
 - Run Python compile and shell syntax checks, validate the five-minute calendar,
   and assert disabled-by-default plus exact managed targets and retained state.
 - For a remote rollout, record boolean evidence for pre-write `sudo -n`, exact
@@ -1520,7 +1559,9 @@ PGPASSWORD="$POSTGRES_PASSWORD" docker exec -e PGPASSWORD db pg_dump ...
 #### Wrong
 
 ```bash
-# Local contract passed, so install and enable remotely and call it active.
+# One healthy sample arrived, so announce recovery immediately and keep
+# alternating alert/recovery messages during a flapping link.
+bash scripts/qa/run-app-health-monitor-contract.sh
 bash deploy/trialrun/install-offhost-app-health-monitor.sh enable
 ```
 
@@ -1529,6 +1570,8 @@ bash deploy/trialrun/install-offhost-app-health-monitor.sh enable
 ```bash
 bash scripts/qa/run-app-health-monitor-contract.sh
 bash scripts/qa/run-automatic-backup-contract.sh
+# Runtime contract: fail x2 -> alert; healthy x2 -> recovery; same fail x2
+# within 60m -> one flapping reminder; continuous health 30m -> silent exit.
 # After separate ARM-write approval: install disabled, inspect, then seek
 # enable/real-message/drill approval before claiming the monitor is active.
 <explicit-controlled-paths> \
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/check.jsonl b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/check.jsonl
new file mode 100644
index 00000000..9dd3234a
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/check.jsonl
@@ -0,0 +1 @@
+{"_example": "Fill with {\"file\": \"<path>\", \"reason\": \"<why>\"}. Put spec/research files only — no code paths. Run `python3 .trellis/scripts/get_context.py --mode packages` to list available specs. Delete this line once real entries are added."}
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/design.md b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/design.md
new file mode 100644
index 00000000..798bee6f
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/design.md
@@ -0,0 +1,106 @@
+# 企微公网健康告警防抖与诊断增强设计
+
+## 1. Architecture and Boundaries
+
+本任务保持现有“ARM systemd timer → 公网探测器 → 脱敏 summary → 共用企微 helper”的边界，不引入守护进程、数据库或外部状态服务。
+
+- 探测器负责网络阶段分类、连续样本计数、恢复确认和抖动周期判定。
+- 告警 helper 继续负责 summary 校验、消息文案、Webhook 安全读取和发送侧去重。
+- 安装脚本继续固定 5 分钟 timer、受控路径、root 所有权和默认 disabled。
+- 本地合同测试通过 stub、受控 socket/DNS 失败和测试 capture 验证，不连接真实机器人。
+
+## 2. Failure Classification Contract
+
+分类优先级固定为：
+
+1. `socket.gaierror` 或等价解析失败 → `dns_resolution_failed`；
+2. TCP 建连拒绝、重置、不可达及其他建连类错误 → `tcp_connection_failed`；
+3. TLS 证书或握手校验失败 → `tls_validation_failed`；
+4. socket/请求超时 → `request_timeout`；
+5. 非 2xx HTTP → `http_unavailable`；
+6. 响应超过上限 → `response_too_large`；
+7. JSON、编码或 `{data: {ok: true}}` 合同不满足 → `response_contract_invalid`。
+
+异常类型按从具体到通用的顺序匹配，避免 `OSError` 提前吞掉 DNS、TLS 或超时。对无法进一步安全细分的连接阶段异常归入 `tcp_connection_failed`，但不把异常文本写入状态、summary 或日志。
+
+## 3. State Model
+
+### 3.1 State schema
+
+探测状态升级为显式版本，字段只包含：
+
+- `phase`：`healthy`、`failure_pending`、`alerted`、`recovery_pending`、`recovery_watch`、`flapping`；
+- `activeCategory` / `lastRecoveredCategory`：稳定类别 token 或 `null`；
+- `consecutiveFailures` / `consecutiveSuccesses` / `flapTransitions`：有界非负整数；
+- `recoveredAt` / `healthySince`：必要时存在的非负 Unix 秒时间戳。
+
+旧版 schema 只含连续失败数和最后类别。部署时采用保守兼容迁移：有效旧状态映射为等价的 `healthy` 或 `failure_pending`，不凭旧状态补发消息；非法或未知 schema 继续 fail closed。告警 helper 自身状态 schema 保持向后兼容。
+
+### 3.2 Normal incident flow
+
+```text
+healthy --失败1--> failure_pending --失败2--> alerted（发送公网健康告警）
+alerted --成功1--> recovery_pending --成功2--> recovery_watch（发送公网健康恢复）
+recovery_pending --失败--> alerted（不发送恢复；持续事件仍去重）
+```
+
+失败阈值统计持续不可用样本，告警使用达到阈值时的最新稳定类别。已告警期间相同类别不重复发送；类别发生变化时沿用现有最多一次更新通知语义。
+
+### 3.3 Recovery watch and flapping flow
+
+```text
+recovery_watch
+  --60分钟内同类失败达到2次--> flapping（发送一次链路反复抖动提醒）
+  --不同类别失败达到2次-----> alerted（作为新事件发送普通告警）
+  --观察窗到期且健康--------> healthy（静默）
+
+flapping
+  --任意失败/短时成功-------> flapping（只更新脱敏计数）
+  --连续健康满30分钟--------> healthy（静默退出）
+```
+
+“连续健康满 30 分钟”以第一条连续成功样本的时间为起点，期间任何失败都会清空 `healthySince`。状态转换实现为可直接传入当前时间的纯函数，使合同测试可使用合成时间，不依赖等待真实 30/60 分钟，也不增加生产测试参数。
+
+## 4. Notification Contract
+
+应用健康事件使用专用标题：
+
+- `FAIL_APP_HEALTH`：`OA 公网健康告警`；
+- `PASS_APP_HEALTH_READY`：`OA 公网健康恢复`；
+- `FAIL_APP_HEALTH_FLAPPING`：`OA 公网健康链路反复抖动`。
+
+正文只保留状态、稳定类别和操作提示。helper 根据应用健康状态选择专用文案，其他备份/dead-man/提醒状态继续使用原有通用标题。抖动事件采用独立、白名单内的稳定状态 token，防止与普通失败去重键互相覆盖。
+
+状态转换返回“通知候选”而不是自行发送：处于普通故障、已确认恢复或抖动故障时，可在后续同类样本上重复交给 helper 评估，由 helper 的现有状态完成去重；如果上次发送失败且 helper 状态未落盘，下一次样本可以重试。抖动期的健康样本和静默退出不产生恢复候选，由此保证抖动退出不产生额外恢复消息。
+
+## 5. Data Flow and Failure Handling
+
+1. 校验 URL、状态目录、状态文件及阈值。
+2. 读取并严格校验/迁移探测状态。
+3. 执行一次 `/ready` 探测并生成成功或稳定失败类别。
+4. 用当前时间和样本驱动纯状态转换，得到新状态及可选通知候选。
+5. 原子写入 `0600` 探测状态；有通知候选时才写脱敏 summary 并调用 helper，由 helper 去重或重试。
+6. 探测失败仍返回非零，健康样本返回零；通知发送或本地状态失败继续 fail closed。
+
+状态先落盘再通知，保持当前“状态不可安全保存则不发送误导消息”的边界。若通知失败，命令返回失败并保留可诊断的稳定错误；后续同类样本再次提交候选，helper 仅在自身发送状态尚未落盘时重试，合同测试必须证明成功发送后不会重复。
+
+## 6. Compatibility and Trade-offs
+
+- 2 次恢复确认会把恢复消息延后约 5 分钟，这是减少短时成功噪声的明确代价。
+- 同类抖动期不逐次通知，会降低每次瞬断的可见性；本地计数和最终稳定退出状态保留诊断能力。
+- 不同类别复发仍作为新事件处理，消息可能多于“全部静默”，但避免把新根因错误吞掉。
+- 不修改 `/ready` 合同、timer 周期、Webhook 文件、summary 隐私边界或非应用健康告警语义。
+
+## 7. Rollout and Rollback
+
+- 本地阶段仅提交代码、合同测试和必要文档，不触碰 ARM/DNS/真实企微。
+- rollout 获批后先以 `install` 模式部署 disabled unit，核对文件权限、unit 内容、旧状态迁移和只读探测，再单独批准 `enable`。
+- 启用后先观察分类计数和消息频率；DNS 调整只以实际 `dns_resolution_failed` 证据为依据。
+- 回滚时停用 timer，恢复上一版探测器/helper/unit；保留并备份脱敏状态。若旧版不能读取新 schema，则在明确目标文件后将新状态移出状态目录，由旧版建立健康基线，避免伪造恢复通知。
+
+## 8. Security and Privacy
+
+- 禁止写入 URL、域名、IP、文件路径、异常文本、响应 payload、Webhook 或 Secret。
+- capture 仅在既有显式测试模式下可用，测试输出权限保持 `0600`。
+- 所有状态枚举和类别经过白名单/token 校验；计数与时间戳必须有类型和范围检查。
+- 生产、企业微信和 DNS 写入继续受独立批准门禁控制。
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.jsonl b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.jsonl
new file mode 100644
index 00000000..9dd3234a
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.jsonl
@@ -0,0 +1 @@
+{"_example": "Fill with {\"file\": \"<path>\", \"reason\": \"<why>\"}. Put spec/research files only — no code paths. Run `python3 .trellis/scripts/get_context.py --mode packages` to list available specs. Delete this line once real entries are added."}
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.md b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.md
new file mode 100644
index 00000000..f2969f11
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/implement.md
@@ -0,0 +1,69 @@
+# 企微公网健康告警防抖与诊断增强实施计划
+
+## Preconditions
+
+- 用户审核并明确批准本规划后，才运行 `task.py start`。
+- Phase 2 开始前加载 `trellis-before-dev`，读取 operations 相关 spec、当前任务三份规划材料及引用的运维事实源。
+- 本任务采用 Codex inline 实施，不整理或依赖 `implement.jsonl` / `check.jsonl`。
+- 保持现有报销任务未提交改动不动，不纳入本任务 diff、校验或提交。
+
+## Ordered Implementation Checklist
+
+1. **探测分类**
+   - 在 `scripts/check-public-app-health.py` 按具体异常优先级拆分 DNS 与 TCP 类别。
+   - 保持 TLS、超时、HTTP、响应过大和响应合同分类稳定，日志只打印 token。
+2. **状态 schema 与纯状态转换**
+   - 引入显式新版探测状态、严格校验、有界计数和旧 schema 保守迁移。
+   - 将“样本 + 当前时间 → 新状态 + 可选通知候选”提取为纯逻辑。
+   - 实现 2 次失败、2 次成功、60 分钟同类复发提醒、抖动期静默计数和连续健康 30 分钟静默退出。
+3. **专用消息文案与兼容**
+   - 为应用健康普通告警、恢复和抖动事件增加明确的“OA 公网健康”标题。
+   - 扩展 helper 白名单及发送判断，但保持备份、dead-man 和其他既有状态文案/去重行为不变。
+4. **合同测试**
+   - 扩充 `scripts/qa/run-app-health-monitor-contract.sh`：分别模拟 DNS、TCP、TLS、超时、HTTP、格式、过大响应。
+   - 使用纯状态转换或合成时间覆盖全部正常、恢复确认、同类抖动、不同类别复发、30/60 分钟边界和旧状态迁移。
+   - 断言消息数量、顺序、专用标题、静默退出及敏感内容禁入。
+   - 回归损坏状态 fail-closed、文件权限、固定路径、默认 disabled 和 5 分钟 timer。
+5. **共享合同与事实源**
+   - 运行自动备份合同，证明共用 helper 无回归。
+   - 仅在形成稳定可复用规则时，更新既有 `.trellis/spec/hyb-oa/` 或 `docs/试运行与自动化验证手册.md`；不新建无必要文档。
+6. **本地质量门禁**
+   - 逐项运行下列验证，限定敏感扫描和 diff 审核在本任务相关文件。
+   - 对权限/状态机变更完成主会话只读复核；inline 模式不派发 implement/check 子代理。
+7. **提交与交付**
+   - 只暂存当前 Trellis task 和本任务实现/测试/必要事实源文件。
+   - 使用中文 commit message，并引用 `wecom-public-health-alert-diagnostics-and-debounce`。
+   - 本地完成后仅报告“本地验证通过”；未获 ARM 部署与公网检查批准前，不告知用户可在企业微信测试。
+
+## Validation Commands
+
+```bash
+python3 -m py_compile scripts/check-public-app-health.py scripts/send-wecom-group-alert.py
+bash scripts/qa/run-app-health-monitor-contract.sh
+bash scripts/qa/run-automatic-backup-contract.sh
+bash -n deploy/trialrun/install-offhost-app-health-monitor.sh
+node scripts/qa/check-process-contract.mjs --baseline --json
+node scripts/qa/audit-docs.mjs --check-registry
+node scripts/qa/check-trellis-governance.mjs
+python3 ./.trellis/scripts/task.py validate 08-23-wecom-public-health-alert-diagnostics-and-debounce
+git diff --check
+```
+
+敏感扫描采用稳定关键词和本任务文件清单，不打印潜在 Secret 值或原始 payload。
+
+## Risky Files and Rollback Points
+
+- `scripts/check-public-app-health.py`：状态迁移和通知时机会改变；每一步先以合同测试锁定消息数量，再进入下一步。
+- `scripts/send-wecom-group-alert.py`：被备份/dead-man 共用；任何改动必须先通过应用健康合同，再通过完整自动备份合同。
+- `deploy/trialrun/install-offhost-app-health-monitor.sh`：若无需新增生产参数则尽量不改；如需改动，必须保持默认 disabled 和固定安全目标。
+- 状态 schema：必须支持有效旧状态保守迁移；回滚前先停 timer，并只对已解析的精确状态文件执行保留/移出操作。
+
+## External Approval Gates
+
+以下动作均不属于本地实施批准，执行前必须分别再次获得明确授权：
+
+1. ARM 文件、systemd 或生产配置写入；
+2. 启用/重启公网健康 timer；
+3. 向真实企业微信群发送测试或生产消息；
+4. 修改 DNS、resolver 或任何网络策略；
+5. 写入真实业务数据或执行恢复操作。
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/prd.md b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/prd.md
new file mode 100644
index 00000000..f7d2b888
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/prd.md
@@ -0,0 +1,92 @@
+# 企微公网健康告警防抖与诊断增强
+
+## Goal
+
+让 ARM 公网健康探测准确区分 DNS、TCP、TLS、HTTP 和响应合同故障，以明确来源的企业微信文案报告真实事件，并在链路反复抖动时减少无意义的“告警—恢复”消息，同时保持约 10 分钟量级的持续 OA 不可用发现能力。
+
+## Background and Confirmed Facts
+
+- 上一排查任务已确认群内反复消息来自 ARM 对公网 `/ready` 的健康探测，不是备份失败或数据恢复；生产自动备份和新鲜度 watchdog 近 14 天均为 14/14 成功、0 条失败。
+- 实施前基线 `986668a58762` 每 5 分钟探测，连续 2 次失败即告警、1 次成功即恢复；基线实现见该提交的 `scripts/check-public-app-health.py:220-279` 和 `deploy/trialrun/install-offhost-app-health-monitor.sh:94-113`。
+- 实施前基线 `scripts/check-public-app-health.py:131-170` 将 DNS 与 TCP 故障合并为 `dns_or_connect_failed`。
+- 实施前基线 `scripts/send-wecom-group-alert.py:133-140` 为所有健康恢复统一使用“内部 OA 运维恢复通知”，接收人无法直接判断这是公网健康监控。
+- 当前 ARM 健康。单一静态 DNS 是可疑单点，但现有证据不足以断言历史故障均由 DNS 导致，因此本任务先增强诊断和防抖，不直接修改 DNS。
+
+## Requirements
+
+### R1 — 故障阶段分类
+
+- 将 `dns_or_connect_failed` 至少拆分为 `dns_resolution_failed` 与 `tcp_connection_failed`。
+- 保留并稳定区分 `tls_validation_failed`、`request_timeout`、`http_unavailable`、`response_contract_invalid` 和 `response_too_large`。
+- 分类只记录稳定类别，不记录解析结果、目标地址、异常原文、响应正文或请求 payload。
+
+### R2 — 公网健康专用文案
+
+- `FAIL_APP_HEALTH`、`PASS_APP_HEALTH_READY` 和新增的 `FAIL_APP_HEALTH_FLAPPING` 抖动事件必须明确显示“OA 公网健康”。
+- 恢复文案不得使用可能被理解为数据库、附件或备份恢复的泛化标题。
+- 自动备份、异机 dead-man、季度提醒等共用告警 helper 的既有标题、发送条件和脱敏合同保持兼容。
+
+### R3 — 均衡防抖状态机
+
+- 保持 5 分钟探测频率和连续 2 次失败告警，持续不可用仍在约 10 分钟量级发送一次告警。
+- 已发送故障告警后，必须连续 2 次成功才发送一次恢复；单次短暂成功只进入恢复待确认状态，不发送恢复。
+- 正常恢复后的 60 分钟观察窗内，同一故障类别再次连续达到失败阈值时，只发送一次“OA 公网健康链路反复抖动”提醒并进入抖动状态。
+- 抖动状态内后续失败、成功及同类切换只更新本地脱敏计数，不逐次发送告警或恢复。
+- 抖动状态在连续健康满 30 分钟后静默退出；退出本身不另发消息。之后的新故障按新的正常周期处理。
+- 恢复观察窗内出现不同故障类别时，按新的故障事件执行连续 2 次失败门槛，避免隐藏不同根因；同一持续故障期间类别变化仍按既有“类别变化最多通知一次”的兼容行为处理。
+
+### R4 — 状态与隐私边界
+
+- 本地状态仅允许保存 schema 版本、稳定状态/类别、连续成功或失败计数、必要的 Unix 时间戳和抖动计数。
+- 状态文件继续使用原子写入、`0600` 权限和严格 schema 校验；损坏或越界状态 fail closed，并只输出稳定脱敏错误类别。
+- 不保存主机名、IP、URL、路径、异常原文、Webhook、Secret 或原始响应；测试捕获不得访问真实企业微信。
+
+### R5 — 发布门禁
+
+- 本地实现和合同验证不授权 ARM 写入、systemd 变更、DNS 修改或真实企业微信消息测试。
+- 后续 ARM rollout 必须再次取得明确批准，并按“disabled 安装 → 只读检查 → 受控启用”执行；DNS 调整必须依据增强分类后的观测证据另行决策。
+
+## Scope
+
+- `scripts/check-public-app-health.py` 的故障分类、恢复确认、抖动状态机和脱敏状态 schema。
+- `scripts/send-wecom-group-alert.py` 的公网健康专用文案及共享行为兼容。
+- `scripts/qa/run-app-health-monitor-contract.sh`、`scripts/qa/run-automatic-backup-contract.sh` 和 ARM 安装合同的相关回归。
+- 必要的既有运维事实源更新，以及后续 ARM rollout/rollback 操作设计。
+
+## Out of Scope
+
+- 规划阶段修改产品代码、ARM 配置、systemd 或 DNS。
+- 自动重启 OA、切换流量、数据库修复或任何数据恢复。
+- 部署 Prometheus/Alertmanager 或引入新的消息供应商。
+- 未取得证据前指定公共 DNS、覆盖主机网络策略或停用健康监控。
+- 在本任务内执行生产部署、真实企微消息发送或真实数据写入。
+
+## Acceptance Criteria
+
+- [x] **AC1 / R1**：无外网依赖的合同测试可稳定区分 DNS、TCP、TLS、超时、HTTP、响应格式及响应过大故障。
+- [x] **AC2 / R2**：应用健康告警、恢复及抖动提醒的标题均明确包含“OA 公网健康”，测试证明消息不含地址、路径、异常原文或 payload。
+- [x] **AC3 / R3**：5 分钟频率与 2 次失败阈值保持不变；持续故障只发送一次告警，相同故障持续期间不重复发送。
+- [x] **AC4 / R3**：状态机测试证明 1 次成功不恢复、2 次连续成功只恢复一次；恢复后 60 分钟内同类复发只提醒一次，抖动期切换静默计数，连续健康 30 分钟后静默退出。
+- [x] **AC5 / R3**：恢复观察窗内的不同类别故障不会被同类抖动规则吞掉，并仍经过连续 2 次失败门槛。
+- [x] **AC6 / R4**：本地状态 schema、权限、原子写入、损坏状态 fail-closed 和敏感字段禁入均有自动化覆盖。
+- [x] **AC7 / R2**：自动备份、异机 dead-man 和既有共享 helper 合同全部通过，非应用健康消息行为无回归。
+- [x] **AC8 / R5**：安装合同保持默认 disabled、5 分钟周期、固定安全目标及独立启用门禁；本地验证不产生真实企微或远端写入。
+
+## Test Plan
+
+- `python3 -m py_compile scripts/check-public-app-health.py scripts/send-wecom-group-alert.py`
+- `bash scripts/qa/run-app-health-monitor-contract.sh`
+- `bash scripts/qa/run-automatic-backup-contract.sh`
+- `bash -n deploy/trialrun/install-offhost-app-health-monitor.sh`
+- `node scripts/qa/check-process-contract.mjs --baseline --json`
+- `node scripts/qa/audit-docs.mjs --check-registry`
+- `node scripts/qa/check-trellis-governance.mjs`
+- 针对本任务变更执行敏感信息扫描和 `git diff --check`。
+- ARM 写入获批后才执行 disabled 安装、只读状态检查、受控启用和回滚验证；真实消息测试与 DNS 修改仍需分别批准。
+
+## Local Verification Result
+
+- `bash scripts/qa/run-app-health-monitor-contract.sh`：PASS，36 项检查。
+- `bash scripts/qa/run-automatic-backup-contract.sh`：PASS，74 项检查，共享备份/dead-man 标题和去重无回归。
+- Python/Shell 语法、process baseline、文档注册审计、敏感扫描、Trellis governance、task validate 和 `git diff --check` 均通过；文档审计仅保留既有的未登记文件 warning。
+- 当前结论为 `LOCAL_FIX_VERIFIED`；ARM 部署、真实群消息、DNS 调整和外部行为验证仍为 `EXTERNAL_GATED`。
diff --git a/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/task.json b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/task.json
new file mode 100644
index 00000000..f0132a93
--- /dev/null
+++ b/.trellis/tasks/08-23-wecom-public-health-alert-diagnostics-and-debounce/task.json
@@ -0,0 +1,33 @@
+{
+  "id": "wecom-public-health-alert-diagnostics-and-debounce",
+  "name": "wecom-public-health-alert-diagnostics-and-debounce",
+  "title": "企微公网健康告警防抖与诊断增强",
+  "description": "细分 ARM 公网健康探测故障阶段，优化企微告警来源文案与防抖状态机，并为后续 DNS 调整建立证据门槛。",
+  "status": "in_progress",
+  "dev_type": null,
+  "scope": "operations",
+  "package": null,
+  "priority": "P2",
+  "creator": "shunli",
+  "assignee": "shunli",
+  "createdAt": "2026-08-23",
+  "completedAt": null,
+  "branch": null,
+  "base_branch": "main",
+  "worktree_path": null,
+  "commit": null,
+  "pr_url": null,
+  "subtasks": [],
+  "children": [],
+  "parent": null,
+  "relatedFiles": [
+    "scripts/check-public-app-health.py",
+    "scripts/send-wecom-group-alert.py",
+    "scripts/qa/run-app-health-monitor-contract.sh",
+    "scripts/qa/run-automatic-backup-contract.sh",
+    ".trellis/spec/hyb-oa/deployment-trialrun.md",
+    "docs/试运行与自动化验证手册.md"
+  ],
+  "notes": "本地实现与合同验证完成；ARM 部署、真实企微消息、DNS 调整和外部行为验证仍需独立批准。",
+  "meta": {}
+}
diff --git a/docs/试运行与自动化验证手册.md b/docs/试运行与自动化验证手册.md
index 55ecdb20..5fd3aa7b 100644
--- a/docs/试运行与自动化验证手册.md
+++ b/docs/试运行与自动化验证手册.md
@@ -154,9 +154,11 @@ cd client && E2E_BASE_URL=https://oa.hyinda.com npx playwright test tests/e2e/dg
 
 ### ARM 异机应用健康监控
 
-`07-27-app-health-monitoring` 为公网 `/ready` 增加轻量异机探测：ARM 每 5 分钟经标准 HTTPS 校验访问一次；连续两次失败才发送一条脱敏企微群机器人告警，相同类别持续失败不重复发送，恢复后只通知一次。它覆盖 OA 应用、数据库就绪、OA 主机不可达、DNS/Caddy/TLS 已失效等基础故障，不替代登录、审批或报销等业务 smoke，也不执行自动重启或恢复。
+`07-27-app-health-monitoring` 为公网 `/ready` 增加轻量异机探测：ARM 每 5 分钟经标准 HTTPS 校验访问一次；连续两次失败才发送一条脱敏企微群机器人告警，相同类别持续失败不重复发送。它覆盖 OA 应用、数据库就绪、OA 主机不可达、DNS/Caddy/TLS 已失效等基础故障，不替代登录、审批或报销等业务 smoke，也不执行自动重启或恢复。
 
-2026-07-27 已完成 ARM 安装、启用和真实故障/恢复演练：timer 为 enabled/active，首次失败 pending、第二次发送、重复失败降噪、正式恢复发送、重复健康降噪，用户确认群内实际收到一条故障与一条恢复通知；备份 dead-man 未受影响，临时上传和临时 sudoers 已删除。后续重装、真实消息或再次演练仍需单独批准。先运行本地无真实消息合同：
+2026-07-27 已完成 ARM 安装、启用和真实故障/恢复演练：timer 为 enabled/active，首次失败 pending、第二次发送、重复失败降噪、正式恢复发送、重复健康降噪，用户确认群内实际收到一条故障与一条恢复通知；备份 dead-man 未受影响，临时上传和临时 sudoers 已删除。
+
+`08-23-wecom-public-health-alert-diagnostics-and-debounce` 已完成本地候选增强：DNS 与 TCP 分类分离；应用消息标题明确为“OA 公网健康”；恢复改为连续 2 次健康确认；正常恢复后 60 分钟内同类故障再次达到 2 次失败时只发送一次链路反复抖动提醒，后续切换只保留脱敏计数，连续健康 30 分钟后静默退出。该结论目前只代表本地合同通过，尚未部署到 ARM，不能据此声称群消息已经改变。ARM rollout、真实消息和 DNS 修改仍需分别批准。先运行本地无真实消息合同：
 
 ```bash
 bash scripts/qa/run-app-health-monitor-contract.sh
@@ -171,7 +173,7 @@ systemctl list-timers hyb-oa-public-app-health.timer --no-pager
 systemctl status hyb-oa-public-app-health.service --no-pager
 ```
 
-受控演练必须证明：第一次失败仅进入 pending；第二次持续失败只发一条；重复失败降噪；恢复后只发一条。群消息只允许状态和稳定类别，不粘贴地址、路径、响应、凭证或 journal 原文。ARM 自身宕机仍会令该 timer 失声，首版保留为人工 timer 巡检风险，不据此宣称具备第三方 dead-man。
+增强版受控演练必须证明：第一次失败仅进入 pending；第二次持续失败只发一条；重复失败降噪；第一次健康不恢复、第二次连续健康只恢复一条；60 分钟内同类复发只产生一条抖动提醒；连续健康 30 分钟后静默退出。群消息只允许状态和稳定类别，不粘贴地址、路径、响应、凭证或 journal 原文。ARM 自身宕机仍会令该 timer 失声，首版保留为人工 timer 巡检风险，不据此宣称具备第三方 dead-man。
 
 发布候选进入公网或小范围试运行前，先按 [`docs/trialrun/发布试运行准备清单.md`](trialrun/发布试运行准备清单.md) 完成证据整理、权限直连审计、部署预检、live smoke 和试运行分级。
 
diff --git a/scripts/check-public-app-health.py b/scripts/check-public-app-health.py
index 2f08f80f..b2d74b9c 100755
--- a/scripts/check-public-app-health.py
+++ b/scripts/check-public-app-health.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Probe the public OA readiness endpoint and drive a deduplicated alert."""
+"""Probe the public OA readiness endpoint and drive a debounced alert."""
 
 from __future__ import annotations
 
@@ -14,6 +14,7 @@ import stat
 import subprocess
 import sys
 import tempfile
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -23,8 +24,44 @@ from pathlib import Path
 SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
 SAFE_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
 MAX_RESPONSE_BYTES = 64 * 1024
+MAX_COUNTER = 1_000_000
+MAX_TIMESTAMP = (1 << 53) - 1
+STATE_SCHEMA_VERSION = 2
+RECOVERY_THRESHOLD = 2
+RECOVERY_WATCH_SECONDS = 60 * 60
+FLAPPING_STABLE_SECONDS = 30 * 60
 HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
 FAILURE_STATUS = "FAIL_APP_HEALTH"
+FLAPPING_STATUS = "FAIL_APP_HEALTH_FLAPPING"
+FAILURE_CATEGORIES = {
+    "dns_resolution_failed",
+    "tcp_connection_failed",
+    "tls_validation_failed",
+    "request_timeout",
+    "http_unavailable",
+    "response_contract_invalid",
+    "response_too_large",
+}
+LEGACY_FAILURE_CATEGORIES = FAILURE_CATEGORIES | {"dns_or_connect_failed"}
+PHASES = {
+    "healthy",
+    "failure_pending",
+    "alerted",
+    "recovery_pending",
+    "recovery_watch",
+    "flapping",
+}
+STATE_KEYS = {
+    "schemaVersion",
+    "phase",
+    "activeCategory",
+    "lastRecoveredCategory",
+    "consecutiveFailures",
+    "consecutiveSuccesses",
+    "recoveredAt",
+    "healthySince",
+    "flapTransitions",
+}
 
 
 class ProbeFailure(Exception):
@@ -89,9 +126,139 @@ def validate_ready_url(value: str) -> str:
     return value
 
 
+def new_probe_state() -> dict[str, object]:
+    return {
+        "schemaVersion": STATE_SCHEMA_VERSION,
+        "phase": "healthy",
+        "activeCategory": None,
+        "lastRecoveredCategory": None,
+        "consecutiveFailures": 0,
+        "consecutiveSuccesses": 0,
+        "recoveredAt": None,
+        "healthySince": None,
+        "flapTransitions": 0,
+    }
+
+
+def is_bounded_int(value: object, minimum: int = 0, maximum: int = MAX_COUNTER) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum
+
+
+def is_category(value: object, allow_none: bool = True) -> bool:
+    if value is None:
+        return allow_none
+    return isinstance(value, str) and value in LEGACY_FAILURE_CATEGORIES and bool(SAFE_TOKEN.fullmatch(value))
+
+
+def is_timestamp(value: object, allow_none: bool = True) -> bool:
+    if value is None:
+        return allow_none
+    return is_bounded_int(value, maximum=MAX_TIMESTAMP)
+
+
+def validate_v2_state(payload: dict[str, object]) -> bool:
+    if set(payload) != STATE_KEYS or payload.get("schemaVersion") != STATE_SCHEMA_VERSION:
+        return False
+    phase = payload.get("phase")
+    active = payload.get("activeCategory")
+    recovered = payload.get("lastRecoveredCategory")
+    failures = payload.get("consecutiveFailures")
+    successes = payload.get("consecutiveSuccesses")
+    recovered_at = payload.get("recoveredAt")
+    healthy_since = payload.get("healthySince")
+    transitions = payload.get("flapTransitions")
+    if (
+        phase not in PHASES
+        or not is_category(active)
+        or not is_category(recovered)
+        or not is_bounded_int(failures)
+        or not is_bounded_int(successes)
+        or not is_timestamp(recovered_at)
+        or not is_timestamp(healthy_since)
+        or not is_bounded_int(transitions)
+    ):
+        return False
+    if phase == "healthy":
+        return (
+            active is None
+            and recovered is None
+            and failures == 0
+            and successes == 0
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "failure_pending":
+        return (
+            active is not None
+            and failures >= 1
+            and successes == 0
+            and healthy_since is None
+            and transitions == 0
+            and ((recovered is None and recovered_at is None) or (recovered is not None and recovered_at is not None))
+        )
+    if phase == "alerted":
+        return (
+            active is not None
+            and failures >= 1
+            and successes == 0
+            and recovered is None
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "recovery_pending":
+        return (
+            active is not None
+            and failures == 0
+            and successes >= 1
+            and recovered is None
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "recovery_watch":
+        return (
+            active is None
+            and recovered is not None
+            and failures == 0
+            and successes == 0
+            and recovered_at is not None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase != "flapping" or active is None or active != recovered or recovered_at is None or transitions < 1:
+        return False
+    if healthy_since is None:
+        return failures >= 1 and successes == 0
+    return failures == 0 and successes >= 1
+
+
+def migrate_v1_state(payload: dict[str, object]) -> dict[str, object] | None:
+    if set(payload) != {"schemaVersion", "consecutiveFailures", "lastCategory"}:
+        return None
+    count = payload.get("consecutiveFailures")
+    category = payload.get("lastCategory")
+    if payload.get("schemaVersion") != 1 or not is_bounded_int(count) or not is_category(category):
+        return None
+    state = new_probe_state()
+    if count == 0:
+        return state if category is None else None
+    if category is None:
+        return None
+    state.update(
+        {
+            "phase": "failure_pending",
+            "activeCategory": category,
+            "consecutiveFailures": count,
+        }
+    )
+    return state
+
+
 def load_probe_state(path: Path) -> dict[str, object]:
     if not path.exists():
-        return {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None}
+        return new_probe_state()
     validate_regular_file(path, "local_state", private=True)
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
@@ -99,17 +266,14 @@ def load_probe_state(path: Path) -> dict[str, object]:
         fail("local_state_invalid")
     if not isinstance(payload, dict):
         fail("local_state_invalid")
-    count = payload.get("consecutiveFailures")
-    category = payload.get("lastCategory")
-    if (
-        payload.get("schemaVersion") != 1
-        or not isinstance(count, int)
-        or isinstance(count, bool)
-        or count < 0
-        or (category is not None and (not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category)))
-    ):
+    if payload.get("schemaVersion") == STATE_SCHEMA_VERSION:
+        if not validate_v2_state(payload):
+            fail("local_state_invalid")
+        return payload
+    migrated = migrate_v1_state(payload)
+    if migrated is None:
         fail("local_state_invalid")
-    return payload
+    return migrated
 
 
 def atomic_write(path: Path, payload: dict[str, object]) -> None:
@@ -128,13 +292,193 @@ def atomic_write(path: Path, payload: dict[str, object]) -> None:
         raise
 
 
-def classify_url_error(error: urllib.error.URLError) -> str:
-    reason = error.reason
+def capped(value: int, maximum: int = MAX_COUNTER) -> int:
+    return min(value, maximum)
+
+
+def elapsed(now: int, started_at: object) -> int:
+    if not isinstance(started_at, int) or isinstance(started_at, bool):
+        return 0
+    return max(0, now - started_at)
+
+
+def transition_probe_state(
+    state: dict[str, object],
+    category: str | None,
+    now: int,
+    failure_threshold: int,
+) -> tuple[dict[str, object], tuple[str, str] | None]:
+    """Return the next redacted state and an optional notification candidate."""
+    if not validate_v2_state(state) or not is_timestamp(now, allow_none=False):
+        raise ValueError("invalid transition input")
+    if category is not None and category not in FAILURE_CATEGORIES:
+        raise ValueError("invalid transition category")
+    phase = str(state["phase"])
+
+    if category is None:
+        if phase == "healthy":
+            return state.copy(), None
+        if phase == "failure_pending":
+            if state["recoveredAt"] is not None:
+                next_state = state.copy()
+                next_state.update(
+                    {
+                        "phase": "recovery_watch",
+                        "activeCategory": None,
+                        "consecutiveFailures": 0,
+                    }
+                )
+                return next_state, None
+            return new_probe_state(), None
+        if phase == "alerted":
+            next_state = state.copy()
+            next_state.update(
+                {
+                    "phase": "recovery_pending",
+                    "consecutiveFailures": 0,
+                    "consecutiveSuccesses": 1,
+                }
+            )
+            return next_state, None
+        if phase == "recovery_pending":
+            successes = capped(int(state["consecutiveSuccesses"]) + 1)
+            if successes < RECOVERY_THRESHOLD:
+                next_state = state.copy()
+                next_state["consecutiveSuccesses"] = successes
+                return next_state, None
+            recovered_category = str(state["activeCategory"])
+            next_state = new_probe_state()
+            next_state.update(
+                {
+                    "phase": "recovery_watch",
+                    "lastRecoveredCategory": recovered_category,
+                    "recoveredAt": now,
+                }
+            )
+            return next_state, (HEALTHY_STATUS, "public_ready_healthy")
+        if phase == "recovery_watch":
+            if elapsed(now, state["recoveredAt"]) >= RECOVERY_WATCH_SECONDS:
+                return new_probe_state(), None
+            return state.copy(), (HEALTHY_STATUS, "public_ready_healthy")
+        if phase == "flapping":
+            next_state = state.copy()
+            if state["healthySince"] is None:
+                next_state["healthySince"] = now
+                next_state["consecutiveFailures"] = 0
+                next_state["consecutiveSuccesses"] = 1
+                next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
+            else:
+                next_state["consecutiveSuccesses"] = capped(int(state["consecutiveSuccesses"]) + 1)
+            if elapsed(now, next_state["healthySince"]) >= FLAPPING_STABLE_SECONDS:
+                return new_probe_state(), None
+            return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
+
+    if phase == "flapping":
+        next_state = state.copy()
+        if state["healthySince"] is not None:
+            next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
+        next_state.update(
+            {
+                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
+                "consecutiveSuccesses": 0,
+                "healthySince": None,
+            }
+        )
+        return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
+
+    if phase == "alerted":
+        next_state = state.copy()
+        next_state.update(
+            {
+                "activeCategory": category,
+                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
+                "consecutiveSuccesses": 0,
+            }
+        )
+        return next_state, (FAILURE_STATUS, category)
+
+    if phase == "recovery_pending":
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "alerted",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+            }
+        )
+        return next_state, (FAILURE_STATUS, category)
+
+    if phase == "recovery_watch":
+        within_watch = elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "failure_pending",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+                "lastRecoveredCategory": state["lastRecoveredCategory"] if within_watch else None,
+                "recoveredAt": state["recoveredAt"] if within_watch else None,
+            }
+        )
+        return next_state, None
+
+    if phase == "healthy":
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "failure_pending",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+            }
+        )
+        return next_state, None
+
+    failures = capped(int(state["consecutiveFailures"]) + 1)
+    if failures < failure_threshold:
+        next_state = state.copy()
+        next_state.update({"activeCategory": category, "consecutiveFailures": failures})
+        return next_state, None
+    same_recent_failure = (
+        state["lastRecoveredCategory"] == category
+        and state["recoveredAt"] is not None
+        and elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
+    )
+    if same_recent_failure:
+        next_state = state.copy()
+        next_state.update(
+            {
+                "phase": "flapping",
+                "activeCategory": category,
+                "consecutiveFailures": failures,
+                "consecutiveSuccesses": 0,
+                "healthySince": None,
+                "flapTransitions": 1,
+            }
+        )
+        return next_state, (FLAPPING_STATUS, category)
+    next_state = new_probe_state()
+    next_state.update(
+        {
+            "phase": "alerted",
+            "activeCategory": category,
+            "consecutiveFailures": failures,
+        }
+    )
+    return next_state, (FAILURE_STATUS, category)
+
+
+def classify_connection_error(reason: object) -> str:
     if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
         return "tls_validation_failed"
     if isinstance(reason, (socket.timeout, TimeoutError)):
         return "request_timeout"
-    return "dns_or_connect_failed"
+    if isinstance(reason, socket.gaierror):
+        return "dns_resolution_failed"
+    return "tcp_connection_failed"
+
+
+def classify_url_error(error: urllib.error.URLError) -> str:
+    return classify_connection_error(error.reason)
 
 
 def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
@@ -166,8 +510,10 @@ def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
         raise ProbeFailure("request_timeout") from None
     except (ssl.SSLError, ssl.CertificateError):
         raise ProbeFailure("tls_validation_failed") from None
+    except socket.gaierror:
+        raise ProbeFailure("dns_resolution_failed") from None
     except (http.client.HTTPException, ConnectionError, OSError):
-        raise ProbeFailure("dns_or_connect_failed") from None
+        raise ProbeFailure("tcp_connection_failed") from None
     if len(body) > MAX_RESPONSE_BYTES:
         raise ProbeFailure("response_too_large")
     try:
@@ -179,7 +525,7 @@ def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
 
 
 def write_summary(path: Path, status: str, category: str) -> None:
-    finding = status == FAILURE_STATUS
+    finding = status in {FAILURE_STATUS, FLAPPING_STATUS}
     payload = {
         "issueId": "APP_HEALTH",
         "status": status,
@@ -217,6 +563,26 @@ def evaluate_alert(alert_helper: Path, webhook_file: Path, summary_file: Path, a
         fail("alert_evaluation_failed")
 
 
+def persist_transition(
+    state_file: Path,
+    summary_file: Path,
+    alert_state_file: Path,
+    state: dict[str, object],
+    candidate: tuple[str, str] | None,
+    alert_helper: Path,
+    webhook_file: Path,
+) -> None:
+    try:
+        atomic_write(state_file, state)
+        if candidate is not None:
+            status, category = candidate
+            write_summary(summary_file, status, category)
+    except OSError:
+        fail("local_state_invalid")
+    if candidate is not None:
+        evaluate_alert(alert_helper, webhook_file, summary_file, alert_state_file)
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--ready-url", required=True)
@@ -238,45 +604,46 @@ def main() -> None:
     summary_file = args.state_dir / "last-app-health.json"
     alert_state_file = args.state_dir / "app-health-alert-state.json"
     state = load_probe_state(probe_state_file)
+    now = int(time.time())
 
     try:
         probe(ready_url, args.timeout_seconds, args.ca_file)
     except ProbeFailure as error:
-        failures = min(int(state["consecutiveFailures"]) + 1, args.failure_threshold)
         try:
-            atomic_write(
-                probe_state_file,
-                {"schemaVersion": 1, "consecutiveFailures": failures, "lastCategory": error.category},
-            )
-        except OSError:
+            next_state, candidate = transition_probe_state(state, error.category, now, args.failure_threshold)
+        except ValueError:
             fail("local_state_invalid")
-        if failures >= args.failure_threshold:
-            try:
-                write_summary(summary_file, FAILURE_STATUS, error.category)
-            except OSError:
-                fail("local_state_invalid")
-            evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
-            print(
-                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=evaluated",
-                file=sys.stderr,
-            )
-        else:
-            print(
-                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=pending",
-                file=sys.stderr,
-            )
+        persist_transition(
+            probe_state_file,
+            summary_file,
+            alert_state_file,
+            next_state,
+            candidate,
+            args.alert_helper,
+            args.webhook_file,
+        )
+        alert_result = "evaluated" if candidate is not None else "pending"
+        print(
+            f"[app-health][ERROR] health_check_failed category={error.category} "
+            f"consecutive={next_state['consecutiveFailures']} alert={alert_result}",
+            file=sys.stderr,
+        )
         raise SystemExit(1)
 
     try:
-        atomic_write(
-            probe_state_file,
-            {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None},
-        )
-        write_summary(summary_file, HEALTHY_STATUS, "public_ready_healthy")
-    except OSError:
+        next_state, candidate = transition_probe_state(state, None, now, args.failure_threshold)
+    except ValueError:
         fail("local_state_invalid")
-    evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
-    print("[app-health][PASS] public readiness is healthy")
+    persist_transition(
+        probe_state_file,
+        summary_file,
+        alert_state_file,
+        next_state,
+        candidate,
+        args.alert_helper,
+        args.webhook_file,
+    )
+    print(f"[app-health][PASS] public readiness is healthy phase={next_state['phase']}")
 
 
 if __name__ == "__main__":
diff --git a/scripts/qa/run-app-health-monitor-contract.sh b/scripts/qa/run-app-health-monitor-contract.sh
index e717ab63..3dcd1481 100755
--- a/scripts/qa/run-app-health-monitor-contract.sh
+++ b/scripts/qa/run-app-health-monitor-contract.sh
@@ -137,6 +137,7 @@ MAIN_CAPTURE="$TEST_DIR/main-capture.jsonl"
 new_state_dir "$MAIN_STATE"
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "0" ]]
+[[ "$(stat -c %a "$MAIN_STATE/app-health-probe-state.json")" == "600" ]]
 pass "healthy baseline writes state without sending an alert"
 
 set_mode unavailable
@@ -155,14 +156,74 @@ pass "changed failure category sends one updated alert"
 
 set_mode healthy
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "2" ]]
+pass "first healthy sample keeps recovery pending"
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
-pass "recovery sends exactly one notification"
+pass "second healthy sample sends exactly one recovery notification"
+
+set_mode invalid-json
+expect_failure "first same-category recurrence stays pending" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
+expect_failure "second same-category recurrence enters flapping" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+expect_failure "continued flapping failure remains deduplicated" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+set_mode healthy
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
+import json, os, sys, time
+path = sys.argv[1]
+payload = json.load(open(path, encoding="utf-8"))
+assert payload["phase"] == "flapping", payload
+payload["healthySince"] = int(time.time()) - 1800
+with open(path, "w", encoding="utf-8") as handle:
+    json.dump(payload, handle)
+    handle.write("\n")
+os.chmod(path, 0o600)
+PY
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["phase"] == "healthy", payload
+assert payload["flapTransitions"] == 0, payload
+PY
+pass "flapping sends once and exits silently after thirty healthy minutes"
+
+DIFFERENT_STATE="$TEST_DIR/different-state"
+DIFFERENT_CAPTURE="$TEST_DIR/different-capture.jsonl"
+new_state_dir "$DIFFERENT_STATE"
+set_mode healthy
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+set_mode unavailable
+expect_failure "different-case first failure stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+expect_failure "different-case second failure alerts" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+set_mode healthy
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
+set_mode invalid-json
+expect_failure "new-category recurrence first sample stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
+expect_failure "new-category recurrence becomes a normal alert" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "3" ]]
+python3 - "$DIFFERENT_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["phase"] == "alerted", payload
+assert payload["activeCategory"] == "response_contract_invalid", payload
+PY
+pass "different failure category is not swallowed by same-category flapping debounce"
 
 TRANSIENT_STATE="$TEST_DIR/transient-state"
 TRANSIENT_CAPTURE="$TEST_DIR/transient-capture.jsonl"
 new_state_dir "$TRANSIENT_STATE"
+set_mode healthy
 probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null
 set_mode unavailable
 expect_failure "single transient outage is recorded" probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null 2>&1
@@ -184,10 +245,48 @@ expected = {
     "timeout": "request_timeout",
 }[sys.argv[2]]
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == expected, payload
+assert payload["activeCategory"] == expected, payload
 PY
 done
 
+python3 - "$ROOT_DIR/scripts/check-public-app-health.py" <<'PY'
+import importlib.util
+import socket
+import sys
+import urllib.error
+
+spec = importlib.util.spec_from_file_location("app_health", sys.argv[1])
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+assert module.classify_url_error(urllib.error.URLError(socket.gaierror(-2, "private-dns-text"))) == "dns_resolution_failed"
+assert module.classify_url_error(urllib.error.URLError(ConnectionRefusedError())) == "tcp_connection_failed"
+
+watch = module.new_probe_state()
+watch.update({
+    "phase": "recovery_watch",
+    "lastRecoveredCategory": "http_unavailable",
+    "recoveredAt": 1000,
+})
+pending, candidate = module.transition_probe_state(watch, "http_unavailable", 4599, 2)
+assert pending["phase"] == "failure_pending" and candidate is None, (pending, candidate)
+flapping, candidate = module.transition_probe_state(pending, "http_unavailable", 4599, 2)
+assert flapping["phase"] == "flapping", flapping
+assert candidate == ("FAIL_APP_HEALTH_FLAPPING", "http_unavailable"), candidate
+
+expired, candidate = module.transition_probe_state(watch, "http_unavailable", 4600, 2)
+assert expired["phase"] == "failure_pending" and expired["recoveredAt"] is None, expired
+alerted, candidate = module.transition_probe_state(expired, "http_unavailable", 4600, 2)
+assert alerted["phase"] == "alerted", alerted
+assert candidate == ("FAIL_APP_HEALTH", "http_unavailable"), candidate
+
+flapping["healthySince"] = 5000
+flapping["consecutiveFailures"] = 0
+flapping["consecutiveSuccesses"] = 1
+stable, candidate = module.transition_probe_state(flapping, None, 6800, 2)
+assert stable["phase"] == "healthy" and candidate is None, (stable, candidate)
+PY
+pass "DNS/TCP classification and exact thirty/sixty-minute boundaries are deterministic"
+
 UNUSED_PORT="$(python3 - <<'PY'
 import socket
 sock = socket.socket()
@@ -207,9 +306,9 @@ expect_failure "connection failure is rejected" env \
 python3 - "$CONNECT_STATE/app-health-probe-state.json" <<'PY'
 import json, sys
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == "dns_or_connect_failed", payload
+assert payload["activeCategory"] == "tcp_connection_failed", payload
 PY
-pass "connection failure is stored as a stable category"
+pass "TCP connection failure is stored separately from DNS resolution failure"
 
 TLS_STATE="$TEST_DIR/tls-state"
 new_state_dir "$TLS_STATE"
@@ -222,10 +321,39 @@ expect_failure "untrusted TLS certificate is rejected" env \
 python3 - "$TLS_STATE/app-health-probe-state.json" <<'PY'
 import json, sys
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == "tls_validation_failed", payload
+assert payload["activeCategory"] == "tls_validation_failed", payload
 PY
 pass "TLS failure is stored as a stable category"
 
+MIGRATION_STATE="$TEST_DIR/migration-state"
+MIGRATION_CAPTURE="$TEST_DIR/migration-capture.jsonl"
+new_state_dir "$MIGRATION_STATE"
+printf '%s\n' '{"schemaVersion":1,"consecutiveFailures":0,"lastCategory":null}' \
+  >"$MIGRATION_STATE/app-health-probe-state.json"
+chmod 0600 "$MIGRATION_STATE/app-health-probe-state.json"
+set_mode healthy
+probe "$MIGRATION_STATE" "$MIGRATION_CAPTURE" >/dev/null
+python3 - "$MIGRATION_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["schemaVersion"] == 2, payload
+assert payload["phase"] == "healthy", payload
+PY
+[[ "$(capture_lines "$MIGRATION_CAPTURE")" == "0" ]]
+pass "valid legacy state migrates without backfilling a notification"
+
+INVALID_SCHEMA_STATE="$TEST_DIR/invalid-schema-state"
+new_state_dir "$INVALID_SCHEMA_STATE"
+printf '%s\n' '{"schemaVersion":2,"phase":"healthy","activeCategory":null,"lastRecoveredCategory":null,"consecutiveFailures":1000001,"consecutiveSuccesses":0,"recoveredAt":null,"healthySince":null,"flapTransitions":0}' \
+  >"$INVALID_SCHEMA_STATE/app-health-probe-state.json"
+chmod 0600 "$INVALID_SCHEMA_STATE/app-health-probe-state.json"
+set_mode healthy
+expect_failure "out-of-range v2 state fails closed" probe "$INVALID_SCHEMA_STATE" \
+  "$TEST_DIR/invalid-schema-capture.jsonl" >"$TEST_DIR/invalid-schema-stdout" 2>"$TEST_DIR/invalid-schema-stderr"
+grep -Fxq '[app-health][ERROR] local_state_invalid' "$TEST_DIR/invalid-schema-stderr"
+! grep -Fq '1000001' "$TEST_DIR/invalid-schema-stderr"
+pass "v2 state schema rejects inconsistent or out-of-range fields without disclosure"
+
 CORRUPT_STATE="$TEST_DIR/corrupt-state"
 new_state_dir "$CORRUPT_STATE"
 printf '%s\n' 'private-payload-must-not-appear' >"$CORRUPT_STATE/app-health-probe-state.json"
@@ -241,15 +369,19 @@ pass "local state error output is stable and redacted"
 python3 - "$MAIN_CAPTURE" <<'PY'
 import json, sys
 messages = [json.loads(line)["text"]["content"] for line in open(sys.argv[1], encoding="utf-8")]
-assert len(messages) == 3, messages
+assert len(messages) == 4, messages
 assert "FAIL_APP_HEALTH" in messages[0]
 assert "http_unavailable" in messages[0]
 assert "response_contract_invalid" in messages[1]
 assert "PASS_APP_HEALTH_READY" in messages[2]
+assert "FAIL_APP_HEALTH_FLAPPING" in messages[3]
+assert messages[0].startswith("OA 公网健康告警\n"), messages[0]
+assert messages[2].startswith("OA 公网健康恢复\n"), messages[2]
+assert messages[3].startswith("OA 公网健康链路反复抖动\n"), messages[3]
 for forbidden in ("do-not-leak", "localhost", "https://", "/tmp/", "not-json-private-payload"):
     assert all(forbidden not in message for message in messages), (forbidden, messages)
 PY
-pass "captured alerts contain only stable status and category text"
+pass "captured application alerts use explicit source titles and stable redacted categories"
 
 expect_failure "plain HTTP readiness URL is rejected" env \
   HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$TEST_DIR/http-capture.jsonl" \
diff --git a/scripts/qa/run-automatic-backup-contract.sh b/scripts/qa/run-automatic-backup-contract.sh
index 71c49a3b..c7dbf660 100755
--- a/scripts/qa/run-automatic-backup-contract.sh
+++ b/scripts/qa/run-automatic-backup-contract.sh
@@ -701,6 +701,16 @@ mv "$TEST_DIR/deadman-withheld-batch" "$DEADMAN_LAYOUT/archive/$SECOND_BATCH"
   --webhook-file "$TEST_DIR/wecom-webhook" --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" >/dev/null
 [[ "$(wc -l <"$DEADMAN_CAPTURE")" == "2" ]]
 pass "ARM dead-man sends one recovery notification after finalized archives resume"
+python3 - "$DEADMAN_CAPTURE" <<'PYTHON'
+import json
+import sys
+
+messages = [json.loads(line)["text"]["content"] for line in open(sys.argv[1], encoding="utf-8")]
+assert messages[0].startswith("内部 OA 运维告警\n"), messages
+assert messages[1].startswith("内部 OA 运维恢复通知\n"), messages
+assert all("OA 公网健康" not in message for message in messages), messages
+PYTHON
+pass "shared helper keeps dead-man alert and recovery titles unchanged"
 
 DRIFT_BUILD_SHA="0123456789abcdef0123456789abcdef01234567"
 expect_failure "wrapper rejects an active runtime build that drifted from its installed contract" \
diff --git a/scripts/send-wecom-group-alert.py b/scripts/send-wecom-group-alert.py
index 49ded119..e0508499 100755
--- a/scripts/send-wecom-group-alert.py
+++ b/scripts/send-wecom-group-alert.py
@@ -18,6 +18,7 @@ from pathlib import Path
 SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
 ALLOWED_STATUSES = {
     "FAIL_APP_HEALTH",
+    "FAIL_APP_HEALTH_FLAPPING",
     "FAIL_AUTOMATIC_BACKUP",
     "FAIL_OFFHOST_DEADMAN",
     "PASS_APP_HEALTH_READY",
@@ -131,7 +132,13 @@ def should_send(previous: dict[str, object], status: str, category: str) -> bool
 
 
 def safe_text(status: str, category: str) -> str:
-    if status.startswith("FAIL_"):
+    if status == "FAIL_APP_HEALTH":
+        headline = "OA 公网健康告警"
+    elif status == "FAIL_APP_HEALTH_FLAPPING":
+        headline = "OA 公网健康链路反复抖动"
+    elif status == "PASS_APP_HEALTH_READY":
+        headline = "OA 公网健康恢复"
+    elif status.startswith("FAIL_"):
         headline = "内部 OA 运维告警"
     elif status == "PASS_WITH_FINDINGS":
         headline = "内部 OA 运维提醒"