运维(08-28-production-arm-backup-transfer-incident): 固化只读调查合同

.trellis/spec/hyb-oa/deployment-trialrun.md

@@ -2865,6 +2865,37 @@ current shared mapping -> fine-grained stages + mutation listeners
 - Read-only investigation output is an allowlist of status/category/count/age
   and capacity fields. Never print unit command lines, raw JSON, raw journal,
   remote identities, paths, run ids, webhook data or diagnostic stderr.
+- Before classifying an off-host transport failure, validate the newest source
+  batch with the same manifest schema/status, exact three-file set, run-id
+  pairing, byte-count and SHA-256 contract used by the receiver. Report only
+  whether a complete local recovery point exists, its age/count/aggregate
+  bytes and one of `production_single_point` or `no_proven_recovery_point`.
+  Off-host staleness alone does not distinguish those two materially different
+  production risks.
+- Inspect cleanup semantics as executable ordering, not a remembered claim.
+  Source publication must precede transfer; local pruning must follow verified
+  off-host finalize and success-state publication; generation staging must be
+  protected by an exit cleanup. Also report current valid/invalid/staging
+  counts so repeated transfer failures cannot silently consume source space.
+- A retained `last-error.log` may provide an rsync code only when the matching
+  safe run summary says `phase_offhost_transfer_failed`. Never parse a log from
+  another phase or substitute service exit `1`. If the wrapper did not persist
+  transfer start/end clocks, emit `NOT_AVAILABLE`; service duration, mtime and
+  conversation time are not transfer-duration evidence.
+- A read-only inspector must prevalidate every output as a fixed key plus an
+  enum/integer value. Malformed manifests, unreadable files and unexpected
+  runtime exceptions collapse to stable categories without a traceback,
+  because tracebacks can expose absolute paths even when normal output is
+  redacted.
+- If existing read-only state cannot converge on a cause, stop expanding the
+  read surface. The next step is a separately approved, single-attempt
+  observational retransmit with frozen input, fixed telemetry, bounded timeout
+  and no automatic retry. It is a diagnostic write, not part of read-only
+  authority.
+- After a transport repair, one fresh atomic batch may unlock a blocked restore
+  drill, but incident closeout requires two consecutive natural schedules.
+  Each cycle records source and receiver capacity plus receiver partial count
+  and aggregate bytes; a manual run does not replace either natural sample.
 
 ### 4. Validation & Error Matrix
 
@@ -2886,6 +2917,15 @@ current shared mapping -> fine-grained stages + mutation listeners
 - Path cannot be traversed by the approved account -> `ACCESS_DENIED`; request
   separate read authority only if the unresolved field materially changes the
   decision.
+- Complete source batch valid + off-host point stale -> production has a local
+  recovery point but remains a single point; prefer the next safe natural run
+  unless a separately approved transition run is justified.
+- No source batch passes the complete contract -> `no_proven_recovery_point`;
+  raise the transition-recovery-point priority, but do not trigger a backup
+  until root cause, capacity and stop conditions are approved.
+- `last-error.log` contains rsync-like text but the safe last-run phase is not
+  `offhost_transfer` -> exit code remains `NOT_AVAILABLE` rather than attaching
+  stale diagnostics to the current failure.
 
 ### 5. Good/Base/Bad Cases
 
@@ -2915,6 +2955,13 @@ current shared mapping -> fine-grained stages + mutation listeners
   rsync codes `23/30/12/11/255`; assert the incident summary preserves code and
   elapsed-time evidence separately from wrapper exit `1` and does not output
   file names.
+- Fixture one complete source batch, no complete source batch, checksum/size
+  corruption and malformed manifest. Assert the risk classification and
+  transition recommendation change without emitting paths, run ids, artifact
+  names, raw diagnostics or tracebacks.
+- Mechanically mutate the wrapper ordering so pruning appears before transfer;
+  the cleanup/retention contract must turn false. A prose claim that failure
+  retains data is not sufficient regression evidence.
 - Sensitive scan the probe and assert raw `ExecStart`, journal, JSON, paths,
   remote identities and run ids cannot appear in stdout.
 - Every root probe must self-remove its fixed entry/sudoers and an independent
@@ -2928,6 +2975,8 @@ current shared mapping -> fine-grained stages + mutation listeners
 timer active -> backup healthy
 state path test returned false -> state missing
 partial count > 0 -> finalize failed
+off-host stale -> assume one complete backup still exists on production
+service exit 1 -> call it the rsync exit code and use service duration
 ```
 
 #### Correct
@@ -2939,4 +2988,6 @@ derive unit identity from installer -> timer trigger + service Result/exit
 -> partial proves initial SSH acceptance; compare byte offset + rsync code + duration
 -> dead-man decision + dedup state + sent flag
 -> safe scheduler/source/transfer/finalize/alert conclusion
+-> validate the source batch first and classify single-point vs no proven point
+-> bind retained diagnostics to the safe failure phase; unavailable clocks stay unavailable
 ```

.trellis/tasks/08-23-isolated-backup-restore-drill/task.json

@@ -21,6 +21,6 @@
   "children": [],
   "parent": null,
   "relatedFiles": [],
-  "notes": "",
+  "notes": "2026-08-28：恢复演练保持 in_progress/P2，但被 P1 operations 事件 08-28-production-arm-backup-transfer-incident 阻塞。该事件负责生产到 ARM 传输链修复、生产容量处置、告警响应计时和连续两个自然调度周期验证；在新原子批次形成并重新通过 24 小时身份门禁前，本任务不得继续实际恢复。既有恢复候选 SHA 仍有效，无需因事件任务重新渲染。",
   "meta": {}
-}
+}

.trellis/tasks/08-28-production-arm-backup-transfer-incident/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."}

.trellis/tasks/08-28-production-arm-backup-transfer-incident/design.md

@@ -0,0 +1,95 @@
+# 生产到 ARM 备份传输中断设计
+
+## 设计结论
+
+采用“先只读判因、再最小变更、最后两个自然周期验证”的事件处置路径。传输修复和容量处置属于同一个 P1 operations 任务，但每项远端写操作维持独立审批；恢复演练作为下游消费者保持阻塞，不参与本任务的根因修复。
+
+## 证据链与判定矩阵
+
+```text
+生产 daily timer 触发
+  -> 本地成对批次生成
+  -> rsync 建立 SSH/receiver 会话
+  -> ARM inbox/<batch>.partial 写入
+  -> rsync 完成
+  -> receiver finalize 原子发布
+  -> ARM dead-man + 生产 watchdog 判定新鲜
+```
+
+当前证据已经到达 `.partial 写入`，所以初始 SSH 认证、host key 与 receiver 授权不是首要分支。后续只读调查按以下矩阵裁决：
+
+| 证据 | 主要指向 | 保留边界 |
+| --- | --- | --- |
+| 两个 partial 停止偏移接近，失败时长也接近 | 确定性大小/读取/会话上限 | 结合 rsync 原始码确认，不直接改超时 |
+| 停止偏移差异明显，时长不收敛 | 网络或间歇超时 | 读取客户端/服务端安全状态，不抓原始 payload |
+| rsync `23` | 部分传输/源或目标文件错误 | 结合私有阶段类别，不输出 stderr |
+| rsync `30` | 超时 | 定位是哪一层超时，禁止无限放宽 |
+| rsync `12` | 协议/数据流错误 | 核对双方版本和会话中断证据 |
+| rsync `11` | 文件 I/O | 联查源读取、目标写入和容量 |
+| SSH/包装层 `255` 且已有 partial | 会话在启动后断开 | 不误判为初始认证失败 |
+| partial 满足完整三文件/manifest/字节合同 | finalize 未完成 | 当前现场为 0，不作为默认假设 |
+
+## 只读采集合同
+
+- 生产侧首先输出最新本地候选是否通过完整 manifest/三文件/字节/SHA-256 合同、年龄、完整批次数量和聚合大小；不输出 run id、文件名或路径。该结果先于传输根因分析，决定当前是“生产单点”还是“无已证明可用备份”。
+- 对成功与失败路径分别输出本地完整批次、生成期临时对象和保留策略的聚合结果，明确 `retained_after_failure`、`pruned_after_success` 和连续失败累积趋势；不从单次目录计数猜测脚本语义。
+- partial 只输出每个匿名样本的总字节数、预期总字节数、安全比例和年龄；不输出名称、路径、run id 或 artifact 名称。
+- 生产私有诊断只输出固定的 `transfer_exit_code`、`failure_category`、开始/结束 epoch 和持续秒数。若现有状态没有持久化真实 rsync 码或开始时刻，记录 `NOT_AVAILABLE`，需要本地设计可观测性修订并另行批准部署，不能现场改脚本。
+- 容量只输出 filesystem 使用率/可用 GiB、完整/不完整/临时批次数量和聚合字节，不列目录。
+- 采集器先探测账号能力；无 sudo 时不得尝试临时扩权。需要 root-owned 状态时先提交固定 allowlist 入口设计并单独审批。
+
+若上述静态/历史只读证据仍无法收敛，停止追加远端读取项。下一候选步骤是一次单独审批的受控观测性重传：冻结一个已校验本地批次，只允许一次传输尝试，实时记录固定阶段、rsync/SSH 原始码、开始/结束 epoch 和聚合字节进度，设置有界超时且失败不重试。它是诊断性写操作，不属于第一阶段只读授权。
+
+## 容量预算
+
+生产修复前冻结以下聚合量：最近成功完整本地批次大小、一次备份生成期间的峰值临时增量、当前保留批次总量和文件系统可用量。最低通过线为：
+
+```text
+available_before_schedule >= observed_full_batch_peak * 2
+```
+
+其中 `observed_full_batch_peak` 包含 database dump、storage archive、manifest 和生成期临时开销；两倍口径提供一个新批次的实际需求和等量安全余量。若现场无法可靠测得生成期峰值，则使用“最近最大完整批次聚合大小 × 2 + 已知最大临时开销”的更保守口径。任何降低该门槛的决定须重新审阅 PRD；扩容优先于删除唯一恢复证据。
+
+## 变更与回滚边界
+
+- 根因明确后只设计一个最小变更集，例如有界传输超时、具体 I/O 修复、网络会话修复或容量扩展；不同根因不打包试错上线。
+- 修改 systemd/SSH/rsync 配置前保存脱敏旧值指纹和固定回滚命令；不输出配置正文。
+- 清理自动备份自有对象时必须先以固定模式、文件类型、所有者、时间和引用关系验证身份，逐个命中；不使用宽泛递归清理。
+- 手工补跑仅用于经单独批准的修复后快速验证，不计入两个自然周期；失败时保留新的脱敏 finding，不连续重试。
+
+## 告警响应计时
+
+从本任务开始使用两个独立 epoch：
+
+- `alert_sent_epoch`：helper 返回 `sent=true` 所在调用时刻；
+- `human_ack_epoch`：指定责任人首次明确确认收到并开始处置的时刻。
+
+`response_seconds = human_ack_epoch - alert_sent_epoch`。缺任一锚点即 `NOT_MEASURED`，不得用文件 mtime、对话时间或事后回忆替代。该指标只评价响应闭环，不改变 dead-man 的检测正确性。
+
+## 验收状态机
+
+```text
+PLANNING
+ -> READ_ONLY_DIAGNOSIS_APPROVED
+ -> ROOT_CAUSE_EVIDENCED
+ -> CHANGE_PLAN_APPROVED
+ -> FIX_APPLIED
+ -> FIRST_ATOMIC_BATCH_PASS
+ -> NATURAL_CYCLE_1_PASS
+ -> NATURAL_CYCLE_2_PASS
+ -> INCIDENT_GO
+```
+
+任何一步失败都停在当前状态并保留证据。`FIRST_ATOMIC_BATCH_PASS` 形成一个满足 24 小时门禁的新鲜原子批次后，即可解除对恢复演练的事件阻塞；恢复演练仍需自己的身份探针和实际恢复审批。P1 事件本身只有到 `NATURAL_CYCLE_2_PASS` 才能结案，两条线不互相等待。
+
+## 过渡恢复点决策
+
+- `complete_local_recovery_point=yes`：若下个自然周期在可接受窗口内且容量预算允许，默认 `wait_natural`；仍可基于风险窗口单独申请手工补跑，但不是自动动作。
+- `complete_local_recovery_point=no`：事件升级为当前无已证明可用备份，默认建议 `request_manual_run`，但必须先保证根因不会使补跑无界失败、容量满足停止条件，并单独审批。
+- 任何手工补跑都只建立过渡恢复点和验证修复，不替代两个自然周期。
+
+## 安全与脱敏
+
+- 腾讯云生产和 ARM 地址、账号、绝对现场路径、完整 run id/digest、密钥、known_hosts、webhook、原始日志和业务内容不进入仓库或聊天。
+- 远端临时探针若获批，固定目标、SHA-256、语法、sudoers 和自清理均需机械校验，并由外层独立报告入口/暂存残留。
+- timer `active`、partial 存在、单次备份成功或告警 `sent=true` 都是局部证据，不能扩写为链路整体正常。

.trellis/tasks/08-28-production-arm-backup-transfer-incident/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."}

.trellis/tasks/08-28-production-arm-backup-transfer-incident/implement.md

@@ -0,0 +1,72 @@
+# 生产到 ARM 备份传输中断实施计划
+
+## Phase 0：规划与授权
+
+- [x] 创建 P1、`operations` Trellis task，记录来源事件和与恢复演练的阻塞关系。
+- [x] 固定只读调查、生产/ARM 写操作、手工补跑和测试告警为不同审批边界。
+- [x] 用户审阅并批准本规划及补充项后运行 `task.py start`；该批准只授权进入本地实施，不自动授权连接远端。
+
+## Phase 1：本地只读调查准备
+
+- [x] 从 `scripts/run-automatic-backup.sh`、receiver、dead-man 和 systemd installer 提取真实单位名、状态路径来源、阶段和退出码合同，禁止手写近似常量。
+- [x] 实现只输出 allowlist 的跟踪只读采集器：生产本地完整恢复点合同、本地成功/失败清理保留语义、partial 匿名字节比例、rsync 原始码、无法取得的传输时钟、两端容量；timer/service 继续复用既有只读 systemd 聚合合同，不在 Python 中另写一套。
+- [x] 为本地完整/无完整批次、成功清理/失败保留、相近/不同停止偏移、rsync `23/30/12/11/255`、缺失持续时间、权限拒绝、畸形 manifest 和敏感输出补合成回归。
+- [ ] 本地语法、合同、敏感扫描全部通过后，提交远端只读调查摘要与精确权限边界，单独申请执行批准。
+
+### 本地准备结果（2026-08-28）
+
+- 跟踪调查器：`scripts/qa/inspect-backup-transfer-incident.py`；仅接受显式绝对路径参数，不回显参数值，未知异常也只输出稳定类别。
+- 可复跑合同：`scripts/qa/run-backup-transfer-incident-contract.sh`；17 项合成检查通过，fixture 与 Python cache 零残留；权限拒绝不会被误写成“无有效备份”，其他阶段的旧日志不会冒充本轮 rsync 诊断。
+- 既有 `run-automatic-backup-contract.sh` 74 项通过；生产 wrapper 未改。现状没有持久化传输专属开始/结束时刻，故持续时间必须为 `NOT_AVAILABLE`。
+- rsync 原始码只在安全摘要明确为 `phase_offhost_transfer_failed` 时从 root 私有 `last-error.log` 固定正则提取，避免把 paired-backup/finalize 的旧诊断错误归因给传输。
+
+## Phase 2：获批的远端只读调查
+
+- [ ] 在不修改生产/ARM 的前提下采集四类证据；每个临时入口在同轮精确清理并由外层独立复核零残留。
+- [ ] 先判定“生产单点”或“无已证明可用备份”，再交叉判定主要根因、次要候选、清理/累积语义和容量预算；无法取得的字段标为 `NOT_AVAILABLE/NOT_MEASURED`。
+- [ ] 记录过渡恢复点决策；只读证据无法收敛时停止扩读，提交一次受控观测性重传的独立审批方案，不直接执行。
+- [ ] 更新本 task 证据，给出单一最小修复方案、回滚方案和生产写操作清单；停止并请求用户审阅。
+
+## Phase 3：本地修复与变更审批
+
+- [ ] 仅修改证据支持的脚本/installer/合同；不轮换已通过初始认证的密钥，不放宽 manifest/finalize 门禁。
+- [ ] 容量方案优先扩容或精确处置已确认的自动备份自有对象，证明不会触碰唯一恢复点、数据库或附件源。
+- [ ] 运行自动异机备份、receiver、dead-man、故障注入、回滚和敏感输出合同。
+- [ ] 列出生产/ARM 的每个写目标、旧值指纹、候选 SHA、执行/回滚命令和停止条件，单独申请实际变更批准。
+
+## Phase 4：获批的实际修复
+
+- [ ] 执行最小变更和容量处置；不做未批准的手工备份、timer 修改、partial 清理或测试告警。
+- [ ] 若另获手工补跑批准，只运行一次并记录真实 rsync 码、时长、原子 finalize 和残留；失败不自动重试。
+- [ ] 验证至少一个新原子批次、manifest/三文件/字节/SHA-256、三组 service exit `0` 和容量预算。
+
+## Phase 5：两个自然周期观察
+
+- [ ] 自然周期 1：daily 新原子批次 PASS；watchdog/dead-man service success；记录生产/ARM 容量、ARM partial 数量/聚合字节趋势和适用的告警响应锚点。
+- [ ] 自然周期 2：重复同一组断言；不以手工补跑或旧批次替代；ARM partial/容量趋势不得继续恶化。
+- [ ] 两个周期均通过后标记事件 `GO`，否则保留失败并返回 Phase 2/3，不临时放宽阈值。
+
+## Phase 6：解除阻塞与收口
+
+- [ ] 在 `08-23-isolated-backup-restore-drill` 追加本任务结果交叉引用并解除事件阻塞；恢复演练仍需新批次身份探针和单独实际恢复审批。
+- [ ] 运行 task validate、Trellis governance、process baseline、docs registry、敏感扫描和 `git diff --check`。
+- [ ] 中文提交引用 `08-28-production-arm-backup-transfer-incident`；不自动 push，不归档未满足两个自然周期的任务。
+
+## 停止条件
+
+- 只读操作将输出路径、身份、凭据、原始日志/JSON 或业务内容。
+- 需要新增 sudo、读取超出 allowlist 的 root 状态或改变生产/ARM 状态但尚未获独立批准。
+- 清理对象、配置目标、旧值或回滚路径不唯一。
+- 新原子批次不完整、service 非零、容量低于预算或任一自然周期失败。
+- 告警发送/人工确认锚点缺失时不得补造响应延迟。
+
+## 验证命令
+
+- `bash -n` / `python3 -m py_compile` 覆盖新增或修改的运维脚本。
+- 运行自动备份、receiver、dead-man 和相关敏感输出合同。
+- `python3 ./.trellis/scripts/task.py validate .trellis/tasks/08-28-production-arm-backup-transfer-incident`
+- `node scripts/qa/check-trellis-governance.mjs`
+- `node scripts/qa/check-process-contract.mjs --baseline --json`
+- `node scripts/qa/audit-docs.mjs --check-registry`
+- `node scripts/qa/check-sensitive-content.mjs`
+- `git diff --check`

.trellis/tasks/08-28-production-arm-backup-transfer-incident/prd.md

@@ -0,0 +1,113 @@
+# 生产到 ARM 备份传输中断
+
+## Goal
+
+恢复腾讯云生产到 ARM 的自动异机备份链，消除持续扩大的异机恢复点缺口，同时把生产备份落盘容量恢复到可持续范围，并证明修复不是一次偶然成功。
+
+## Scope
+
+- 只读定位 partial 停止偏移、rsync 原始退出码、失败持续时间、生产本地批次累积与容量预算。
+- 基于证据设计并在另行批准后实施最小传输链修复和生产容量处置。
+- 验证新 ARM 原子归档、daily/watchdog/dead-man 的 timer 与 service，并连续观察两个自然调度周期。
+- 建立真实告警发送调用时刻、人工确认时刻和响应延迟的脱敏记录合同。
+- 维护与 `08-23-isolated-backup-restore-drill` 的阻塞交叉引用；不在本任务内执行恢复演练。
+
+## Background
+
+- 腾讯云服务器是真实 OA 生产；ARM 是异机备份、dead-man 和隔离恢复节点，不是生产。
+- 2026-08-27，任务 `08-23-isolated-backup-restore-drill` 的最新批次探针发现 ARM 最新原子归档已超过 24 小时目标；随后获批的只读调查确认生产 daily/watchdog timer 正常触发，但对应 service 最近均以 `exit-code/1` 失败。
+- ARM inbox 有两个未完成 `.partial`，其中最新一个与最近自然备份时点对应；两者均未满足三文件、manifest 身份和产物字节数的完整传输合同，正式归档事务临时对象为零。因此当前 finding 为 `offhost_transfer_incomplete`，不是调度漏跑或完整传输后的 finalize 失败。
+- partial 已证明 rsync 通过初始 SSH 认证、host key 校验和 receiver 授权并开始写入。密钥失效、`authorized_keys` 初始拒绝和 host key 初始拒绝不是首要假设；仍需调查传输中段的源端读取、网络/超时、目标写入和会话/传输限制。
+- ARM dead-man 以显式 `93600` 秒阈值正确判定 `archive_stale`，且 journal 聚合存在一次 `sent=true`。检测与发送调用没有被去抖静默抑制，但该告警没有形成及时人工响应；本次首次知晓时刻未采集，历史响应延迟不可补造。
+- 生产相关文件系统只读观测约为 82% 使用、10 GiB 可用。现有证据不能把本次传输中断直接归因于空间不足，但该余量不足以作为持续生成完整 dump、storage 归档和失败重试的安全基线，必须与传输故障并行处置。
+
+## Requirements
+
+### R1. 分阶段授权与数据保护
+
+- 任务创建和规划不授权连接生产或 ARM，也不授权任何远端写入。
+- 第一阶段只读调查必须另行批准，只允许读取聚合字节数、rsync 原始退出码、固定阶段时刻/耗时、systemd 状态、源端本地批次聚合和文件系统容量；不得输出远端地址、完整路径、run id、文件名、原始 stderr/journal/JSON、凭据或业务内容。
+- 清理空间、删除或移动本地/ARM partial、调整 SSH/rsync/systemd 参数、修改 timer、重启服务、手工触发备份、重跑传输或发送测试告警均属于生产/ARM 写操作，必须在根因和变更方案审阅后单独明确批准。
+- 不修改或删除已有 ARM 正式原子归档；诊断不得把真实备份下载到开发机。
+
+### R2. 只读根因定位
+
+- 第一项先判断生产本地是否至少存在一个完整可用批次，并对最新本地候选执行与异机归档相同的 manifest schema/status、精确三文件集合、run id 配对、字节数和 SHA-256 校验。输出只含 `complete_local_recovery_point=yes|no`、年龄、数量和聚合大小。若为 `yes`，当前风险是异机副本缺失/生产单点；若为 `no`，当前不存在任何已证明可用的备份，事件风险和过渡恢复点优先级立即升级。
+- 分别核对成功运行和失败运行对生产本地批次/临时对象的保留与清理行为：成功后是否按保留策略清理，失败后是否保留完整本地批次、是否残留生成期临时对象，以及连续失败是否逐轮累积。不得仅统计“是否累积”而遗漏清理语义。
+- 比较两个 partial 的聚合总字节数、各预期产物已写入字节数或安全比例，不输出文件名/路径。相同或接近停止偏移作为确定性限制信号，明显不同偏移作为网络/超时等间歇中断信号；阈值和计算口径须在取数前固定。
+- 从生产备份 wrapper 的私有状态/诊断中提取真实 rsync 退出码和稳定类别，不以 systemd/service 的通用 exit `1` 替代。至少区分 rsync `23`、`30`、`12`、`11` 与 SSH 层 `255`；不得输出原始错误文本。
+- 记录每次失败的传输阶段开始、失败时刻和持续秒数，判断是否收敛到相近时长；无法从现有状态可靠取得时明确标为未测，不从 mtime 或聊天时间猜测。
+- 只读统计生产当前可用空间、最近完整本地批次聚合大小、当前本地完整/不完整/临时批次数量及成功/失败路径的清理与累积行为，形成容量预算。
+- 将 partial 偏移、rsync 原始码、失败时长和两端容量交叉判定，输出一个主要根因类别及仍未排除的候选，不为尽快修复而伪造单一确定根因。
+- 若现有只读证据不能收敛，尤其是故障疑似位于不可控链路侧，不扩大远端读取范围或反复追加低信号探针；转为设计“一次受控观测性重传”，冻结输入、实时采集允许字段并有界停止，作为独立审批项。该路径不在只读调查授权内。
+
+### R3. 传输链修复
+
+- 只修复 R2 证据直接支持的最小故障面；不得顺手轮换已证明可以完成初始认证的密钥、重建 receiver、放宽 manifest/finalize 安全门禁或修改恢复 runner。
+- 若根因为超时/会话限制，变更须有明确的客户端、服务端或中间链路归属、取值依据、回滚值和有界失败行为；禁止无限超时或无界重试。
+- 若根因为源端读取、目标写入、网络或 rsync 行为，优先修复可机械验证的具体错误，并保留原子 finalize、partial 隔离和失败关闭。
+- 生产和 ARM 的实际变更必须先形成脱敏变更清单、备份/回滚点与验证命令，再单独申请执行批准。
+
+### R4. 生产容量处置
+
+- 基于最近成功批次和当前生成流程计算至少容纳一个完整新批次的实际峰值预算，并增加保守安全余量；容量阈值必须在生产写操作前写入设计和证据。
+- 优先识别可安全处置的自动备份自有产物、失败临时对象或保留策略问题；任何删除/移动必须精确匹配已验证身份，禁止宽泛清理、跨任务清理或触碰生产数据库/附件源数据。
+- 容量处置不得以降低备份保留、删除唯一恢复点或跳过本地完整批次为代价。若空间扩容比删除更安全，应选择扩容方案并单独审批。
+- 修复后生产必须具备“一个完整新批次实际峰值 + 安全余量”的可验证空间，且两个自然周期内空间趋势不继续逼近失败线。
+
+### R5. 恢复验证与持续观察
+
+- 修复后必须形成新的 ARM 原子归档批次，而不是仅出现 partial；该批次须通过 manifest schema/status、精确三文件集合、run id/目录配对、字节数和 SHA-256 校验。
+- daily、watchdog、dead-man 不得仅以 timer `active` 作为通过。对应 timer 必须正常触发，最近 service 必须 `Result=success`、退出码 `0`，状态摘要须与新鲜归档一致。
+- 必须连续观察至少两个自然 daily 调度周期，每个周期都形成新的完整原子归档并通过相同合同；手工补跑成功不能替代两个自然周期。
+- 两个周期都必须记录 ARM inbox partial 数量/聚合字节变化和 ARM 文件系统可用量/使用率趋势；不得只观察生产容量，避免传输失败对象在 ARM 侧继续累积形成新的容量故障。
+- 观察期间不得为通过验收而删除失败证据、改选旧批次或临时放宽新鲜度/完整性阈值。
+
+### R5a. 过渡恢复点决策
+
+- R2 得出本地备份可用性后，必须单独记录 `transition_recovery_point_action=wait_natural|request_manual_run` 及理由。
+- 若生产本地已有完整可用批次且下一自然周期时间/容量风险可接受，优先等待自然调度；若生产本地也没有完整可用批次，手工补跑建立过渡恢复点的优先级显著上升，但仍须先有根因/容量停止条件并取得单独批准。
+- 手工补跑即使形成过渡恢复点，也不计入 AC5 的两个自然周期；若没有获批，不得为缩短风险窗口自行执行。
+
+### R6. 告警送达与响应闭环
+
+- 保留本次“检测与发送调用正常、响应未及时形成”的双重结论，不把 `sent=true` 等同于人员已阅读。
+- 从本任务开始，每次真实备份/dead-man 告警须分别记录脱敏的发送调用时刻和首位责任人确认时刻，并计算响应延迟；不得记录 webhook、群标识、人员账号或消息原文。
+- 若后续告警仍只有 `sent=true` 而无人确认，调查群送达、可见性、责任人和升级流程；发送测试告警须单独批准。
+
+### R7. 与恢复演练解耦
+
+- 本任务优先级为 P1、scope 为 `operations`，高于 P2 的 `08-23-isolated-backup-restore-drill`。
+- 演练任务保持 `in_progress`，明确被本任务阻塞；在新鲜原子批次形成并通过 24 小时身份门禁前不得继续实际恢复。
+- 既有恢复候选 SHA 保持有效，无需因本事件重新渲染；本任务不修改恢复 runner 或冒充恢复可用性结论。
+
+## Acceptance Criteria
+
+- [ ] AC1：只读调查首先明确生产本地是否存在通过完整合同校验的可用批次，并区分“异机副本缺失/生产单点”与“当前无已证明可用备份”；同时取得成功/失败运行的清理保留语义、两个 partial 的聚合停止偏移、真实 rsync 退出码、可靠失败持续时间（或明确未测原因）、生产空间与累积状态，形成证据支持的根因分类和过渡恢复点建议。
+- [ ] AC2：经单独批准的最小传输修复与容量处置完成，变更、回滚点和最终容量预算有脱敏证据；未修改正式归档、恢复门禁或业务数据。
+- [ ] AC3：ARM 形成至少一个新的完整原子归档批次，并通过 manifest、精确文件集、run id 配对、字节数和 SHA-256 校验；partial 不得冒充完成。
+- [ ] AC4：生产 daily/watchdog 与 ARM dead-man 的 timer 正常触发，最近 service 均 `Result=success`、exit `0`，新鲜度状态与最新原子归档一致。
+- [ ] AC5：连续两个自然 daily 调度周期均产生新的完整原子归档并通过 AC3/AC4；手工补跑不计入这两个周期；两个周期的 ARM partial 数量/聚合字节和 ARM 容量趋势均已记录且未继续恶化。
+- [ ] AC6：生产容量达到经审阅的“一个完整新批次实际峰值 + 安全余量”预算，两个自然周期内无继续逼近失败线的趋势。
+- [ ] AC7：告警响应计时合同已启用；观察窗口内若出现真实告警/恢复通知，则同时记录发送调用时刻、人工确认时刻和响应延迟；若没有自然通知，明确标为未触发，不发送未批准的测试消息，也不把缺样本写成响应闭环已验证。
+- [ ] AC8：演练任务保留阻塞交叉引用；task validate、Trellis governance、流程合同、文档审计、敏感扫描和 `git diff --check` 通过。
+
+## Out of Scope
+
+- 本轮规划不授权连接生产/ARM、不授权只读调查执行，也不授权任何修复、清理、重跑或告警发送。
+- 不恢复数据库或附件，不启动隔离 OA，不修改恢复候选、migration、业务代码、生产数据库或业务附件。
+- 不轮换备份 SSH 密钥、不重建 receiver，除非后续只读证据推翻“初始认证已成功”的当前事实并重新审阅设计。
+- 不删除 ARM 正式归档或历史 partial；历史 partial 的清理须在身份与保留价值审阅后单独批准。
+- 不把一次手工成功、timer active、service 通用 exit 或 `sent=true` 单独作为结案依据。
+
+## Blocking Open Questions
+
+- 无。第一阶段只读调查、任何本地实现以及生产/ARM 写操作仍分别需要后续明确批准。
+
+## Test Plan
+
+- 本地使用合成 fixture 验证 partial 聚合比较、rsync 退出码映射、失败时长聚类、容量预算和敏感输出 allowlist。
+- 获批后对生产/ARM 只读采集 R2 字段，保持原始日志、路径、身份和业务数据不出远端私有边界。
+- 修复候选先通过 Bash/Python 语法、既有自动异机备份/receiver/dead-man 合同和回滚演练；实际变更另行批准。
+- 实际修复后校验新原子批次与三组 timer/service，并连续观察两个自然调度周期。
+- 运行 task validate、Trellis governance、process baseline、docs registry、敏感扫描和 `git diff --check`。

.trellis/tasks/08-28-production-arm-backup-transfer-incident/task.json

@@ -0,0 +1,26 @@
+{
+  "id": "production-arm-backup-transfer-incident",
+  "name": "production-arm-backup-transfer-incident",
+  "title": "生产到 ARM 备份传输中断",
+  "description": "调查并修复生产到 ARM 的自动异机备份传输中断，同时处置生产容量风险并验证连续两个自然调度周期。",
+  "status": "in_progress",
+  "dev_type": null,
+  "scope": "operations",
+  "package": null,
+  "priority": "P1",
+  "creator": "shunli",
+  "assignee": "shunli",
+  "createdAt": "2026-08-28",
+  "completedAt": null,
+  "branch": null,
+  "base_branch": "main",
+  "worktree_path": null,
+  "commit": null,
+  "pr_url": null,
+  "subtasks": [],
+  "children": [],
+  "parent": null,
+  "relatedFiles": [],
+  "notes": "来源证据见 08-23-isolated-backup-restore-drill 的 2026-08-27 自动备份链只读调查。创建与规划不授权连接生产/ARM；只读深查、生产/ARM 写操作、手工补跑和测试告警分别审批。",
+  "meta": {}
+}

scripts/qa/inspect-backup-transfer-incident.py

@@ -0,0 +1,505 @@
+#!/usr/bin/env python3
+"""Read-only, redacted inspection of an interrupted automatic-backup chain."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import stat
+import sys
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterable
+
+
+BATCH_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))$")
+PARTIAL_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))\.partial$")
+TEMP_RE = re.compile(r"^\.hyb-oa-backup-.+\.tmp\..+$")
+EXPECTED_FILES = {"database.dump", "storage.tar.gz", "manifest.json"}
+ARTIFACTS = {"database": "database.dump", "storage": "storage.tar.gz"}
+RSYNC_CODE_RE = re.compile(r"rsync error:[^\r\n]*\(code (11|12|23|30|255)\)")
+RSYNC_CATEGORIES = {
+    "11": "file_io_error",
+    "12": "protocol_stream_error",
+    "23": "partial_transfer_error",
+    "30": "transport_timeout",
+    "255": "ssh_session_error",
+}
+SIMILAR_RELATIVE_PPM = 50_000
+SIMILAR_ABSOLUTE_BYTES = 64 * 1024 * 1024
+SAFE_KEY_RE = re.compile(r"^[a-z0-9_]+$")
+SAFE_VALUE_RE = re.compile(r"^(?:[A-Za-z][A-Za-z0-9_]*|-?[0-9]+)$")
+
+
+class InspectionError(Exception):
+    """Stable, non-sensitive inspection failure."""
+
+    def __init__(self, category: str):
+        super().__init__(category)
+        self.category = category
+
+
+@dataclass(frozen=True)
+class BatchResult:
+    valid: bool
+    total_bytes: int
+    created_epoch: int | None
+    expected_artifact_bytes: int | None
+    received_artifact_bytes: int
+
+
+def emit(items: Iterable[tuple[str, object]]) -> None:
+    rendered_items: list[tuple[str, str]] = []
+    for key, value in items:
+        if isinstance(value, bool):
+            rendered = "yes" if value else "no"
+        else:
+            rendered = str(value)
+        if not SAFE_KEY_RE.fullmatch(key) or not SAFE_VALUE_RE.fullmatch(rendered):
+            print("probe_result=FAIL")
+            print("probe_category=output_allowlist_violation")
+            raise SystemExit(1)
+        rendered_items.append((key, rendered))
+    for key, rendered in rendered_items:
+        print(f"{key}={rendered}")
+
+
+def safe_root(path: Path) -> Path:
+    try:
+        if not path.is_absolute() or path.is_symlink() or not path.is_dir():
+            raise InspectionError("root_invalid")
+        list(path.iterdir())
+    except PermissionError as error:
+        raise InspectionError("access_denied") from error
+    except OSError as error:
+        raise InspectionError("root_unreadable") from error
+    return path
+
+
+def file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def tree_bytes(path: Path) -> int:
+    total = 0
+    try:
+        for root, directories, files in os.walk(path, followlinks=False):
+            directories[:] = [
+                name for name in directories if not (Path(root) / name).is_symlink()
+            ]
+            for name in files:
+                item = Path(root) / name
+                if item.is_symlink():
+                    continue
+                total += item.stat().st_size
+    except PermissionError as error:
+        raise InspectionError("access_denied") from error
+    except OSError as error:
+        raise InspectionError("entry_unreadable") from error
+    return total
+
+
+def run_epoch(stamp: str) -> int | None:
+    try:
+        parsed = datetime.strptime(stamp, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
+    except ValueError:
+        return None
+    return int(parsed.timestamp())
+
+
+def validate_batch(path: Path, expected_run_id: str, stamp: str) -> BatchResult:
+    total = tree_bytes(path)
+    received_artifact_bytes = 0
+    expected_artifact_bytes: int | None = None
+    created_epoch = run_epoch(stamp)
+    if created_epoch is None:
+        return BatchResult(False, total, None, None, 0)
+    try:
+        entries = list(path.iterdir())
+        regular_names = {
+            item.name
+            for item in entries
+            if not item.is_symlink() and stat.S_ISREG(item.stat().st_mode)
+        }
+        if len(entries) != 3 or regular_names != EXPECTED_FILES:
+            return BatchResult(False, total, created_epoch, None, 0)
+        manifest = json.loads((path / "manifest.json").read_text(encoding="utf-8"))
+        if (
+            manifest.get("schemaVersion") != 1
+            or manifest.get("status") != "complete"
+            or manifest.get("runId") != expected_run_id
+            or not (manifest.get("buildSha") or manifest.get("sourceImageDigest"))
+        ):
+            return BatchResult(False, total, created_epoch, None, 0)
+        artifacts = manifest.get("artifacts") or {}
+        expected_artifact_bytes = 0
+        for key, expected_name in ARTIFACTS.items():
+            item = artifacts.get(key) or {}
+            target = path / expected_name
+            if item.get("file") != expected_name or isinstance(item.get("bytes"), bool):
+                return BatchResult(False, total, created_epoch, None, 0)
+            expected_size = item.get("bytes")
+            if not isinstance(expected_size, int) or expected_size < 0:
+                return BatchResult(False, total, created_epoch, None, 0)
+            expected_artifact_bytes += expected_size
+            actual_size = target.stat().st_size
+            received_artifact_bytes += actual_size
+            if actual_size != expected_size or file_sha256(target) != item.get("sha256"):
+                return BatchResult(
+                    False,
+                    total,
+                    created_epoch,
+                    expected_artifact_bytes,
+                    received_artifact_bytes,
+                )
+        return BatchResult(
+            True,
+            total,
+            created_epoch,
+            expected_artifact_bytes,
+            received_artifact_bytes,
+        )
+    except PermissionError as error:
+        raise InspectionError("access_denied") from error
+    except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
+        return BatchResult(False, total, created_epoch, expected_artifact_bytes, received_artifact_bytes)
+
+
+def inspect_partial(path: Path, run_id: str, stamp: str) -> BatchResult:
+    """Measure a partial without requiring the complete three-file contract."""
+    total = tree_bytes(path)
+    received = 0
+    expected: int | None = None
+    try:
+        manifest_path = path / "manifest.json"
+        if manifest_path.is_file() and not manifest_path.is_symlink():
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+            if manifest.get("runId") == run_id and manifest.get("schemaVersion") == 1:
+                artifacts = manifest.get("artifacts") or {}
+                sizes: list[int] = []
+                for key, expected_name in ARTIFACTS.items():
+                    item = artifacts.get(key) or {}
+                    size = item.get("bytes")
+                    if (
+                        item.get("file") != expected_name
+                        or isinstance(size, bool)
+                        or not isinstance(size, int)
+                        or size < 0
+                    ):
+                        sizes = []
+                        break
+                    sizes.append(size)
+                    target = path / expected_name
+                    if target.is_file() and not target.is_symlink():
+                        received += target.stat().st_size
+                if len(sizes) == 2:
+                    expected = sum(sizes)
+        validated = validate_batch(path, run_id, stamp)
+        return BatchResult(validated.valid, total, run_epoch(stamp), expected, received)
+    except PermissionError as error:
+        raise InspectionError("access_denied") from error
+    except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
+        return BatchResult(False, total, run_epoch(stamp), expected, received)
+
+
+def filesystem_fields(prefix: str, root: Path) -> list[tuple[str, object]]:
+    values = os.statvfs(root)
+    total = values.f_blocks * values.f_frsize
+    available = values.f_bavail * values.f_frsize
+    used_percent = 0 if total == 0 else ((total - available) * 100) // total
+    return [
+        (f"{prefix}_disk_used_percent", used_percent),
+        (f"{prefix}_disk_free_gib", available // (1024**3)),
+    ]
+
+
+def source_contract(wrapper: Path, backup_script: Path) -> list[tuple[str, object]]:
+    try:
+        if (
+            wrapper.is_symlink()
+            or not wrapper.is_file()
+            or backup_script.is_symlink()
+            or not backup_script.is_file()
+        ):
+            raise InspectionError("source_contract_invalid")
+        wrapper_text = wrapper.read_text(encoding="utf-8")
+        backup_text = backup_script.read_text(encoding="utf-8")
+    except PermissionError as error:
+        raise InspectionError("source_contract_access_denied") from error
+    except OSError as error:
+        raise InspectionError("source_contract_unreadable") from error
+
+    if "run_backup() {" not in wrapper_text or "\ncheck_stale() {" not in wrapper_text:
+        raise InspectionError("source_contract_invalid")
+    run_backup_body = wrapper_text.split("run_backup() {", 1)[1].split("\ncheck_stale() {", 1)[0]
+    markers = {
+        "local_publish": '[[ -d "$batch_dir" && -f "$batch_dir/manifest.json" ]]',
+        "transfer": "capture_command offhost_transfer rsync",
+        "finalize": "capture_command offhost_finalize ssh",
+        "success": 'write_success_state "$run_id" "$manifest_sha"',
+        "prune": 'prune_local_backups "$batch_name"',
+    }
+    positions = {key: run_backup_body.find(marker) for key, marker in markers.items()}
+    all_present = all(position >= 0 for position in positions.values())
+    retained = all_present and positions["local_publish"] < positions["transfer"] < positions["prune"]
+    pruned = all_present and positions["finalize"] < positions["success"] < positions["prune"]
+    temp_cleanup = all(
+        marker in backup_text
+        for marker in (
+            "trap cleanup EXIT",
+            'TEMP_DIR="$(mktemp -d',
+            'rm -rf -- "$TEMP_DIR"',
+            'mv "$TEMP_DIR" "$FINAL_DIR"',
+            'TEMP_DIR=""',
+            "trap - EXIT",
+        )
+    )
+    return [
+        ("retained_after_transfer_failure", retained),
+        ("pruned_after_offhost_success", pruned),
+        ("generation_temp_cleanup_on_failure", temp_cleanup),
+        ("cleanup_contract_source_verified", all_present and temp_cleanup),
+    ]
+
+
+def last_backup_phase(state_dir: Path) -> str:
+    state_file = state_dir / "last-run.json"
+    if not state_file.exists():
+        return "NOT_AVAILABLE"
+    try:
+        if state_file.is_symlink() or not state_file.is_file():
+            return "STATE_INVALID"
+        payload = json.loads(state_file.read_text(encoding="utf-8"))
+        findings = payload.get("findings") or []
+        detail = findings[0].get("detail") if len(findings) == 1 and isinstance(findings[0], dict) else None
+        match = re.fullmatch(r"phase_([a-z_]+)_failed", detail or "")
+        allowed = {
+            "runtime_identity",
+            "paired_backup",
+            "offhost_transfer",
+            "offhost_finalize",
+            "success_state",
+            "stale_check",
+            "quarterly_reminder",
+        }
+        return match.group(1) if match and match.group(1) in allowed else "NOT_AVAILABLE"
+    except PermissionError as error:
+        raise InspectionError("diagnostic_access_denied") from error
+    except (AttributeError, json.JSONDecodeError, OSError, TypeError):
+        return "STATE_INVALID"
+
+
+def parse_transfer_diagnostic(state_dir: Path) -> list[tuple[str, object]]:
+    phase = last_backup_phase(state_dir)
+    diagnostic = state_dir / "last-error.log"
+    if phase != "offhost_transfer" or not diagnostic.exists():
+        code = "NOT_AVAILABLE"
+    else:
+        try:
+            if diagnostic.is_symlink() or not diagnostic.is_file():
+                raise InspectionError("diagnostic_invalid")
+            text = diagnostic.read_text(encoding="utf-8", errors="replace")
+        except PermissionError as error:
+            raise InspectionError("diagnostic_access_denied") from error
+        except OSError as error:
+            raise InspectionError("diagnostic_unreadable") from error
+        matches = RSYNC_CODE_RE.findall(text)
+        code = matches[-1] if matches else "NOT_AVAILABLE"
+    return [
+        ("last_backup_failure_phase", phase),
+        ("transfer_exit_code", code),
+        ("transfer_exit_code_source", "last_error_log_regex" if code != "NOT_AVAILABLE" else "NOT_AVAILABLE"),
+        ("transfer_failure_category", RSYNC_CATEGORIES.get(code, "not_available")),
+        ("transfer_started_epoch", "NOT_AVAILABLE"),
+        ("transfer_failed_epoch", "NOT_AVAILABLE"),
+        ("transfer_duration_seconds", "NOT_AVAILABLE"),
+        ("transfer_duration_reason", "wrapper_does_not_persist_transfer_clock"),
+    ]
+
+
+def production(args: argparse.Namespace) -> None:
+    backup_root = safe_root(args.backup_root)
+    state_dir = safe_root(args.state_dir)
+    now = args.now_epoch
+    valid: list[BatchResult] = []
+    invalid_count = invalid_bytes = temp_count = temp_bytes = other_count = 0
+    for entry in backup_root.iterdir():
+        match = BATCH_RE.fullmatch(entry.name)
+        if match and entry.is_dir() and not entry.is_symlink():
+            result = validate_batch(entry, match.group(1), match.group(2))
+            if result.valid and result.created_epoch is not None and result.created_epoch <= now:
+                valid.append(result)
+            else:
+                invalid_count += 1
+                invalid_bytes += result.total_bytes
+        elif TEMP_RE.fullmatch(entry.name) and entry.is_dir() and not entry.is_symlink():
+            temp_count += 1
+            temp_bytes += tree_bytes(entry)
+        else:
+            other_count += 1
+
+    valid.sort(key=lambda item: item.created_epoch or -1)
+    latest = valid[-1] if valid else None
+    output: list[tuple[str, object]] = [
+        ("production_probe_result", "PASS"),
+        ("complete_local_recovery_point", bool(valid)),
+        ("backup_risk_classification", "production_single_point" if valid else "no_proven_recovery_point"),
+        ("complete_local_batch_count", len(valid)),
+        ("complete_local_batch_total_bytes", sum(item.total_bytes for item in valid)),
+        (
+            "latest_complete_local_batch_age_seconds",
+            now - latest.created_epoch
+            if latest and latest.created_epoch is not None
+            else "NOT_AVAILABLE",
+        ),
+        ("latest_complete_local_batch_bytes", latest.total_bytes if latest else "NOT_AVAILABLE"),
+        ("invalid_local_batch_count", invalid_count),
+        ("invalid_local_batch_total_bytes", invalid_bytes),
+        ("generation_temp_count", temp_count),
+        ("generation_temp_total_bytes", temp_bytes),
+        ("unclassified_local_entry_count", other_count),
+        ("transition_recovery_point_action", "wait_natural" if valid else "request_manual_run"),
+    ]
+    output.extend(filesystem_fields("production_backup", backup_root))
+    output.extend(source_contract(args.wrapper_script, args.backup_script))
+    output.extend(parse_transfer_diagnostic(state_dir))
+    emit(output)
+
+
+def stopping_class(samples: list[BatchResult]) -> str:
+    if len(samples) < 2:
+        return "insufficient_samples"
+    sizes = sorted(item.total_bytes for item in samples[-2:])
+    difference = sizes[1] - sizes[0]
+    relative_limit = sizes[1] * SIMILAR_RELATIVE_PPM // 1_000_000
+    return "similar" if difference <= max(SIMILAR_ABSOLUTE_BYTES, relative_limit) else "different"
+
+
+def arm(args: argparse.Namespace) -> None:
+    inbox_root = safe_root(args.inbox_root)
+    archive_root = safe_root(args.archive_root)
+    now = args.now_epoch
+    partials: list[BatchResult] = []
+    other_count = 0
+    for entry in inbox_root.iterdir():
+        match = PARTIAL_RE.fullmatch(entry.name)
+        if match and entry.is_dir() and not entry.is_symlink():
+            partials.append(inspect_partial(entry, match.group(1), match.group(2)))
+        else:
+            other_count += 1
+    partials.sort(key=lambda item: item.created_epoch or -1)
+
+    archives: list[BatchResult] = []
+    invalid_archive_count = 0
+    for entry in archive_root.iterdir():
+        match = BATCH_RE.fullmatch(entry.name)
+        if match and entry.is_dir() and not entry.is_symlink():
+            result = validate_batch(entry, match.group(1), match.group(2))
+            if result.valid and result.created_epoch is not None and result.created_epoch <= now:
+                archives.append(result)
+            else:
+                invalid_archive_count += 1
+    archives.sort(key=lambda item: item.created_epoch or -1)
+
+    output: list[tuple[str, object]] = [
+        ("arm_probe_result", "PASS"),
+        ("arm_partial_count", len(partials)),
+        ("arm_partial_total_bytes", sum(item.total_bytes for item in partials)),
+        ("arm_partial_complete_contract_count", sum(1 for item in partials if item.valid)),
+        ("arm_partial_incomplete_contract_count", sum(1 for item in partials if not item.valid)),
+        ("arm_partial_stop_offset_class", stopping_class(partials)),
+        ("arm_partial_similarity_relative_ppm", SIMILAR_RELATIVE_PPM),
+        ("arm_partial_similarity_absolute_bytes", SIMILAR_ABSOLUTE_BYTES),
+        ("arm_inbox_other_entry_count", other_count),
+        ("arm_valid_atomic_archive_count", len(archives)),
+        ("arm_invalid_atomic_archive_count", invalid_archive_count),
+        (
+            "arm_latest_atomic_archive_age_seconds",
+            now - archives[-1].created_epoch if archives else "NOT_AVAILABLE",
+        ),
+    ]
+    for index, sample in enumerate(partials, 1):
+        ratio = (
+            min(sample.received_artifact_bytes, sample.expected_artifact_bytes)
+            * 1_000_000
+            // sample.expected_artifact_bytes
+            if sample.expected_artifact_bytes and sample.expected_artifact_bytes > 0
+            else "NOT_AVAILABLE"
+        )
+        output.extend(
+            [
+                (f"arm_partial_sample_{index}_total_bytes", sample.total_bytes),
+                (
+                    f"arm_partial_sample_{index}_expected_artifact_bytes",
+                    sample.expected_artifact_bytes
+                    if sample.expected_artifact_bytes is not None
+                    else "NOT_AVAILABLE",
+                ),
+                (f"arm_partial_sample_{index}_received_ratio_ppm", ratio),
+                (
+                    f"arm_partial_sample_{index}_age_seconds",
+                    now - sample.created_epoch
+                    if sample.created_epoch is not None
+                    else "NOT_AVAILABLE",
+                ),
+                (f"arm_partial_sample_{index}_complete_contract", sample.valid),
+            ]
+        )
+    output.extend(filesystem_fields("arm_inbox", inbox_root))
+    output.extend(filesystem_fields("arm_archive", archive_root))
+    emit(output)
+
+
+def absolute_path(value: str) -> Path:
+    path = Path(value)
+    if not path.is_absolute():
+        raise argparse.ArgumentTypeError("an explicit absolute path is required")
+    return path
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Emit redacted backup-chain incident fields")
+    parser.add_argument("--now-epoch", type=int, default=int(datetime.now(timezone.utc).timestamp()))
+    subparsers = parser.add_subparsers(dest="mode", required=True)
+    production_parser = subparsers.add_parser("production")
+    production_parser.add_argument("--backup-root", type=absolute_path, required=True)
+    production_parser.add_argument("--state-dir", type=absolute_path, required=True)
+    production_parser.add_argument("--wrapper-script", type=absolute_path, required=True)
+    production_parser.add_argument("--backup-script", type=absolute_path, required=True)
+    production_parser.set_defaults(handler=production)
+    arm_parser = subparsers.add_parser("arm")
+    arm_parser.add_argument("--inbox-root", type=absolute_path, required=True)
+    arm_parser.add_argument("--archive-root", type=absolute_path, required=True)
+    arm_parser.set_defaults(handler=arm)
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        args.handler(args)
+    except InspectionError as error:
+        emit([("probe_result", "FAIL"), ("probe_category", error.category)])
+        return 1
+    except PermissionError:
+        emit([("probe_result", "FAIL"), ("probe_category", "access_denied")])
+        return 1
+    except (OSError, TypeError, ValueError):
+        emit([("probe_result", "FAIL"), ("probe_category", "inspection_io_error")])
+        return 1
+    except Exception:
+        emit([("probe_result", "FAIL"), ("probe_category", "inspection_internal_error")])
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/qa/run-backup-transfer-incident-contract.sh

@@ -0,0 +1,287 @@
+#!/usr/bin/env bash
+set -eEuo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+INSPECTOR="$ROOT_DIR/scripts/qa/inspect-backup-transfer-incident.py"
+TEST_ROOT="$(mktemp -d)"
+PASS_COUNT=0
+
+cleanup() {
+  if [[ -n "${TEST_ROOT:-}" && -d "$TEST_ROOT" && "$TEST_ROOT" == /tmp/* ]]; then
+    rm -rf -- "$TEST_ROOT"
+  fi
+}
+trap cleanup EXIT
+
+pass() {
+  PASS_COUNT=$((PASS_COUNT + 1))
+  printf '[PASS] %s\n' "$1"
+}
+
+assert_line() {
+  local file="$1"
+  local expected="$2"
+  grep -Fxq -- "$expected" "$file" || {
+    printf '[FAIL] missing field: %s\n' "$expected" >&2
+    exit 1
+  }
+}
+
+make_batch() {
+  local target="$1"
+  local run_id="$2"
+  local db_bytes="${3:-1024}"
+  local storage_bytes="${4:-2048}"
+  mkdir -p "$target"
+  head -c "$db_bytes" /dev/zero >"$target/database.dump"
+  head -c "$storage_bytes" /dev/zero >"$target/storage.tar.gz"
+  python3 - "$target" "$run_id" <<'PY'
+import hashlib
+import json
+import sys
+from pathlib import Path
+
+target = Path(sys.argv[1])
+run_id = sys.argv[2]
+
+def item(name):
+    data = (target / name).read_bytes()
+    return {"file": name, "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}
+
+payload = {
+    "schemaVersion": 1,
+    "runId": run_id,
+    "createdAt": "2026-08-28T00:00:00Z",
+    "buildSha": "a" * 40,
+    "sourceImageDigest": None,
+    "status": "complete",
+    "artifacts": {"database": item("database.dump"), "storage": item("storage.tar.gz")},
+}
+(target / "manifest.json").write_text(json.dumps(payload) + "\n", encoding="utf-8")
+PY
+}
+
+run_production() {
+  local fixture="$1"
+  local output="$2"
+  python3 -B "$INSPECTOR" --now-epoch 1787875200 production \
+    --backup-root "$fixture/backups" \
+    --state-dir "$fixture/state" \
+    --wrapper-script "$ROOT_DIR/scripts/run-automatic-backup.sh" \
+    --backup-script "$ROOT_DIR/scripts/backup-all.sh" >"$output"
+}
+
+make_production_layout() {
+  local fixture="$1"
+  mkdir -p "$fixture/backups" "$fixture/state"
+}
+
+write_transfer_failure_state() {
+  local fixture="$1"
+  printf '%s\n' '{"findings":[{"detail":"phase_offhost_transfer_failed"}]}' >"$fixture/state/last-run.json"
+}
+
+PRODUCTION_VALID="$TEST_ROOT/production-valid"
+make_production_layout "$PRODUCTION_VALID"
+make_batch "$PRODUCTION_VALID/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+write_transfer_failure_state "$PRODUCTION_VALID"
+printf 'rsync error: synthetic redacted fixture (code 23) at fixture.c(1)\n' >"$PRODUCTION_VALID/state/last-error.log"
+run_production "$PRODUCTION_VALID" "$TEST_ROOT/production-valid.out"
+assert_line "$TEST_ROOT/production-valid.out" "complete_local_recovery_point=yes"
+assert_line "$TEST_ROOT/production-valid.out" "backup_risk_classification=production_single_point"
+assert_line "$TEST_ROOT/production-valid.out" "transition_recovery_point_action=wait_natural"
+assert_line "$TEST_ROOT/production-valid.out" "transfer_exit_code=23"
+assert_line "$TEST_ROOT/production-valid.out" "transfer_failure_category=partial_transfer_error"
+assert_line "$TEST_ROOT/production-valid.out" "transfer_duration_seconds=NOT_AVAILABLE"
+pass "生产存在完整本地恢复点时识别为单点风险并保留未测时长"
+
+PRODUCTION_EMPTY="$TEST_ROOT/production-empty"
+make_production_layout "$PRODUCTION_EMPTY"
+run_production "$PRODUCTION_EMPTY" "$TEST_ROOT/production-empty.out"
+assert_line "$TEST_ROOT/production-empty.out" "complete_local_recovery_point=no"
+assert_line "$TEST_ROOT/production-empty.out" "backup_risk_classification=no_proven_recovery_point"
+assert_line "$TEST_ROOT/production-empty.out" "transition_recovery_point_action=request_manual_run"
+pass "生产无完整本地恢复点时升级风险与过渡恢复点建议"
+
+PRODUCTION_CORRUPT="$TEST_ROOT/production-corrupt"
+make_production_layout "$PRODUCTION_CORRUPT"
+make_batch "$PRODUCTION_CORRUPT/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+printf 'tamper' >>"$PRODUCTION_CORRUPT/backups/hyb-oa-backup-auto-20260828T000000Z/database.dump"
+run_production "$PRODUCTION_CORRUPT" "$TEST_ROOT/production-corrupt.out"
+assert_line "$TEST_ROOT/production-corrupt.out" "complete_local_recovery_point=no"
+assert_line "$TEST_ROOT/production-corrupt.out" "invalid_local_batch_count=1"
+pass "manifest 大小或 SHA 篡改不得冒充完整本地恢复点"
+
+PRODUCTION_MALFORMED="$TEST_ROOT/production-malformed"
+make_production_layout "$PRODUCTION_MALFORMED"
+make_batch "$PRODUCTION_MALFORMED/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+printf '%s\n' '[]' >"$PRODUCTION_MALFORMED/backups/hyb-oa-backup-auto-20260828T000000Z/manifest.json"
+run_production "$PRODUCTION_MALFORMED" "$TEST_ROOT/production-malformed.out"
+assert_line "$TEST_ROOT/production-malformed.out" "complete_local_recovery_point=no"
+assert_line "$TEST_ROOT/production-malformed.out" "invalid_local_batch_count=1"
+pass "畸形 manifest 安全归类为无效批次且不输出 traceback"
+
+PRODUCTION_UNREADABLE="$TEST_ROOT/production-unreadable"
+make_production_layout "$PRODUCTION_UNREADABLE"
+make_batch "$PRODUCTION_UNREADABLE/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+chmod 000 "$PRODUCTION_UNREADABLE/backups/hyb-oa-backup-auto-20260828T000000Z/manifest.json"
+if [[ "$(id -u)" == 0 ]] && command -v runuser >/dev/null 2>&1 && id nobody >/dev/null 2>&1; then
+  chmod 0755 "$TEST_ROOT" "$PRODUCTION_UNREADABLE" "$PRODUCTION_UNREADABLE/backups" "$PRODUCTION_UNREADABLE/state" \
+    "$PRODUCTION_UNREADABLE/backups/hyb-oa-backup-auto-20260828T000000Z"
+  if runuser -u nobody -- python3 -B "$INSPECTOR" --now-epoch 1787875200 production \
+    --backup-root "$PRODUCTION_UNREADABLE/backups" \
+    --state-dir "$PRODUCTION_UNREADABLE/state" \
+    --wrapper-script "$ROOT_DIR/scripts/run-automatic-backup.sh" \
+    --backup-script "$ROOT_DIR/scripts/backup-all.sh" >"$TEST_ROOT/production-unreadable.out"; then
+    printf '[FAIL] unreadable manifest unexpectedly passed\n' >&2
+    exit 1
+  fi
+else
+  if run_production "$PRODUCTION_UNREADABLE" "$TEST_ROOT/production-unreadable.out"; then
+    printf '[FAIL] unreadable manifest unexpectedly passed\n' >&2
+    exit 1
+  fi
+fi
+chmod 0600 "$PRODUCTION_UNREADABLE/backups/hyb-oa-backup-auto-20260828T000000Z/manifest.json"
+assert_line "$TEST_ROOT/production-unreadable.out" "probe_category=access_denied"
+pass "访问拒绝保持为 unknown/ACCESS_DENIED 而不是伪装成无有效备份"
+
+assert_line "$TEST_ROOT/production-valid.out" "retained_after_transfer_failure=yes"
+assert_line "$TEST_ROOT/production-valid.out" "pruned_after_offhost_success=yes"
+assert_line "$TEST_ROOT/production-valid.out" "generation_temp_cleanup_on_failure=yes"
+assert_line "$TEST_ROOT/production-valid.out" "cleanup_contract_source_verified=yes"
+pass "生产脚本顺序机械证明失败保留、成功后 prune 与生成临时清理"
+
+SOURCE_MUTATION="$TEST_ROOT/source-mutation"
+cp "$ROOT_DIR/scripts/run-automatic-backup.sh" "$SOURCE_MUTATION"
+python3 - "$SOURCE_MUTATION" <<'PY'
+import sys
+from pathlib import Path
+
+path = Path(sys.argv[1])
+text = path.read_text(encoding="utf-8")
+transfer = text.index("capture_command offhost_transfer rsync")
+text = text[:transfer] + 'prune_local_backups "$batch_name"\n  ' + text[transfer:]
+path.write_text(text, encoding="utf-8")
+PY
+python3 -B "$INSPECTOR" --now-epoch 1787875200 production \
+  --backup-root "$PRODUCTION_VALID/backups" \
+  --state-dir "$PRODUCTION_VALID/state" \
+  --wrapper-script "$SOURCE_MUTATION" \
+  --backup-script "$ROOT_DIR/scripts/backup-all.sh" >"$TEST_ROOT/source-mutation.out"
+assert_line "$TEST_ROOT/source-mutation.out" "retained_after_transfer_failure=no"
+assert_line "$TEST_ROOT/source-mutation.out" "pruned_after_offhost_success=no"
+pass "源码顺序漂移会关闭失败保留与成功后 prune 结论"
+
+for code_category in \
+  23:partial_transfer_error \
+  30:transport_timeout \
+  12:protocol_stream_error \
+  11:file_io_error \
+  255:ssh_session_error; do
+  code="${code_category%%:*}"
+  category="${code_category#*:}"
+  fixture="$TEST_ROOT/rsync-$code"
+  make_production_layout "$fixture"
+  write_transfer_failure_state "$fixture"
+  printf 'rsync error: synthetic redacted fixture (code %s) at fixture.c(1)\n' "$code" >"$fixture/state/last-error.log"
+  run_production "$fixture" "$TEST_ROOT/rsync-$code.out"
+  assert_line "$TEST_ROOT/rsync-$code.out" "transfer_exit_code=$code"
+  assert_line "$TEST_ROOT/rsync-$code.out" "transfer_failure_category=$category"
+done
+pass "rsync 23/30/12/11/255 均映射为稳定脱敏类别"
+
+PRODUCTION_UNKNOWN="$TEST_ROOT/production-unknown"
+make_production_layout "$PRODUCTION_UNKNOWN"
+write_transfer_failure_state "$PRODUCTION_UNKNOWN"
+printf 'unstructured synthetic transport failure\n' >"$PRODUCTION_UNKNOWN/state/last-error.log"
+run_production "$PRODUCTION_UNKNOWN" "$TEST_ROOT/production-unknown.out"
+assert_line "$TEST_ROOT/production-unknown.out" "transfer_exit_code=NOT_AVAILABLE"
+assert_line "$TEST_ROOT/production-unknown.out" "transfer_failure_category=not_available"
+pass "原始 rsync 码不存在时安全失败为 NOT_AVAILABLE"
+
+PRODUCTION_PHASE_MISMATCH="$TEST_ROOT/production-phase-mismatch"
+make_production_layout "$PRODUCTION_PHASE_MISMATCH"
+printf '%s\n' '{"findings":[{"detail":"phase_offhost_finalize_failed"}]}' >"$PRODUCTION_PHASE_MISMATCH/state/last-run.json"
+printf 'rsync error: stale synthetic fixture (code 23) at fixture.c(1)\n' >"$PRODUCTION_PHASE_MISMATCH/state/last-error.log"
+run_production "$PRODUCTION_PHASE_MISMATCH" "$TEST_ROOT/production-phase-mismatch.out"
+assert_line "$TEST_ROOT/production-phase-mismatch.out" "last_backup_failure_phase=offhost_finalize"
+assert_line "$TEST_ROOT/production-phase-mismatch.out" "transfer_exit_code=NOT_AVAILABLE"
+assert_line "$TEST_ROOT/production-phase-mismatch.out" "transfer_exit_code_source=NOT_AVAILABLE"
+pass "私有诊断只在安全摘要明确为 offhost_transfer 时绑定 rsync 原始码"
+
+ARM_SIMILAR="$TEST_ROOT/arm-similar"
+mkdir -p "$ARM_SIMILAR/inbox" "$ARM_SIMILAR/archive"
+make_batch "$ARM_SIMILAR/inbox/hyb-oa-backup-auto-20260827T000000Z.partial" "auto-20260827T000000Z" 1000 1000
+make_batch "$ARM_SIMILAR/inbox/hyb-oa-backup-auto-20260828T000000Z.partial" "auto-20260828T000000Z" 1100 1000
+rm "$ARM_SIMILAR/inbox/hyb-oa-backup-auto-20260827T000000Z.partial/storage.tar.gz"
+rm "$ARM_SIMILAR/inbox/hyb-oa-backup-auto-20260828T000000Z.partial/storage.tar.gz"
+python3 -B "$INSPECTOR" --now-epoch 1787875200 arm --inbox-root "$ARM_SIMILAR/inbox" --archive-root "$ARM_SIMILAR/archive" >"$TEST_ROOT/arm-similar.out"
+assert_line "$TEST_ROOT/arm-similar.out" "arm_partial_count=2"
+assert_line "$TEST_ROOT/arm-similar.out" "arm_partial_stop_offset_class=similar"
+assert_line "$TEST_ROOT/arm-similar.out" "arm_partial_incomplete_contract_count=2"
+pass "两个 partial 接近停止偏移按预先冻结阈值分类为 similar"
+
+ARM_DIFFERENT="$TEST_ROOT/arm-different"
+mkdir -p "$ARM_DIFFERENT/inbox" "$ARM_DIFFERENT/archive"
+make_batch "$ARM_DIFFERENT/inbox/hyb-oa-backup-auto-20260827T000000Z.partial" "auto-20260827T000000Z" 1000 1000
+make_batch "$ARM_DIFFERENT/inbox/hyb-oa-backup-auto-20260828T000000Z.partial" "auto-20260828T000000Z" 70000000 1000
+rm "$ARM_DIFFERENT/inbox/hyb-oa-backup-auto-20260827T000000Z.partial/storage.tar.gz"
+rm "$ARM_DIFFERENT/inbox/hyb-oa-backup-auto-20260828T000000Z.partial/storage.tar.gz"
+python3 -B "$INSPECTOR" --now-epoch 1787875200 arm --inbox-root "$ARM_DIFFERENT/inbox" --archive-root "$ARM_DIFFERENT/archive" >"$TEST_ROOT/arm-different.out"
+assert_line "$TEST_ROOT/arm-different.out" "arm_partial_stop_offset_class=different"
+pass "两个 partial 明显不同停止偏移分类为 different"
+
+ARM_COMPLETE="$TEST_ROOT/arm-complete"
+mkdir -p "$ARM_COMPLETE/inbox" "$ARM_COMPLETE/archive"
+make_batch "$ARM_COMPLETE/inbox/hyb-oa-backup-auto-20260828T000000Z.partial" "auto-20260828T000000Z"
+make_batch "$ARM_COMPLETE/archive/hyb-oa-backup-auto-20260827T000000Z" "auto-20260827T000000Z"
+python3 -B "$INSPECTOR" --now-epoch 1787875200 arm --inbox-root "$ARM_COMPLETE/inbox" --archive-root "$ARM_COMPLETE/archive" >"$TEST_ROOT/arm-complete.out"
+assert_line "$TEST_ROOT/arm-complete.out" "arm_partial_complete_contract_count=1"
+assert_line "$TEST_ROOT/arm-complete.out" "arm_valid_atomic_archive_count=1"
+pass "完整 partial 与正式原子归档分别按同一三文件身份和哈希合同验证"
+
+PERMISSION_FIXTURE="$TEST_ROOT/permission"
+mkdir -p "$PERMISSION_FIXTURE/inbox" "$PERMISSION_FIXTURE/archive"
+chmod 000 "$PERMISSION_FIXTURE/inbox"
+if command -v runuser >/dev/null 2>&1 && id nobody >/dev/null 2>&1 && [[ "$(id -u)" == 0 ]]; then
+  if runuser -u nobody -- python3 -B "$INSPECTOR" arm --inbox-root "$PERMISSION_FIXTURE/inbox" --archive-root "$PERMISSION_FIXTURE/archive" >"$TEST_ROOT/permission.out"; then
+    printf '[FAIL] permission denial unexpectedly passed\n' >&2
+    exit 1
+  fi
+  assert_line "$TEST_ROOT/permission.out" "probe_category=access_denied"
+else
+  chmod 0700 "$PERMISSION_FIXTURE/inbox"
+  ln -s "$PERMISSION_FIXTURE/inbox" "$PERMISSION_FIXTURE/inbox-link"
+  if python3 -B "$INSPECTOR" arm --inbox-root "$PERMISSION_FIXTURE/inbox-link" --archive-root "$PERMISSION_FIXTURE/archive" >"$TEST_ROOT/permission.out"; then
+    printf '[FAIL] invalid root unexpectedly passed\n' >&2
+    exit 1
+  fi
+  assert_line "$TEST_ROOT/permission.out" "probe_category=root_invalid"
+fi
+chmod 0700 "$PERMISSION_FIXTURE/inbox"
+pass "不可读或不安全根目录明确拒绝且不伪装为不存在"
+
+if rg -n --fixed-strings \
+  -e "$TEST_ROOT" \
+  -e "auto-20260828T000000Z" \
+  -e "database.dump" \
+  -e "storage.tar.gz" \
+  -e "unstructured synthetic transport failure" \
+  "$TEST_ROOT"/*.out >/dev/null; then
+  printf '[FAIL] redacted output leaked fixture identity, path, artifact name, or diagnostic text\n' >&2
+  exit 1
+fi
+pass "stdout allowlist 不含路径、run id、产物名或原始诊断"
+
+PYTHONPYCACHEPREFIX="$TEST_ROOT/pycache" python3 -m py_compile "$INSPECTOR"
+bash -n "$ROOT_DIR/scripts/qa/run-backup-transfer-incident-contract.sh"
+pass "Python 与 Bash 语法通过"
+
+cleanup
+[[ ! -e "$TEST_ROOT" ]]
+TEST_ROOT=""
+trap - EXIT
+pass "合成 fixture 与 Python cache 零残留"
+
+printf '[PASS] backup transfer incident contract: %s checks\n' "$PASS_COUNT"