Auto git update

BUSINESS_BACKUP.md

@@ -53,13 +53,15 @@ The `.gitattributes` override and scoped forced staging prevent repository ignor
 
 Remote snapshots also constrain source freshness: previously backed-up terminals may not disappear without an explicitly corroborated deletion receipt. Their sanitized terminal fingerprints must remain unchanged when retained. This conservatively blocks stale source mounts and unexplained terminal rewrites from becoming newer Git backups; incomplete-provenance tombstones alone do not authorize erasing already backed-up parents. Ordinary profile/publication/presentation edits are captured in the next terminal snapshot, not independently triggered.
 
-## Baseline, discovery and consistent collection
+## Baseline, discovery and single-pass collection
 
 On first enable, capture one complete baseline. Historical terminals covered by that snapshot get receipts together and do not each create a commit. Afterwards, polling runs every 60 seconds and discovers unconfirmed IDs in terminal/archived terminal records with statuses `done`, `done_partial`, `failed` or `cancelled`. Already confirmed submission IDs do not trigger repeated backups; multiple new terminals found together are coalesced into one full snapshot, with receipts for all included terminals.
 
 Only existing Exporter-classified business paths are discovered. GT, predictions, media, weights, standalone credentials, raw task/child/performance logs and runtime control/lease files are not copied. Reading a mixed Registry/queue JSON file necessarily reads bytes that may contain authentication/lease fields; existing strict schema cleaning **immediately removes them before writing the offline copy**. Such values are never separately consulted, persisted in the snapshot or logged. The already sanitized private copy is then cleaned/validated again by the unchanged Exporter.
 
-Required source layout must exist. File hashes, consumed bytes and inventory are checked across collection; changed, malformed, unknown or inconsistent data is deferred. No business lock is acquired and no source file is repaired. Export reads only the private local copy, never bypassing the `/data` prohibition. Full verification must pass before staging. Stable terminal/History/Progress relationships are enforced by the existing exporter, not by altering how the Worker produces them. This is bounded best-effort consistency, not an atomic multi-file transaction or immunity to all ABA/cache effects.
+Required source layout must exist. Each initially enumerated business file is read once during collection and immediately sanitized into a private local copy. There is no final live inventory/hash reread and no requirement that production remain idle or keep identical Xet hashes, sizes or timestamps. An unreadable/disappearing enumerated file, partial/malformed content, unknown schema or logically contradictory collected records still fail closed; nothing is silently omitted to make validation pass. No business lock is acquired and no source file is repaired.
+
+Export reads only the private local copy, never bypassing the `/data` prohibition. Its local source-stability checks, payload/manifest digests and full business relationship validation remain enabled. Verify also reconstructs every logical row and compares its digest with the collected sanitized row digest before staging. Thus the guarantee is a strictly validated best-effort collection with recoverable Git history, not a precise point-in-time or atomic cross-file transaction. Active records are included, but their final state and every transient intermediate state are not guaranteed. Several submissions or roughly a day of lag are acceptable; the terminal-triggered policy and bounded retries do not promise a hard maximum backup age during idle periods, failures or disabled operation. See `BUSINESS_BACKUP_SINGLE_PASS_VALIDATION.md` for the latest acceptance evidence.
 
 The exact `historical-deletion-evidence-v1` results remain in the manifest and are not rewritten as successful provenance. A backup is evidence, not authorization to recreate missing submissions or scores or restore old operational state.
 
@@ -103,6 +105,12 @@ Supervisor stop/timeout/resource pressure kills only the owned backup process gr
 
 ## Tests, measurements and validation gaps
 
+Production discovery follow-up: see `BUSINESS_BACKUP_DISCOVERY_DIAGNOSIS.md`. Proven operator compatibility-resume audit fields now have an explicit narrow schema; classified discovery diagnostics do not reset persisted failures or retries. The actual terminal passes the patched schema but still fails strict event chronology, so live acceptance is not cleared by this patch alone.
+
+Post-repair follow-up: see `BUSINESS_BACKUP_POST_REPAIR_VALIDATION.md`. Narrow evidence-backed original-start and incomplete-Case reset validation now passes the current offline production copy, including independent reconstruction comparison. Full acceptance remains blocked because production source metadata changed during validation; the earlier discovery report is historical, not the final validation outcome. No deployment, backup-state reset or production mutation was performed.
+
+The post-repair report above describes the previous acceptance policy. The current single-pass policy treats later production metadata changes as informational, not corruption of an otherwise valid collected snapshot; strict collected-content and offline integrity checks remain mandatory. Current findings and deployment prerequisites are recorded in `BUSINESS_BACKUP_SINGLE_PASS_VALIDATION.md`.
+
 Exhaustion-email follow-up: **11 focused checks**, **31 existing backup checks**, **26 existing lifecycle-alert checks** and **49 lifecycle-control checks** pass (**117 checks** rerun for this follow-up). Tests include pre-exhaustion silence, final-confirmation success/failure, restart deduplication, independent operations, configuration refusal, provider uncertainty, persistence-before-send, safe payloads, resolved rate-limit incidents and releasing the backup lock before notification. Brevo is mocked; no live email/API calls or fake production records are used. Existing Stage B/Stage A results below are retained separately, not claimed as newly rerun exporter tests.
 
 Offline tests use synthetic business data and disposable local bare Git repositories, not production GitHub. They cover disabled behavior; baseline and sequential/coalesced triggers; lost receipts and restart recovery; collection/export/push crashes; malformed/unknown/inconsistent sources; incomplete-provenance tombstones; secret canaries; authentication/privacy/destination refusal; retry exhaustion/final confirmation; remote conflicts/source regression; namespace ownership/filters; auxiliary locking and supervisor cancellation; no lifecycle/Worker hooks.

BUSINESS_BACKUP_DISCOVERY_DIAGNOSIS.md

@@ -0,0 +1,73 @@
+# Business backup discovery failure — 2026-10-13
+
+## Confirmed production facts (read-only)
+
+- Baseline remains `false`; confirmed terminal receipts: **0**.
+- A persisted **baseline** operation exists, with **3** attempts, phase `failed`, recorded failure `git_operation_failed`.
+- Destination is pinned, but there is no candidate commit or recorded parent commit. GitHub identity verification was reached in an earlier attempt; the current repeated discovery failure is not proof that GitHub was never contacted.
+- An independent `business_backup_exhausted` receipt already exists: classification `git_operation_failed`, delivery `sent`, tries **1**. This means provider acceptance recorded by the application, not proof of inbox arrival.
+- Current terminal inventory has **28** files; the exhausted operation recorded **27** terminal identities. There is **1** newly completed identity available to the existing retry policy; no evaluation rerun is needed.
+- Read-only Frontend repository source at revision `916e4b4e781faee57131a63b671948a119ec712f` has the same generic `discovery_deferred` branch and lacks the recovery-audit schema/diagnostic patch. This confirms remote source, not independent proof of the currently running build's exact revision.
+- Metadata classification reported no unsupported business paths. Of the inspected terminal documents, **27** passed the pre-patch schema and **1** failed `unclassified_field`. The inspected runtime heartbeat also passed. This is not a claim that the live Frontend FUSE mount cannot separately have I/O problems.
+
+Only supported business metadata, auxiliary receipts, terminal documents and the heartbeat were read. Required downloads used ACL-restricted temporary directories and were removed. No GitHub pushes/API writes, Bucket mutations, deployment, Worker restarts, baseline resets or emails were performed. No private record values or submission identities are included here.
+
+## Exact call chain and root cause
+
+`_run` → `_supervise` → `run_child` → `BackupSession.run` → `export.inventory` → `terminal_map` → `clean(..., SCHEMAS['submission'])`.
+
+The affected legitimate operator recovery audit has three unsupported schema locations:
+
+1. `evaluation_provenance.compatibility_recoveries`: a list of structured compatibility-resume evidence.
+2. `submission_events[]` with event `operator_compatibility_resume`: finite fields `backup_directory`, `previous_evaluator`, `retained_results_digest`, `retained_scenes`, `verification`.
+3. Submission `requeued_at`: a timezone-qualified timestamp.
+
+The strict dictionary cleaner rejects unregistered fields with `ExportError('unclassified_field')`. The failure can first arise while cleaning `evaluation_provenance`; the event's extra fields are another confirmed incompatibility, not the sole first failing condition. Source values are not needed to expose the diagnosis.
+
+`BackupSession.run` catches the exception. Because an older operation already has attempts=3, it returns `discovery_deferred` **before persisting the newly observed discovery exception**. The stored failure therefore stays the older `git_operation_failed`, masking today's `unclassified_field`. The supervisor sleeps 60 seconds after each run; runtime overhead explains observations around 60–65 seconds.
+
+This is not the initial-baseline waiting for a new Submission. A fresh baseline executes without a terminal trigger, but its inventory/schema admission must still pass. A redeployment or Worker run can reveal evolving state and consume finite attempts; merely finishing evaluation cannot repair unknown audit schemas. Repository initialization cannot resolve source schema errors.
+
+The older Git failure occurred before a parent commit was recorded. In this path that is consistent with clone/HEAD resolution failing after identity pinning. Git stderr was intentionally discarded, so its exact underlying cause cannot be reconstructed from `git_operation_failed` alone. The operator's now-initialized private repository is not independently retested here; a successful future clone remains a live acceptance requirement.
+
+## Minimal local correction
+
+Files changed:
+
+- `tools/export_schema.py`: explicitly classify the proven recovery record, its provenance list and `requeued_at`. Recovery evidence is accepted only for the exact recovery event type, with required fields, fixed verification keys, finite prior-evaluator schema, nonempty unique retained-scene IDs, SHA-1/SHA-256 proof formats and timezone-qualified timestamps. Other event types cannot acquire these extra fields. Unknown fields still fail closed; existing credential cleaning applies inside the evidence. Audit metadata does not grant recovery/authorization or verify its historical claims by itself.
+- `src/submission/business_backup.py`: add transient, allowlisted discovery diagnostics for `inventory` versus `terminal_validation`, fixed failure classification, attempt count and `backoff`/`exhausted`. The supervisor logs only those fields. No raw exception, path, user ID or record content is logged. Old persisted failure, attempts, baseline, pending identity and confirmed receipts are untouched by the diagnostic change.
+- Focused additions in `business_export_regression.py`, `business_backup_regression.py`, `business_backup_alerts_regression.py`.
+- This diagnosis document and the short cross-reference in `BUSINESS_BACKUP.md`.
+
+Example safe log:
+
+```text
+Business backup discovery unavailable (step=terminal_validation, failure=unclassified_field, attempts=3, retry=exhausted)
+```
+
+The previous generic `Business backup deferred (discovery_deferred)` remains; it now has a useful classified companion. Runtime Linux/resource/retry integration and Worker code are unchanged.
+
+## Further validation gate — do not re-enable yet
+
+The actual affected terminal document **passes the patched schema** in an isolated local copy. However, its existing event sequence subsequently fails the preserved strict `_events` validation with **`event_order_mismatch`**. A resumed `running` event carries the original first-start timestamp but appears after later recovery events; this is a separate chronology inconsistency. It was not repaired or sorted away by this patch.
+
+Therefore this patch removes the immediate discovery-schema barrier but does **not** establish that a complete live snapshot will currently export successfully. No complete current production snapshot acceptance is claimed. Missing/unordered audit evidence must not be hidden just to make the backup pass.
+
+Minimum additional action, under separate authorization: inspect/correct that one audit chronology with evidence-preserving provenance, or separately review the event producer that reused the original start timestamp. Do not weaken the exporter's ordering check, skip records, rerun GPU scoring or delete backup auxiliary state. This task explicitly forbids production mutation and Worker changes, so neither is performed.
+
+## Retry and alert coverage
+
+- Discovery failures create a pending operation when absent and consume up to three execution attempts, with existing backoff. Repeated exhausted polls do **not** consume further attempts.
+- The generic warning may repeat indefinitely while source admission keeps failing. This is repeated observation, not infinite Git pushes.
+- After each normal child return, `notify_exhaustion` runs outside the backup lock. Even `discovery_deferred` is eligible once persisted attempts=3 and no uncertain candidate needs final confirmation. Persistent operation UUID deduplication prevents repeated accepted emails.
+- Actual production already has one sent exhaustion receipt. A new alert mechanism is unnecessary. Disabled/missing Brevo configuration or ambiguous delivery may prevent a delivered email; that does not clear backup exhaustion.
+- Current failed discovery returns before GitHub transport. Earlier operation reached GitHub identity/Git. Once schema and chronology are resolved, the **existing** policy can create a new bounded baseline operation because one genuinely new terminal identity is present, without manually resetting the baseline or reevaluating the Submission.
+- If no new terminal identity were present, re-enabling alone would not reset an exhausted operation. Any deliberate retry then requires separately authorized, lock-protected operator review, retaining candidate/receipt evidence.
+
+## Validation and deployment decision
+
+Focused tests cover recovery-audit roundtrip with nested credential sanitization, wrong-event/unknown-field/malformed-proof refusal, unchanged exhausted receipts while reporting the current discovery error, private errno classification, supervisor log sanitization, and three discovery failures producing one deduplicated alert without Git calls.
+
+Fresh offline suites passed: **49 exporter**, **34 backup**, **12 backup-alert** and **26 lifecycle-alert** checks: **121 total**, including **6 new focused checks**. The final focused schema/log checks were additionally rerun after the last edits. Compilation, Ruff and `git diff --check` passed; only the existing Ruff configuration deprecation warning remains. No production backup execution or actual email is part of these tests.
+
+**Re-enable decision: NO-GO for successful backup acceptance yet.** Review the source event-order gate first, then deploy the local correction under separate authorization, preserving existing auxiliary state. The existing new-terminal policy can retry the baseline naturally; complete Export/Verify and Git confirmation must still pass. No Worker restart or scoring rerun is required to include the finished Submission.

BUSINESS_BACKUP_POST_REPAIR_VALIDATION.md

@@ -0,0 +1,130 @@
+# BUSINESS_BACKUP_POST_REPAIR_VALIDATION
+
+日期：2026-10-13。最后一次只读采集开始于 **08:10:07 UTC**。目标以内部 SID 定位，本文不输出账号、Case 名、原始业务记录或私有数据路径。
+
+## 1. 最终结论与停止边界
+
+**目标修复记录一致；本地兼容性修正通过真实数据 Export → Verify → Reconstruct 和独立逻辑记录比较。但完整验收为 NO-GO：生产业务文件在验证期间发生变化，源稳定性条件未满足。**
+
+剩余阻塞为 `immutability / snapshot_sources / production_source_changed_during_validation`，不是 schema、事件顺序或重建错误。最后一轮前后清单无文件增删，但 **1 份 Submission、1 份 Progress** 的 Xet hash/大小/时间元数据比较发生变化。不能认定获得了完整稳定生产快照，不能将离线通过当作首次 GitHub baseline 放行。
+
+恢复决策选择 **A：现有辅助状态具备自动创建新一轮有限 baseline 操作的资格**。这不代表可以绕过稳定快照验收、GitHub 凭据及 Linux 环境检查。
+
+**平台状态与背景不符：** 早先只读 metadata 返回 Worker `SLEEPING`，最后一次返回 **`RUNNING`**，不是背景中的 `PAUSED`。本轮没有请求 resume/restart，也没有访问 Worker app 唤醒它；无法确定谁或哪项机制启动了它。不得声称 Worker 一直暂停。本轮停止云端调查，不为获得通过结果暂停 Worker 或无限重采样。
+
+没有生产数据/辅助状态写入、评分重算、GPU 评测、部署、应用代码提交、真实邮件或 GitHub 推送。受限临时原始副本及本轮缓存已清理，只保留脱敏统计、本地代码和报告。
+
+## 2. 当前目标修复核对
+
+- 恰好 **1 份权威 Terminal、1 条对应 History**，均为 `done`。
+- Progress **103 个成功 Case、0 个失败 Case**，当前阶段/场景为空；无目标 Pending/Processing 残留。
+- Terminal/History 的分数、指标、完成数、provenance 一致；分数 **3.254874**。
+- 独立纯聚合使用现有 Case 指标，结果一致；不计算 GPU 指标。
+- 保留的 **80 个 Case** 与恢复审计 retained results digest 一致；其余 23 个结果未在本轮重算或修改。
+- 原失败、兼容性恢复与状态修复审计仍存在，append 顺序保留。
+
+上述结论适用于读取的隔离副本及其交叉关系。全量稳定性检查失败，因此不承诺所有在线记录在整个验证期间静止不变。
+
+## 3. 本地最小修正
+
+### 恢复 schema
+
+此前未部署补丁明确分类 `compatibility_recoveries`、`operator_compatibility_resume`、`requeued_at`、旧 evaluator、保留结果 digest 和 verification。本轮补充 `reset_incomplete_case_starts`，仅允许显式恢复事件中的 **Case ID → 带时区原开始时间** 映射。未知字段、缺失证明、矛盾记录仍 fail closed，不开放任意字典。
+
+### 原开始时间引用
+
+Worker `claim_one` 保留 `started_at` 并以它生成 resumed running timestamp：append 位置是追加顺序，operator timestamp 是恢复发生时间，后续 running timestamp 是原评测开始时间引用。
+
+`tools/business_export.py::_recovery_start_reuse` / `_events` 仅接受完整成功 `done`，紧邻 failed → operator resume → running，同一提交与后续 worker attempt，有唯一原 `running:1`，重用时间等于 started_at，恢复时间等于 requeued_at，原开始 ≤ 失败 ≤ 恢复，并有唯一且内容一致的 provenance 副本。Verify 另核对 retained Case 与实际 Progress digest。
+
+接受引用时不降低 occurrence-time watermark。随后普通时间回退仍拒绝；不按特定 SID 开后门。
+
+### 明确重置的未完成 Case 复用 started ID
+
+真实 Progress 中一个未完成 Case 在恢复前后使用同一 started 事件 ID，attempt 都是 1，仅时间/Worker actor 不同。生产者 `_run_scene_with_inline_retry` 在开始时间被清除后仍使用固定 started ID。
+
+新增 `_reset_case_start` 只认可有明确 reset 映射的这种复用：原 started 在恢复前，新 started 在恢复后且不晚于成功结果；新时间等于 Progress `case_started_at`；Case 有成功结果、无失败且不在保留完成列表中；两个事件除时间/actor 外完全一致。
+
+Verify 要求 Progress 恢复证明与 Terminal 已验证证明的核心字段相同；若 Terminal 也含 reset 映射，映射也须相同。**Case attempt 单调检查未放宽，2 → 1 仍失败。** 无证据的冲突 ID、普通/Case 时间回退仍拒绝。两个原事件在重建中保留，不排序、去重或改写数组。
+
+### Digest 合同
+
+恢复审计 retained results digest 使用 **排序键 compact JSON、无尾部换行**；manifest 的既有 `record_digest` 使用不同规范格式。新增 `_retained_results_digest` 严格按原审计合同计算，不改 manifest、生产 digest 或评分。一次中间诊断误用 manifest 格式出现 retained mismatch，修正离线计算格式后通过；不是生产成绩变动，也没有采用多格式宽松 fallback。
+
+### 实际文件范围
+
+- `tools/export_schema.py`：恢复事件可选 reset 映射。
+- `tools/business_export.py`：原开始引用、受证据约束的 Case started ID 复用、审计 digest 及跨记录核对。
+- `business_recovery_export_regression.py`：14 项隔离恢复测试。
+- 本报告和 `BUSINESS_BACKUP.md` 状态交叉引用。
+
+此前未提交的 schema/discovery 诊断及相关测试补丁继续保留，部署时需一起审查。**Worker 工作树无改动**；评分、GPU、Queue/Lease/Recovery 执行逻辑未修改。
+
+## 4. 最后一次完整副本验收
+
+| 项目 | 结果 |
+|---|---|
+| Inventory / 严格 Schema | 通过 |
+| 业务原文件 | 239 份，2,711,135 bytes |
+| 清洗后源逻辑行 | 327 |
+| Progress 成功 Case 记录 | 1,313；含其他提交，不等同于独立 benchmark Case 数 |
+| Export | 通过，8.250 秒，474 个 payload 文件 |
+| Verify | 通过，3.516 秒 |
+| Reconstruct | 通过，7.703 秒 |
+| 独立逻辑记录 SHA-256 比较 | 327 行一致，无丢失/多余 |
+| 离线原件 SHA-256 / mtime | 不变 |
+| 生产 Xet hash / size / mtime / uploaded_at | **失败：Submission 1、Progress 1 变化；无增删** |
+| 完整验收 | **NO-GO：源稳定性未满足** |
+| 私有临时副本清理 | 完成 |
+
+源逻辑行：migration 1、publication request 43、publications 1、registry 1、submission 30、deletion request 100、progress 25、tombstone 29、history 28、metadata 30、audit 39。Registry/Migration 导出会拆成多个 payload；独立比较按原文件/ordinal 重建，不用 payload 数代替源记录数。
+
+仅采集既有业务分类根。GT、预测 ZIP、模型、raw task/performance/child 日志、独立凭据不导出。Heartbeat 核对固定格式后作为 runtime 排除；备份辅助状态单独只读核对。混合业务文件按既有规则清洗。
+
+source 256MiB、payload 5MiB、每 chunk 200 Cases、manifest 32MiB 边界不变。`historical-deletion-evidence-v1` 本轮计数：**retained parent 20、standalone deletion verified 0、standalone provenance incomplete 9**；父记录缺失保留 explicit-not-restorable，不虚构或推断授权。
+
+前一轮也通过导出/重建与 323 行独立比较，但生产稳定性失败。最后一轮用于定位变化类别并完成辅助检查，没有移除失败门槛。阶段耗时仅代表离线验证，不是 live backup SLA。
+
+## 5. 新鲜回归结果
+
+| 脚本 | 通过 | 失败 |
+|---|---:|---:|
+| `business_export_regression.py` | 49 | 0 |
+| `business_recovery_export_regression.py` | 14 | 0 |
+| `business_backup_regression.py` | 34 | 0 |
+| `business_backup_alerts_regression.py` | 12 | 0 |
+| `lifecycle_alerts_regression.py` | 26 | 0 |
+| `lifecycle_regression.py` | 49 | 0 |
+| **合计** | **184** | **0** |
+
+103-Case fixture 包含 80 个保留/23 个后续结果、旧失败、原开始引用、显式 reset 与修复事件。覆盖完整重建/append 顺序、普通 chronology / Case attempt 仍严格、缺失/矛盾证据、digest 格式误用、冲突 ID 正反例、tombstone 策略不变。
+
+既有备份回归覆盖耗尽 baseline 遇真正新终态后的有限新操作、候选不确定时确认而不重复推送、收据/告警去重、诊断不重置旧失败。邮件用 mock，Git 仅在一次性本地 bare fixture，不操作生产仓库。去掉新测试无用赋值后 14 项再次通过，不重复计入总数。
+
+Compilation、schema/新测试/backup 模块 Ruff、`git diff --check` 通过。完整 Exporter Ruff 存在 **HEAD 已有** `F401: REGISTRY imported but unused`，未顺便修改；既有 Ruff 配置弃用 warning 保留。
+
+## 6. 当前备份状态与恢复决策
+
+只读辅助状态：`baseline=false`，旧 pending trigger=baseline，attempts **3**，failure=`git_operation_failed`；无 candidate、expected parent、confirmed head，receipts **0**。目标不在旧 pending terminal inventory；相对旧操作有 **1 个真正新 terminal identity**。Pinned destination 与指定 repo 一致。
+
+**选择 A：自动重试有资格，无需人工清空 attempts / baseline。** 既有 `BusinessBackup` 在成功 discovery、旧 next_try 到期、耗尽且没有候选提交、发现新 terminal identity 时，可创建新的有限 baseline 操作。资格不等于已执行；本轮没有启动、reset 或触发告警。
+
+保留旧 operation、pinned destination、收据、历史失败与告警去重证据。若将来 candidate 不确定或状态冲突，先只读核对 candidate/parent/manifest，再另行授权锁保护维护。无 interactive terminal 的普通 Space 不能以直接覆盖 JSON 代替安全恢复。
+
+HF Frontend 源 revision 为 `916e4b4e781faee57131a63b671948a119ec712f`，读取的 Exporter **没有本轮 `_recovery_start_reuse`**；源 revision 不等同于证实的运行 build revision。
+
+## 7. 剩余前提与下一步
+
+Repo `vLAR-group/PhysInOne-Leaderboard-Backup` 的 private、README/main 初始化由操作员报告。没有可用指定 GitHub PAT；匿名 metadata=404，不能据此确认或否定私有 repo 存在、初始化或权限。HF 凭据未发送到 GitHub。
+
+尚未验证 PAT effective push、组织批准、分支规则、Git transport、生产 Git 可执行程序、Linux `/tmp`、FUSE 锁/超时、首次持久收据。不能仅凭 README 初始化认定旧 Git 错误已修复；Export 成功不等于 Push 成功。
+
+**本地修正可进入部署审查，但首次受控 baseline 验收暂不能放行。** 最小下一步：
+
+1. 操作员确认平台为什么为 RUNNING；本轮不暂停/重启或影响生产任务。
+2. 另行确认允许的稳定业务窗口，重新做一次完整只读验收，连同源 hash/时间稳定性一起通过。不能以本次离线通过替代该门槛，也不能删除历史证据。
+3. 管理员核实私有 repo main 初始化、仅该 repo 的 fine-grained Contents Read/Write PAT、组织批准和普通直接提交权限。
+4. 另行授权部署审核后的 Frontend schema/Exporter/既有诊断补丁；保留 auxiliary state，按 **A** 使用既有有限重试，不部署/启动 Worker、不改评分参数。
+5. 首次 baseline 核对 committed snapshot Verify、manifest digest、remote SHA、persistent receipt；失败按分类诊断，不 force-push、不删收据。必要时另行授权锁保护维护。
+
+报告交付后停止，不部署、不启用备份、不推送、不发真实邮件、不修改生产记录。

BUSINESS_BACKUP_SINGLE_PASS_VALIDATION.md

@@ -0,0 +1,112 @@
+# BUSINESS_BACKUP_SINGLE_PASS_VALIDATION
+
+日期：2026-10-13。范围：最小本地单次采集政策修正、隔离回归、授权只读生产验收。未部署、推送、修改生产数据/凭据/辅助状态、发送真实邮件或操作 Worker。
+
+## 1. 实际实现：三种检查不同
+
+**原生产代码确有源稳定性门槛，不只存在于前次审计脚本。**
+
+| 层次 | 原行为 | 本轮决定 |
+|---|---|---|
+| `src/submission/business_backup.py:127` `collect` | 初始 inventory；预读每文件 hash；清洗读取时比对 hash；末尾重新 inventory/回读全部 live hash，否则 `source_changed_during_collection` | 删除 live 预读 hash 和末尾稳定性回读；保留初始完整分类、布局/资源限制及一次内容读取 |
+| `_hf_audit/post_repair_validate_readonly.py` | SDK 下载一次；离线验证后比较生产 Xet hash/size/mtime/uploaded_at，变化使整体验收失败 | 前后生产 metadata 仅观察/统计，不阻止有效副本通过 |
+| `tools/business_export.py` `export` | 对输入文件重复 hash/inventory、验证业务关系及输出 | **不删除**：生产调用输入已经是私有本地清洗副本，不是 live Bucket |
+| `tools/business_export.py` `_verify` | 校验 payload/manifest、关系，并重建逻辑行核对原清洗行 digest | **完整保留**，不是生产源是否仍更新的比较 |
+
+`business_backup_git.py:127` `commit` / `:135` `push` 的私有 repo 验证、限定 namespace、普通提交、expected parent、远端确认均未修改。此次无需改变 GitHub 布局、评分或 Worker 架构。
+
+## 2. 最小改动和单次语义
+
+- `src/submission/business_backup.py::collect`：初始分类清单中的每份业务文件只作一次清洗内容读取，立即写私有副本。取消跨采集窗口的 live hash/inventory 稳定性要求；不锁业务操作，不排除 active 记录。
+- `tools/business_export.py:225` `source_records`：`expected_hash=None` 允许生产 collector 无预读 hash；其余现有调用显式传 hash，仍保留 digest 校验。新增已知 JSON 短读/长度异常、JSONL 未读完打开时已知字节数的拒绝 `source_read_incomplete`，防止短读即使返回可解析记录也被接受。这是单份文件读取完整性检查，不是全生产跨窗口稳定性门槛。
+- `_hf_audit/post_repair_validate_readonly.py::main`：生产 metadata 变化不决定成功与否；离线原件 SHA-256/mtime 不变、strict schema、全量关系验证及独立重建比较仍为硬门槛。聚合输出分开记录，保留上轮审计结果文件。
+- `business_backup_regression.py`：替换旧 source-change 拒绝测试，新增四项针对性测试，详见下方。
+- `BUSINESS_BACKUP.md` / 本报告：更新保证范围；旧 post-repair 报告作为当时政策下的历史记录保留。
+
+文件初始列出但读取前消失，仍失败；分类扫描/内容读失败、截断/无效 JSON、未知字段、矛盾 Terminal/History/Progress 均失败。不存在“丢了就当作已授权删除”或为了通过而忽略记录的分支。
+
+“单次”限定于 `collect` 的内容采集。触发判断所需的 `terminal_map` 只读发现仍保留；不宣称整个辅助会话从不读取某个终态文件两次。取消的是二次全量内容获取/稳定性回读，而非业务发现和本地完整性校验。
+
+生产后续流程仍为：清洗副本 → Export → Verify → 受限 namespace staging → commit/push/确认/收据。Verify 本身在本地重建每一逻辑行并比较清洗行 SHA-256；审计额外执行物理隔离 Reconstruct 和独立源行比对，没有为了多一次验证而修改生产 Git 机制。
+
+**保留所有既有强校验：** 恢复 schema、原开始时间引用正向证据、Case reset 证据及 attempt 单调性、80 个 retained 结果 compact JSON digest、append 顺序、秘密清洗、tombstone 分类、终态/进度/发布/删除关系、manifest/payload 校验均未放宽。
+
+资源边界保留：每源记录 256MiB、采集预检查总量 512MiB/5000 文件、payload 5MiB、chunk 200 Cases、manifest 32MiB、aux state 8MiB、workspace 2GiB、独立进程/600 秒墙钟限制和空闲磁盘检查。不引入新基础设施。
+
+## 3. 保证和局限
+
+提供 **严格验证内容的 best-effort 生产业务集合，Git 历史可恢复，接受备份落后**。源在读取前后自然更新，不单独构成已采集内容损坏。
+
+不保证所有文件来自同一精确时刻、全局可串行化或跨文件原子事务；不保证 running 提交已为最终状态，或每个瞬间状态都被记录。混合版本只要违反现有业务模型就拒绝，不能因允许过时而接受矛盾完成记录。
+
+完整快照仍存于 `physinone-business-backup/current/`，由 Git 复用历史对象；不是应用增量备份。已确认终态不能未经佐证删除而消失，已保留终态 fingerprint 不得不明变化；该远端防回退机制不变，因此也不是任意旧副本都可覆盖较新已确认备份。
+
+���许落后几个提交或大约一天。现有终态触发和有界重试**不是硬性“一天内必备份”SLA**：禁用、长时间无新终态或持续失败可以更久；没有偷偷增加定时强制备份功能。
+
+## 4. 新鲜离线回归
+
+五个本轮针对性测试：
+
+1. 在采集内容读完、整体采集尚未结束时更新 live metadata，旧的有效副本仍确认；重建证明保存采集版本，而非较新版本。
+2. 一次 inventory，每业务文件一次 `source_records`，禁止 live hash 回读；副本仍通过 Export/Verify。
+3. 初始分类文件读取前消失，拒绝而非默认为删除。
+4. 已知短读返回 `{}` 或合法 JSONL 审计行，仍拒绝；JSON/JSONL 两个子例均测试。
+5. 已有确认备份后，新 Terminal/History 分数矛盾，拒绝；旧 remote head/confirmed head/receipts 不变，无新 push 或假成功；旧快照仍可 Verify。
+
+现有测试继续覆盖 malformed/truncated/重复键 JSON、未知 schema、Case/终态关系、103-Case 恢复的 80+23 结果、审计 append 顺序、tombstone、有限重试、candidate 确认、去重邮件及独立 lifecycle。
+
+最终代码的新鲜六套回归如下，均 exit 0：
+
+| 脚本 | 通过 | 失败 |
+|---|---:|---:|
+| `business_export_regression.py` | 49 | 0 |
+| `business_recovery_export_regression.py` | 14 | 0 |
+| `business_backup_regression.py` | 38 | 0 |
+| `business_backup_alerts_regression.py` | 12 | 0 |
+| `lifecycle_alerts_regression.py` | 26 | 0 |
+| `lifecycle_regression.py` | 49 | 0 |
+| **合计** | **188** | **0** |
+
+针对性测试先行运行及补测不重复计入总数。Compilation、新修改模块 Ruff 与 diff whitespace 检查通过；完整 Exporter 原有 unused REGISTRY 和 Ruff 配置弃用 warning 未顺便修改。告警为 mock，Git 为一次性本地 bare fixture，未访问生产 GitHub 或真实邮件接口。
+
+## 5. 新政策下真实生产只读验收
+
+采集开始 **2026-10-13 08:49:26 UTC**，SDK 各业务原文件只下载一次到受限临时目录。后续所有内容回读/导出均针对离线副本，不再次获取 live 全量内容。
+
+| 项目 | 实测结果 |
+|---|---|
+| Inventory / Schema | 通过，239 个业务文件，2,719,603 bytes |
+| 清洗源逻辑行 | 327 |
+| 成功 Case 记录 | 1,320，含其他提交，不等同于独立 benchmark Case 数 |
+| Export | 通过，8.329 秒，474 个 payload 文件 |
+| Verify | 通过，3.546 秒 |
+| 隔离 Reconstruct | 通过，7.954 秒 |
+| 独立逻辑记录 SHA-256 比较 | 327 行全一致，无丢失/多余 |
+| 离线源 SHA-256 / mtime | 不变 |
+| 生产 metadata | **1 份 Progress 变化**，无文件增删；信息项，不是 corruption |
+| 新政策验收 | **PASS** |
+| 临时原始副本/本轮缓存 | 已清理 |
+
+Tombstone 策略仍 `historical-deletion-evidence-v1`：retained parent 20、standalone deletion verified 0、standalone provenance incomplete 9；没有虚构 missing parent 或推断授权。
+
+修复目标仍唯一 Terminal/History `done`，103/103 成功、0 失败、分数 3.254874；保留 80 个结果 digest 匹配。平台 Worker 为 RUNNING；本轮未暂停或唤醒它。该通过结论不宣称集合为全局精确 point-in-time。
+
+## 6. 重试、告警及现有 baseline
+
+移除 `source_changed_during_collection` 这条生产 live 稳定性失败路径；文件/schema/关系真实失败仍按既有分类失败。离线副本若改变，Exporter 的 `source_changed` 仍为真实完整性失败，不能一并删除。
+
+baseline、终态触发、最多 3 次、60/300/1800 秒退避、candidate/parent/remote 确认、durable receipt/operation、Brevo `business_backup_exhausted` 去重和 Worker 生命周期未改。没有新增邮件分类、reset 历史状态或不必要重复推送。
+
+本次辅助状态只读：baseline=false、旧 baseline attempts=3、failure=git_operation_failed；candidate/parent/confirmed head 均无，receipts=0。Pinned destination 匹配指定 repo；相对旧操作仍有 **1 个真正新 terminal identity**。
+
+**自然重试资格仍成立。** 部署纠正代码、discovery 成功、既有 next_try 到期后，可按既有机制创建有限新 baseline 操作；不需要手工清空 attempts 或收据。本轮未执行此重试、修改 operation 或触发告警。
+
+## 7. GO / NO-GO 和下一步
+
+**数据采集/导出政策：GO；可进入审核后 Frontend 部署和受控首次备份。** 不再需要 Worker idle 或全量生产 metadata 静止。当前 HF Frontend 源 revision 仍为 `916e4b4e781faee57131a63b671948a119ec712f`，没有先前恢复规则，不能假定本地改动已部署。
+
+**GitHub end-to-end 成功：尚未证明。** 指定私有 repo 的 main/README 初始化是操作员提供信息；此前无指定 PAT 的匿名 metadata 404 不能验证其私有状态/权限。没有测试 PAT effective push、组织批准、分支规则、Linux `/tmp`/Git/FUSE/收据写入。旧 git_operation_failed 不能由 Export PASS 自动判定已修复。
+
+最小步骤：管理员核实指��� repo 和受限 PAT/组织批准；另行授权部署审核后的 Frontend 改动，保持 auxiliary state，按现有资格运行一个受控 baseline。核对 committed snapshot Verify、manifest digest、remote SHA 和 persistent receipt。若失败保留分类诊断，不 force-push、不删除历史或假成功记录。
+
+本轮交付后停止：不部署、不启用、不推送、不改凭据/辅助状态、不发真实邮件、不操作 Worker。

EMAIL_AND_BUSINESS_BACKUP_READINESS.md

@@ -0,0 +1,210 @@
+# EMAIL_AND_BUSINESS_BACKUP_READINESS
+
+审计日期：2026-10-13。仅源码审查、匿名只读元数据尝试、隔离离线测试和本报告；未部署、启用开关、修改 Secrets、发送邮件、访问生产业务文件、推送 GitHub 或操作 Worker。
+
+## 1. 执行结论
+
+| 决策 | 结论 | 边界 |
+|---|---|---|
+| 可以部署关闭功能的 Frontend 代码？ | **YES** | 按下述流程，备份、自动告警、测试开关均保持关闭；只部署 Frontend，另行授权并选择合适窗口。 |
+| 可以启用两套系统做完整受控线上验收？ | **NO（目前）** | 邮件只有容器 CLI 测试入口，普通无终端 Space 无法执行邮件 smoke test；线上配置、版本、私有仓库权限与运行时均未确认。GitHub 基线可在这些条件核实后单独条件性验收，但不等于邮件验收通过。 |
+| 可以无人值守生产运行？ | **NO** | 尚未真实验收；GPU 旧待发告警可能在事件解除后发送；耗尽恢复依赖授权文件锁操作，普通 Space 没有管理入口。 |
+
+**本地实现存在不等于已经部署。** 审计开始时 Frontend 和 Worker 工作树均干净；Frontend HEAD 为 `916e4b4e781faee57131a63b671948a119ec712f`，提交时间 `2026-10-13T13:46:35+08:00`。先前“未提交”的状态已不适用于本次审计。这个 SHA 不能证明 Space 当前源版本或正在运行的构建版本。
+
+## 2. 实际检查范围和入口
+
+| 文件 / 函数 | 核实范围 |
+|---|---|
+| `app.py:40`, `app.py:1427` | 导入备份模块；FastAPI 启动 `[start_controller, business_backup.start]`，关闭 `[business_backup.stop, stop_controller]`。没有测试邮件 UI / HTTP / 启动触发入口。 |
+| `src/submission/evaluator_space.py:47`, `:418`, `:446`, `:456` | `configured`、观察投递、`start_controller`、`stop_controller`；沿用原生命周期控制，不是备份代码。 |
+| `src/submission/lifecycle_alerts.py:32`, `:64`, `:88`, `:132` | `_mail_config` / `_config`、`_send_brevo`、`observe`、`Notifications.load/consume/deliver`。 |
+| 同文件 `:293`, `:315`, `:393`, `:440` | 独立 `BackupNotifications`、耗尽告警、`send_test_email`、CLI `main`。 |
+| `src/submission/business_backup.py:35`, `:95`, `:113`, `:163` | 开关、终态发现、脱敏离线采集、`BackupSession.load/run/success/owned_snapshot`。 |
+| 同文件 `:377`, `:392`, `:425`, `:433`, `:466` | 本地临时空间、受限子进程、处理后通知、进程监督和轮询。 |
+| `src/submission/business_backup_git.py` | `GitRemote.identity/environment/clone/contains/commit/push`；私有身份校验、认证、普通 Git 推送。 |
+| `tools/business_export.py:143`, `:173`, `:225`, `:440`, `:762`, `:924`, `:966` | 分类、清洗读取、关系验证、Export / Verify / Reconstruct。 |
+| `tools/export_schema.py` | 显式业务 schema、秘密/临时字段排除、自由文本和 URL 清洗。 |
+| `src/envs.py`, `README.md`, `requirements.txt`, `packages.txt` | `/data` 默认路径、Gradio 5.43.1、httpx 依赖、系统 `git` 依赖声明。 |
+| 九个回归脚本、`BUSINESS_BACKUP.md`、三份 Stage A 报告、`OPERATIONAL_AUXILIARIES.md` | 测试真实性、历史证据与操作说明。 |
+
+Worker 只核对 Git 状态，没有修改或执行 GPU 评测。以下源码行号基于本次本地版本。
+
+## 3. 邮件实现与实际限制
+
+### 3.1 Brevo 传输
+
+唯一实际传输为 HTTPS `POST https://api.brevo.com/v3/smtp/email`，`api-key` 请求头使用配置值；发件人和单一收件人来自配置，不能由普通用户指定。没有 Resend 回退请求。旧 `provider=resend` 收据只是迁移后保留且停放，不重新发送。
+
+连接超时 10 秒，其余 HTTP I/O 超时 20 秒，拒绝重定向。它们是 I/O 超时，不是整次调用严格 20 秒总时限；响应体没有独立大小上限。备份告警另受备份子进程总限制约束，GPU 邮件线程不具备相同进程沙箱。异常只记录类型，响应正文不入日志，message ID 经过验证并拒绝已知凭据前缀/当前密钥。缺失或非法配置直接不启动/不发送，不改变主业务。
+
+`201` 记 `sent`（API 接受，不保证进收件箱）；`429` 是明确拒绝，最多三次投递，等待 60 / 300 秒；5xx、超时、意外 2xx 记 `unknown`，不自动重发。其他失败记 `failed`。所有实际请求之前先持久化 `unknown`，降低重启后重复发送风险，代价是进程在“持久化之后、发送之前”退出也不重发。待发邮件 23 小时过期。
+
+### 3.2 四类事件
+
+| 类别 | 精确触发与去重 | 重启 / 解除行为 |
+|---|---|---|
+| `restart_budget_exhausted` | `observe` 接收控制器状态 `blocked=exhausted`；`consume` 创建随机 32 位十六进制事件 ID，类别 latch 持久化。不是邮件代码自己判定 3+3 或修改预算。 | latch 保留，重复观察不重建；预算明确恢复时解除 latch。旧待发事件不会随 latch 一起取消。 |
+| `starting_30_minutes` | 连续新鲜 `APP_STARTING` / `RUNNING_APP_STARTING` 观察达 1800 秒；最大连续间隔 45 秒；正常控制轮询间隔 15 秒。 | 启动计时不跨 Frontend 重启延续，缺口/未知状态重置；已有事件 ID/latch 持久化。离开 Starting 解除 latch，但不取消旧待发事件。 |
+| `unexpected_stop_or_runtime_error` | 观察新鲜且 active=true；`RUNTIME_ERROR` 直接符合；`STOPPED` 还要求前一连续观察为 active 的 RUNNING/Starting。并不证明容器确实崩溃。 | 初始 STOPPED、空闲、PAUSED/SLEEPING 不作为该事件；RUNNING 且 attempts=0 可解除 latch，旧待发邮件仍可能送出。 |
+| `business_backup_exhausted` | 备份 pending.attempts=3；没有候选 commit 时可通知；有候选 commit 必须先走最终确认，`final_confirmation` 已置位且没有确认成功。UUID operation 为持久事件 ID。 | 使用独立 outbox，只投递当前耗尽 operation；确认成功清除 pending，抑制告警；新 operation 独立通知，不重放旧 operation。 |
+
+GPU 收据为 `/data/auxiliary/lifecycle-alerts/notifications.json`；备份邮件为 `/data/auxiliary/business-backup-alerts/notifications.json`，锁和事件独立。前两次备份失败不告警，第三次 uncertain 不立即告警，最后一次只确认、不额外推送。`final_confirmation` 是“已开始最终确认”的持久标记，不证明 API 查到失败；中途退出需人工检查，邮件保留 uncertainty。
+
+**真实缺陷（离线复现）：** `Notifications.consume` 解除 GPU latch 后，`deliver` 只检查 pending/provider/next_try/年龄，不核对事件是否已解除。测试先生成耗尽事件，再输入 RUNNING、attempts=0，最后 mocked deliver：latch 已解除，旧邮件仍发送。Starting 与异常停止事件使用同一机制，也存在待发/429 重试期间的过期通知路径。这不是重复事件创建，也不影响 GPU 控制，但不能声称“解除后不会发送旧告警”。最小修正建议：在现有消费/投递路径停放已解除且尚未发送的对应事件，保留收据，增加 focused 测试；不更改 3+3 策略。本次没有修复。
+
+### 3.3 管理员测试入口：BLOCKED
+
+只有 `python -m src.submission.lifecycle_alerts send-test`。它需要 Frontend 容器执行环境；没有管理员 UI、没有一次性启动触发。仅设置 `EVALUATOR_ALERT_TEST_ENABLED=true` **不会发邮件**。不能在本地运行该 CLI 并声称已经验证“部署 Frontend 的传输”。
+
+CLI 先持久化独立 `/data/auxiliary/lifecycle-alert-tests/test-send.json`，有 60 秒人工调用冷却；它不是跨重启永久单次 request 去重器，60 秒后新的人工调用可再发送。没有自动启动发送，所以也不存在重启自动重复发送。测试不开 GPU/备份真实事件，不由公共用户触发。
+
+普通 Space 无容器终端时，邮件 smoke test 阶段被阻断。最小待授权修正：独立、默认关闭、显式请求的一次性启动测试，固定发件/收件配置，用持久请求身份在发送前去重，重启不重复，未知结果不自动重发。不要把当前 CLI 直接改成“每次启动调用”。本审计不实现替代入口，也不把尚不存在的参数列为现有设置。
+
+## 4. 备份实现核实
+
+- 默认关闭；有效开关为 `true` / `yes` / `1`。启用后首先一个完整基线，基线终态统一生成收据；随后每轮发现新终态 ID，支持四种终态，不按 Case 触发。多个新终态合并为一个完整快照。宕机期间的新终态重新扫描即可发现。
+- Profile / publication 编辑没有独立触发，进入下一次正常完整快照。脱敏终态 fingerprint 被当作不可变证据；异常重写或无完成删除证据的丢失会失败，不默默覆盖旧快照。不能将旧历史重复触发当作新的备份事件。
+- 原 Stage A `export` 仍拒绝直接 live `/data` 输入。辅助代码只读分类来源，先清洗再写受限本地副本，然后调用原 Exporter 和完整 Verify。采集前后核对库存/哈希和消费哈希，不获取主业务锁；这是尽力一致性而非全局原子快照，变化则延期。
+- 显式保留 Registry 身份、旧 Team、账户归属、同意快照、预留与 quota 字段、Submission、Progress/Case 分片指标、History、provenance、presentation、publication、水位/请求、审计及删除证据。阈值仍为源单文档/JSONL 行 256MiB、输出 5MiB、Case 分片 200、manifest 32MiB。
+- 历史策略仍是 `historical-deletion-evidence-v1`，Verify 重算三种计数，不制造缺失父记录或评分。不将备份授权为可恢复 operational 状态。此前 19 / 0 / 9 是 `BUSINESS_EXPORT_TOMBSTONE_POLICY.md` 记载的已接受历史快照，不是本次重新下载的现状统计。
+- 独立 secrets、运行重启文件、GT、ZIP、模型、原始任务/性能/child 日志不进入导出；lease/auth/token 字段被移除。混合业务 JSON 的原始读取必然接触其中的秘密字段字节，清洗后才落盘；heartbeat 为严格格式校验后排��。不能将这说成完全不读取这些混合文件的运行字段。
+- 完整 schema、sanitizer、关系和 SHA 校验经离线测试验证；自由文本清洗不是对任意未知格式秘密的数学保证。导出保留合法个人业务信息，**不是匿名公开数据**，必须保持仓库私有、限制访问者。
+
+### Git 与恢复
+
+`GitRemote.identity` 要求精确 configured full_name、private=true、整数 repo ID、permissions.push=true、非 archived/disabled/fork、有效 default_branch。初始化 README/默认分支是前置条件；空仓库不能 clone 预期 HEAD。需要真实只读元数据确认 PAT/组织审批，token 前缀不能证明所有授权范围，缺少 permission metadata 也会 fail closed。分支 protection/ruleset 是否允许直接推送仍需独立核对。
+
+默认分支/仓库 ID/name 写入 pinned destination；不提供 branch 设置。每次克隆及推送前检查私有身份。Git token 仅在 scoped 环境 HTTP header，不在参数、remote URL、`.git/config`、marker/manifest。Git 子进程不继承 Brevo/HF/ORCID Secrets，空私有 HOME、不使用全局/系统配置/credential helper/hooks/外部协议/重定向/LFS smudge。机器同身份管理员可读取进程环境，不能对其承诺秘密不可见。
+
+只更新 `physinone-business-backup/{backup.json,.gitattributes,current/}`，不删除/暂存其他内容；owned marker/完整导出校验失败拒绝覆盖。`git add --force` 是 scoped 暂存，不是 force-push；实际 push 为普通 fast-forward，检查预期父 SHA。内容版本保留在 Git 提交里，无每事件独立目录。冲突、缺失已知备份、改写历史、不能在 50 提交有界 fetch 内证明祖先都会安全停止。
+
+`/data/auxiliary/business-backup/backup.json` 保存 pinned destination、confirmed head、每终态收据、pending operation、attempts、时间、候选 commit、parent、manifest digest、固定 failure。推送前先保存候选；失败/未知时先确认远端，成功丢本地收据可由验证后的远端 marker 重建，不重复提交。三次执行尝试，正常退避 60 / 300 秒；耗尽后的未知候选有一次最后确认，通常等待 1800 秒。没有无限推送。损坏状态不覆盖。新真正终态可重建已耗尽的无候选 operation；未知候选不会被新终态抛弃。
+
+备份告警在释放 backup.lock 后由受限子进程调用，没有反向 lifecycle 控制依赖、不修改重试/收据、不新推送；缺配置/邮件失败保持原 pending。事件 ID 独立持久化，unknown 投递不重试。独立邮件 outbox 最大 1000 项，拒绝溢出而不是删除去重证据；备份状态最大 8MiB。长期容量达到上限需要人工审查，不能声称无限保留。
+
+Linux 运行：`/tmp/physinone-business-backup`、0700/umask077，拒绝 symlink/FUSE 临时根；启动前要求至少约 3GiB filesystem free。源集合限 5000 文件/512MiB；工作树抽样限 2GiB；child 每进程 AS 2GiB、CPU 180秒、nice+10、禁 core；整轮 600秒，Git 命令 120秒，GitHub metadata 连接10/I/O30秒且响应1MiB。不是进程树总内存硬限额，采样可能短时超额。只杀自有备份 process group；正常/后续启动回收自有 workspace，FUSE/不可中断系统调用仍有 OS 边界。没有改 Worker、评分或 Queue 生产逻辑。
+
+## 5. 配置与部署事实
+
+本地仅检查了环境变量名称和非空状态，未显示值：下表设置在当前进程全部未设置。默认关闭可确认；这不是对 HF Secrets 的缺失判断。没有读取凭据文件或使用缓存 HF/GitHub credential。
+
+匿名只读 HF API/source 页面和 GitHub repo metadata 尝试无法取得数据（web 获取失败；本地匿名 HTTPS 为 ConnectError）。没有取得远端 SHA、代码、配置名称、日志、repo 私有属性或权限；这不证明仓库不存在/公开/未部署。以下全部线上项目为 UNKNOWN，需要 operator 确认：
+
+| 现有设置 | Frontend 类型 / 实际要求 | 本地 | 已部署 |
+|---|---|---|---|
+| `EVALUATOR_BUSINESS_BACKUP_ENABLED` | Variable；初始 `false`，验收批准后 `true` | 未设置，默认关闭 | UNKNOWN |
+| `EVALUATOR_BACKUP_GITHUB_REPO` | Variable；**`vLAR-group/PhysInOne-Leaderboard-Backup`**，不是 HTTPS / `.git` URL | 未设置 | UNKNOWN |
+| `EVALUATOR_BACKUP_GITHUB_TOKEN` | Secret；只选上述仓库的 fine-grained PAT、Contents Read/Write | 未设置 | UNKNOWN |
+| `EVALUATOR_ALERTS_ENABLED` | Variable；初始 `false`；自动邮件明确批准后 `true` | 未设置，默认关闭 | UNKNOWN |
+| `EVALUATOR_ALERT_BREVO_API_KEY` | Secret；有效 Brevo transactional API key | 未设置 | UNKNOWN |
+| `EVALUATOR_ALERT_FROM` | Secret；Brevo 已允许的发件地址 | 未设置 | UNKNOWN |
+| `EVALUATOR_ALERT_TO` | Secret；管理员单一收件地址 | 未设置 | UNKNOWN |
+| `EVALUATOR_SPACE_ID` | Variable；`vLAR/PhysInOne-Leaderboard-Worker`；GPU 告警及现有 CLI test 必需，备份耗尽邮件不必需 | 未设置 | UNKNOWN |
+| `EVALUATOR_ALERT_TEST_ENABLED` | Variable；现有 CLI 测试明确批准时 `true`，其余 `false`；不是启动触发器 | 未设置，默认关闭 | UNKNOWN |
+| `EVALUATOR_MANAGEMENT_TOKEN` 或 `HF_TOKEN` | 既有 controller 的 Secret，至少一个可管理该 Worker；沿用正常现有值，不为邮件/备份额外旋转或扩大权限 | 本审计不读取/核验 | UNKNOWN |
+| `PHYS_DATA_ROOT` | 既有可选 Variable，默认 `/data`；需实际共享业务挂载。不要为审计改变现有路径 | 未设置，默认 `/data` | UNKNOWN |
+
+没有其他现有邮件测试 request 参数，也没有备份 branch 参数。系统依赖 `packages.txt` 的 `git`、`requirements.txt` 的 `httpx>=0.27,<1` 和 Python 环境已在本地确认；本地 Git 为 Windows 2.50.1，不证明 Linux 构建已安装。README 仅声明 Gradio，不证明 Bucket 挂载存在、可读写、锁/fsync 可用。辅助源码要求业务挂载和 `/tmp`，没有在 UI 显示 Git 版本/收据的专用入口。
+
+部署确认须查看 Frontend Files/对应提交与 Build Logs：核对三模块、exporter/schema、`app.py` hooks、`packages.txt` 与本地内容一致且该构建正在运行。HF Settings 核对设置名称/类型/非空，密钥值只由管理员在平台保管；勿截图回传值。运行 `/data` 与 auxiliary 写入、临时目录、Linux 限制，仍须真实基线和只读收据检查，不以“有配置声明”代替。
+
+## 6. Readiness matrix
+
+PASS 仅指明确证据列的范围，不代表线上已通过。
+
+| Component | Status | Evidence | Required action |
+|---|---|---|---|
+| Brevo API transport | PASS | 实际源码 + mocked 请求/回归 | 真实 Frontend transport smoke test 未做 |
+| Three GPU alerts | PARTIAL | 触发/持久去重测试通过；旧 pending 解除后发送已复现 | 无人值守前修正过期事件投递 |
+| Backup exhaustion email | PASS | 最终确认、独立 ID/outbox、失败隔离离线覆盖 | 自然失败时核对真实投递，不造故障 |
+| Administrator test-email entry | BLOCKED | CLI only，无普通 Space 入口 | 待授权补最小一次性测试入口，或独立取得实际容器执行渠道 |
+| Brevo HF configuration | UNKNOWN | 未访问部署 Secrets/Variables | 管理员 Settings 确认 |
+| Backup baseline | PASS | 本地真实代码 + bare Git tests | 受控真实基线未验收 |
+| Terminal-triggered backups | PASS | 四终态、合并、重启扫描、去重 tests | 一个正常终态后的线上快照验收 |
+| Export privacy/integrity | PASS | 新跑完整 exporter suite；历史 Stage A 证据 | 对当前 committed snapshot 再 Verify，保持私有 |
+| GitHub private repository | UNKNOWN | 未取得授权元数据 | GitHub UI/授权只读 metadata 确认 |
+| GitHub PAT permissions | UNKNOWN | 仅源码/mock 校验 | 确认组织审批、只选指定 repo、Contents、effective push 和 rulesets |
+| GitHub HF configuration | UNKNOWN | 本地未设置，不知道线上 | Settings 核对三设置 |
+| Frontend deployment revision | UNKNOWN | 只有本地 HEAD；远端不可读 | Files/revision/Build Logs 确认 |
+| HF Linux runtime dependencies | PARTIAL | git 包与 supervisor 实现存在，本地仅 Windows | Build Logs + 真基线，确认 `/data`、local `/tmp` |
+| Backup restart/crash recovery | PARTIAL | 本地 crash/receipt/conflict tests；实际 Linux/FUSE 未验证 | 授权且安全的 Frontend-only restart 验收 |
+| Deployment rollback readiness | PARTIAL | default-off、持久证据保留步骤存在 | 确認管理权限和无终端时的授权恢复渠道 |
+| Separate test repository | N/A | 本次使用最终指定私有 repo | 不切换 pinned destination |
+
+## 7. 新跑离线测试
+
+本次重新执行九套脚本，**245 项通过、0 项失败**，不是引用历史 234/117 次结果。实际解释器为工作区 `.venv-frontend/Scripts/python.exe`，工作目录为 Frontend。发现的脚本均使用隔离 fixture/mock，不传 real-snapshot 参数，不用生产 GitHub/PAT/Brevo。Git 提交/推送仅发生在一次性本地 bare fixture，不提交/推送项目仓库。
+
+| Executed command（Frontend 目录） | Pass | Fail |
+|---|---:|---:|
+| `python business_export_regression.py` | 47 | 0 |
+| `python business_backup_regression.py` | 31 | 0 |
+| `python business_backup_alerts_regression.py` | 11 | 0 |
+| `python lifecycle_alerts_regression.py` | 26 | 0 |
+| `python lifecycle_regression.py` | 49 | 0 |
+| `python identity_history_regression.py` | 21 | 0 |
+| `python stable_identity_regression.py` | 27 | 0 |
+| `python publication_order_regression.py` | 25 | 0 |
+| `python token_consent_regression.py` | 8 | 0 |
+| **合计** | **245** | **0** |
+
+各 suite 本次耗时分别约 180.296、68.862、12.188、0.244、0.766、0.414、0.262、23.296、0.093 秒；全部 exit=0。不是 GPU 性能数据。
+
+另完成六个运行源码 `py_compile`、五个相关模块/测试 Ruff 和 `git diff --check`。只有既有 Ruff 配置弃用 warning。额外独立 mock probe 直接证实过期 GPU pending 仍发送；它不属于上述 unittest 数量，说明现有绿色测试未覆盖“解除事件后取消旧待发告警”。不新增测试或修复源码。
+
+## 8. 操作员受控验收流程（本次不执行）
+
+### Phase 1 — 安全部署
+
+1. GitHub UI 打开最终 `vLAR-group/PhysInOne-Leaderboard-Backup`：确认 private、非 fork/archived、README 初始化的 default branch 存在；Settings 查看 branch protection/rulesets 允许服务身份普通直接提交。检查组织 PAT approval 已完成，只选此仓库、Contents Read/Write；不要真实 push 来探测权限。检查无 Actions/submodules/不受控 webhook，不能由 Contents-only token 自动证明这些条件。
+2. Frontend Settings 保持 backup=false、alerts=false、test=false。填入上述正确 Variable/Secret；已有正常密钥不必旋转。不触碰 Worker 设置。
+3. 明确另行授权后仅上传/部署当前 Frontend，包括 exporter/schema、backup/Brevo 模块、packages/requirements 和 hooks；核对 Files revision、Build Logs、RUNNING 与普通登录/Queue/History UI。不能只上传一个模块或认为本地提交自动上线。
+4. 选择运营批准窗口。**既有** Frontend controller 在启动可管理 Worker，关闭会核对 active 并调整 sleep policy；辅助代码没新增 Worker 操作，但不能承诺 Frontend 重启完全没有原控制器行为。不要暂停/重建 active Worker，也不要为验收制造 GPU 错误。
+
+### Phase 2 — Frontend 邮件 smoke test：目前阻断
+
+普通 Space 无终端时现在没有可以点击/设置即执行的测试入口。停在此步骤，不能把已完成的 Brevo 手工 API 调用算成集成验收，也不能以制造 GPU 失败替代。需另行批准最小一次性入口后才给该入口的准确操作；本报告不发明当前不存在的设置。
+
+仅在**另行取得授权的实际 Frontend 容器执行渠道**时，现有入口才可用：Settings test=true，保留 Space ID 和三个有效邮件配置，在容器执行 `python -m src.submission.lifecycle_alerts send-test` 一次。核对返回 `api_accepted`、独立 test-send receipt 和 Brevo Transactional Logs；`201` 不证明邮箱到达，需收件箱/投递记录确认。返回 unknown 则查 provider，不能立即盲重发。测试后 test=false。不注册公共 endpoint、不在本地冒充 Frontend。当前 CLI 只能验证单次调用/冷却；没有启动触发器可验收“同一 startup request 跨重启去重”。
+
+### Phase 3 — 初次 GitHub 基线
+
+1. 满足部署/私有仓库/权限和只读收据检查渠道条件后，另行批准 Settings backup=true；alerts 如未做 smoke test 可继续 false，备份自身能独立运行。
+2. 第一轮启动立即异步执行，之后每轮结束等待 60 秒；普通终态发现延迟约一轮间隔再加当前工作耗时。初轮/单轮受 600 秒限制，按“约 0–11 分钟观察单次结果”安排检查，不承诺这个窗口必成功。锁/源变化会延期；失败按 60/300 秒退避，未知候选最后确认通常再等 1800 秒，不要反复重建。OS/FUSE 阻塞不是严格 deadline 保证。
+3. GitHub Files/Commits 核对新提交，namespace 下有 `backup.json`、`.gitattributes`、完整 `current/manifest.json` 和 payload/chunks；无每 Case commit、不删除 README。
+4. 通过事先确认可用的 **授权只读 Bucket 下载渠道** 下载 `/data/auxiliary/business-backup/backup.json` 到受限本地目录，不用容器终端、不修改 Bucket。没有该渠道则“receipt agree”验收仍是 UNKNOWN，不能判 PASS。收据可能含 submission IDs，仅管理员本地核对，不公开原始文件。
+5. 在 GitHub 对应确认提交下载完整快照，仅将 `physinone-business-backup/current` 解到隔离受限目录。Frontend 项目目录、本地 Python 执行 `python -m tools.business_export verify --export "<隔离目录>/current"`。可选 `python -m tools.business_export reconstruct --export "<隔离目录>/current" --output "<新的隔离目录>/reconstructed"`；绝不还原到 live `/data`。
+6. 用本地 SHA-256 工具算 `current/manifest.json`，对齐 repo marker.manifest_digest、持久 last_result.manifest_digest；对齐 GitHub commit SHA 与持久 confirmed_head / last_result.commit（初次没有他人后续提交时应相同）。远端 marker 自身不保存其所在 Git commit SHA。收据 baseline=true、pending=null、last_result.status=confirmed；payload 完整 Verify 通过才认可基线。
+7. 查看聚合历史分类，不要求仍等于 19/0/9；确认不是把 incomplete 标成 verified。只在受限环境查脱敏和完整性，不把个人信息/原记录/Secrets贴报告。确认普通 UI/Queue/evaluation 没有受干扰。处理完删除本地临时业务副本。
+
+### Phase 4 — 正常终态触发
+
+等待一个真实用户正常提交完成，不插假记录、不为审计主动触发评测。等 terminal/History/Progress 一致，核对下一提交、对应终态收据，以及 Verify 后的相应业务/Case metrics。数轮观察不新增相同 snapshot。Profile/publication 不要求独立 commit。
+
+在独立批准且安全的时间安排 Frontend-only restart，验证基线/已确认终态不重复，未观察到的新终态可补发现；不要重启 active Worker。部署动作执行和观测由操作员完成，本次审计不执行。
+
+### Phase 5 — 耗尽邮件
+
+生产不故意令 GitHub 重试耗尽、不给假业务记录。11 个离线 mock 测试是耗尽分支证据，不是实际投递证据。如自然失败，通过授权只读收据核对 operation、attempts=3、failure、candidate commit/parent/manifest；存在候选时还需 final_confirmation/远端状态，不能只看三次就宣告未推送。
+
+核对独立 backup-alerts outbox 中相同 operation 的一条事件、投递状态和 Brevo 日志，邮件只有固定分类/时间/trigger/uncertainty；unknown 不自动重发，429 拒绝可能有有限重试但应至多一个被接受邮件。确认 backup pending 未被邮件改成 success、Submission/GPU restart state 未被邮件修改。若父状态无法读取或缺邮件配置，保持 UNKNOWN/未验收，而非造故障补测。
+
+### Phase 6 — 回滚与耗尽恢复
+
+Settings backup=false 停止后续自动备份；alerts=false 停止自动邮件，test=false 禁用人工测试。由平台执行这些设置生效所需的 Frontend 更新，安排安全窗口；不操作 Worker。保留三类 auxiliary receipts/locks、候选 SHA/parent/digest 和 Git 历史，**不删状态、不 force-push、不改 pinned destination**。再启用沿用去重与候选确认，不能靠删 receipts 强制重新上传。
+
+正常配置/提供方问题先按固定分类查配置与 GitHub 分支规则。已有耗尽 operation 不因普通重启自动重置预算；未知候选须先只读核对远端祖先/完整 marker。人工重新尝试需要先关闭功能，在独立授权且持 backup.lock 的维护渠道下，只调整审核后的 pending attempt/backoff，保留 operation/commit/parent/digest/receipts。**当前无普通 Space 管理 UI；无终端/授权维护渠道时这一步不可执行，需停机人工协助，不可直接网页改 JSON 绕过锁。** 本报告没有实施或建议自动写生产恢复状态。
+
+如需代码回滚，Frontend-only 恢复经审核版本，仍保持辅助功能关闭，不回滚 Worker，不回滚业务数据/Secrets；receipt 保留给将来兼容恢复。代码部署和设置重启均需单独授权。
+
+## 9. 剩余阻断事项（按优先级）
+
+1. **邮件受控 smoke test 的入口缺失**：CLI only 与普通 Space 无终端矛盾；需独立授权最小入口修正或真实容器执行渠道。
+2. **线上部署/配置/权限/挂载未证实**：由操作员核对，不将文档或 PAT 创建当作 effective push/部署证据。
+3. **GPU 告警解除后的旧 pending 投递**：源码确认并 mock 复现；无人值守前建议最小修正；没有影响备份/评分证据。
+4. **耗尽的人工恢复渠道缺失**：锁保护的维护依赖终端/授权运维流程；不阻止关闭功能部署，但不是完整无人值守运维能力。
+
+非阻断边界：HTTP I/O 超时并非严格总时限、有限 outbox/state 容量、非全局事务快照、Linux/FUSE 真实行为待验证、成功备份没有 dedicated UI 状态/常规成功日志（需 GitHub/只读收据观测）。这些不应通过改变业务或扩大架构来“让审计通过”。
+
+**最终：关闭功能部署 GO；现状两系统完整线上验收 NO-GO，GitHub 单独基线条件性 GO；无人值守生产 NO-GO。** 本次仅新增此审计报告，没有修复任何运行源码。

business_backup_alerts_regression.py

@@ -12,6 +12,23 @@ from src.submission.business_backup_git import BackupError, runtime_environment
 
 
 class BackupAlertChecks(unittest.TestCase):
+    def test_discovery_failures_exhaust_once_and_alert_without_git_or_reset(self):
+        from tools.export_schema import ExportError
+        with self.environment(), patch.object(backup.export, "inventory", side_effect=ExportError("unclassified_field")), \
+                patch.object(self.remote, "identity") as remote, patch.object(alerts, "_send_brevo", return_value=("sent", None)) as send:
+            for index in range(3):
+                self.run_backup()
+                self.notify()
+                self.now += 2000
+                self.assertEqual(send.call_count, 1 if index == 2 else 0)
+            before = (self.directory / "backup.json").read_bytes()
+            for _ in range(2):
+                self.assertEqual(self.run_backup(), "discovery_deferred")
+                self.notify()
+            send.assert_called_once()
+            remote.assert_not_called()
+            self.assertEqual((self.directory / "backup.json").read_bytes(), before)
+
     setUp = fixtures.BackupChecks.setUp
     state = fixtures.BackupChecks.state
     run_backup = fixtures.BackupChecks.run_backup

business_backup_regression.py

@@ -36,6 +36,45 @@ class LocalRemote(GitRemote):
 
 
 class BackupChecks(unittest.TestCase):
+    def test_exhausted_old_git_failure_reports_current_discovery_without_reset(self):
+        with patch.object(self.remote, "identity", side_effect=BackupError("git_operation_failed")):
+            for _ in range(3):
+                self.run_backup()
+                self.now += 2000
+        before = (self.directory / "backup.json").read_bytes()
+        terminal = json.loads((self.source / "queue/complete/0000000001.json").read_bytes())
+        terminal["unexpected"] = "synthetic-secret-must-not-log"
+        write(self.source, "queue/complete/0000000001.json", terminal)
+        with tempfile.TemporaryDirectory(dir=self.root) as temporary, patch.object(self.remote, "identity") as remote:
+            session = backup.BackupSession(self.source, self.directory, Path(temporary), self.remote, clock=lambda: self.now)
+            for _ in range(2):
+                self.assertEqual(session.run(), "discovery_deferred")
+                self.assertEqual(session.discovery_diagnostic, dict(step="terminal_validation", failure="unclassified_field", attempts=3, retry="exhausted"))
+            remote.assert_not_called()
+        self.assertEqual((self.directory / "backup.json").read_bytes(), before)
+
+    def test_discovery_errno_and_unknown_errors_have_fixed_private_diagnostics(self):
+        import errno
+        for number, code in ((errno.ENOENT, "source_missing"), (errno.EACCES, "source_permission_denied"), (errno.EIO, "source_io_error")):
+            self.assertEqual(backup.discovery_code(OSError(number, "synthetic-secret /private/path")), code)
+        self.assertEqual(backup.discovery_code(BackupError("synthetic-secret-must-not-log")), "schema_validation_failed")
+
+    def test_supervisor_logs_only_allowlisted_discovery_details(self):
+        private = "synthetic-secret /private/path"
+        payload = dict(result="discovery_deferred", discovery=dict(step="terminal_validation", failure="unclassified_field",
+                       attempts=3, retry="exhausted", raw_error=private))
+        process = unittest.mock.Mock(pid=999999, returncode=0)
+        process.poll.return_value = 0
+        def launch(*args, **kwargs):
+            kwargs["stdout"].write(backup.export.encoded(payload))
+            return process
+        with patch.object(backup.subprocess, "Popen", side_effect=launch), patch.object(backup.os, "killpg", create=True), \
+                self.assertLogs(backup.logger, level="WARNING") as logs:
+            self.assertEqual(backup._supervise(self.root), "discovery_deferred")
+        self.assertIn("failure=unclassified_field", "".join(logs.output))
+        self.assertIn("retry=exhausted", "".join(logs.output))
+        self.assertNotIn(private, "".join(logs.output))
+
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory(prefix="business-backup-test-")
         self.addCleanup(self.temporary.cleanup)
@@ -349,20 +388,85 @@ class BackupChecks(unittest.TestCase):
                 self.assertNotIn(self.remote.token.encode(), path.read_bytes())
         self.assertEqual((checkout / "README.md").read_text(), "Synthetic backup fixture; preserve this file.\n")
 
-    def test_source_change_refused_before_upload(self):
-        original = backup.export._hash_file
+    def test_valid_collected_snapshot_accepts_later_production_change(self):
+        original = backup.export.source_records
+        relative = f"presentation_metadata/{TASKS[0]}/0000000001.json"
+        before = json.loads((self.source / relative).read_bytes())
+        def changing(path, *args, **kwargs):
+            yield from original(path, *args, **kwargs)
+            if path == self.source / relative:
+                write(self.source, relative, dict(before, display_name="Newer live metadata"))
+        with patch.object(backup.export, "source_records", side_effect=changing):
+            self.assertEqual(self.run_backup(), "confirmed")
+        checkout, _ = self.remote.clone(self.root)
+        backup.export.reconstruct(checkout / backup.NAMESPACE / "current", self.root / "reconstructed")
+        rows = [json.loads(raw) for raw in (self.root / "reconstructed/business-records.jsonl").read_bytes().splitlines()]
+        captured = next(row["data"] for row in rows if row["source"] == relative)
+        self.assertEqual(captured["display_name"], before["display_name"])
+        self.assertEqual(self.remote.pushes, 1)
+        self.assertIsNone(self.state()["pending"])
+
+    def test_collection_reads_each_business_file_once_without_live_hash_recheck(self):
+        original = backup.export.source_records
         calls = {}
-        def changed(path):
-            digest = original(path)
-            if self.source in path.parents:
-                calls[path] = calls.get(path, 0) + 1
-                if calls[path] > 1:
-                    return "0" * 64
-            return digest
-        with patch.object(backup.export, "_hash_file", side_effect=changed):
-            self.assertIn(self.run_backup(), {"source_changed", "source_changed_during_collection"})
+        def reading(path, *args, **kwargs):
+            calls[path] = calls.get(path, 0) + 1
+            yield from original(path, *args, **kwargs)
+        with patch.object(backup.export, "inventory", wraps=backup.export.inventory) as listing, \
+                patch.object(backup.export, "source_records", side_effect=reading), \
+                patch.object(backup.export, "_hash_file", side_effect=AssertionError("Unexpected content reread")):
+            backup.collect(self.source, self.root / "collected")
+        self.assertEqual(listing.call_count, 1)
+        self.assertTrue(calls)
+        self.assertEqual(set(calls.values()), {1})
+        backup.export.export(self.root / "collected", self.root / "export", attempts=1)
+        backup.export.verify(self.root / "export")
+
+    def test_disappearing_classified_file_is_not_silently_deleted(self):
+        original = backup.export.inventory
+        def disappearing(source):
+            listing = original(source)
+            (source / f"presentation_metadata/{TASKS[0]}/0000000001.json").unlink()
+            return listing
+        with patch.object(backup.export, "inventory", side_effect=disappearing):
+            with self.assertRaises(FileNotFoundError):
+                backup.collect(self.source, self.root / "collected")
         self.assertEqual(self.remote.pushes, 0)
 
+    def test_short_read_is_rejected_even_if_returned_json_is_parseable(self):
+        from io import BytesIO
+        path = self.source / f"presentation_metadata/{TASKS[0]}/0000000001.json"
+        raw = path.read_bytes()
+        descriptor = os.open(path, os.O_RDONLY)
+        class ShortRead(BytesIO):
+            def fileno(self):
+                return descriptor
+        try:
+            with patch.object(Path, "open", return_value=ShortRead(b"{}")):
+                with self.assertRaisesRegex(backup.ExportError, "source_read_incomplete"):
+                    list(backup.export.source_records(path, "metadata", len(raw) + 1))
+            with patch.object(Path, "open", return_value=ShortRead(b'{"event":"started"}\n')):
+                with self.assertRaisesRegex(backup.ExportError, "source_read_incomplete"):
+                    list(backup.export.source_records(self.root / "short-read.jsonl", "audit", len(raw) + 1))
+        finally:
+            os.close(descriptor)
+
+    def test_failed_new_snapshot_preserves_confirmed_git_and_receipts(self):
+        self.assertEqual(self.run_backup(), "confirmed")
+        head, receipts = self.remote.head(), self.state()["receipts"]
+        self.add_terminal()
+        history_path = self.source / f"results/{TASKS[0]}/_history.2026-10.jsonl"
+        rows = [json.loads(raw) for raw in history_path.read_bytes().splitlines()]
+        rows[-1]["score"] += 1
+        history_path.write_bytes(b"".join(json.dumps(row).encode() + b"\n" for row in rows))
+        self.assertEqual(self.run_backup(), "conflicting_submission_records")
+        self.assertEqual(self.remote.head(), head)
+        self.assertEqual(self.state()["confirmed_head"], head)
+        self.assertEqual(self.state()["receipts"], receipts)
+        self.assertEqual(self.remote.pushes, 1)
+        checkout, _ = self.remote.clone(self.root)
+        backup.export.verify(checkout / backup.NAMESPACE / "current")
+
     def test_remote_conflict_does_not_overwrite_another_commit(self):
         original = self.remote.push
         def advance(directory, commit, parent):

business_export_regression.py

@@ -16,6 +16,51 @@ from tools.export_schema import ExportError, SCHEMAS, clean, sanitize_text
 
 
 class ExportChecks(unittest.TestCase):
+    def resume_event(self):
+        record = self.load("queue/complete/0000000001.json")
+        return dict(event="operator_compatibility_resume", event_id="0000000001:operator-resume",
+                    submission_id="0000000001", timestamp=record["completed_at"], reason="Synthetic compatibility review",
+                    backup_directory="C:/synthetic/private-backup", previous_evaluator=record["evaluation_provenance"]["evaluator"],
+                    retained_results_digest="a" * 64, retained_scenes=["synthetic-case"],
+                    verification=dict(pmf_tree_at_both_commits="b" * 40, worker_source_change="Equivalent full-view scoring",
+                                      full_view_regression="Synthetic aggregation matches"))
+
+    def test_operator_resume_audit_roundtrip_without_schema_bypass(self):
+        event = self.resume_event()
+        def add_evidence(record):
+            record.setdefault("submission_events", []).append(event)
+            record["evaluation_provenance"]["compatibility_recoveries"] = [event]
+            record["requeued_at"] = event["timestamp"]
+        self.mutate("queue/complete/0000000001.json", add_evidence)
+        for path in self.source.glob("results/*/_history.*.jsonl"):
+            rows = [json.loads(line) for line in path.read_bytes().splitlines()]
+            for row in rows:
+                if row.get("submission_id") == "0000000001":
+                    add_evidence(row)
+            path.write_bytes(b"".join(json.dumps(row).encode() + b"\n" for row in rows))
+        manifest = self.export()
+        restored = self.reconstructed(manifest)
+        self.assertIn(clean(event, SCHEMAS["audit"]), restored[("queue/complete/0000000001.json", 0)]["submission_events"])
+        record = restored[("queue/complete/0000000001.json", 0)]
+        self.assertEqual(record["evaluation_provenance"]["compatibility_recoveries"], [clean(event, SCHEMAS["audit"])])
+        self.assertEqual(record["requeued_at"], event["timestamp"])
+        self.assertNotIn(CANARY, json.dumps(list(restored.values())))
+
+    def test_operator_resume_unknown_fields_wrong_event_and_invalid_proof_refused(self):
+        event = self.resume_event()
+        for change, code in ((dict(unexpected="must not be accepted"), "unclassified_field"),
+                             (dict(event="completed"), "unclassified_field"),
+                             (dict(retained_results_digest="invalid"), "invalid_evidence_digest"),
+                             (dict(retained_scenes=["duplicate", "duplicate"]), "invalid_retained_scene_identity")):
+            with self.subTest(code=code):
+                with self.assertRaisesRegex(ExportError, code):
+                    clean({**event, **change}, SCHEMAS["audit"])
+        event.pop("verification")
+        with self.assertRaisesRegex(ExportError, "incomplete_operator_resume_event"):
+            clean(event, SCHEMAS["audit"])
+        with self.assertRaisesRegex(ExportError, "invalid_operator_resume_event"):
+            clean([dict(event="completed")], SCHEMAS["submission"]["evaluation_provenance"]["compatibility_recoveries"])
+
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory(prefix="physinone-export-test-")
         self.addCleanup(self.temporary.cleanup)

business_recovery_export_regression.py

@@ -0,0 +1,213 @@
+"""Strict, evidence-scoped recovery timestamp acceptance; synthetic data only."""
+import copy
+import json
+from pathlib import Path
+import socket
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from tools import business_export as export
+from tools.export_fixture import fixture, write, TASKS
+from tools.export_schema import clean, SCHEMAS, ExportError
+
+
+class RecoveryExportChecks(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="recovery-export-test-")
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.source = fixture(self.root / "source", counts=(3, 103, 3, 3))
+        self.sid = "0000000002"
+        self.path = f"queue/complete/{self.sid}.json"
+        self.record = json.loads((self.source / self.path).read_bytes())
+        self.progress = json.loads((self.source / f"queue/progress/{self.sid}.json").read_bytes())
+        self.start = self.record["started_at"]
+        self.failed, self.resumed, self.completed, self.repaired = [f"2026-10-11T{hour:02d}:00:00Z" for hour in (2, 3, 4, 5)]
+        for index, (case, row) in enumerate(self.progress["scene_results"].items()):
+            row["metrics"] = {f"{metric}_{split}": 0.5 for metric in ("pmf", "psnr", "ssim", "lpips") for split in ("train", "novel")}
+            if index >= 80:
+                row["ts"] = self.completed
+                self.progress["case_started_at"][case] = self.resumed
+                for event in self.progress["submission_events"]:
+                    if event.get("case") == case:
+                        event["timestamp"] = self.resumed if event["event"] == "case_started" else self.completed
+        retained = list(self.progress["scene_results"])[:80]
+        self.recovery = dict(event="operator_compatibility_resume", event_id=f"{self.sid}:operator-resume",
+            submission_id=self.sid, timestamp=self.resumed, reason="Synthetic verified compatibility recovery",
+            backup_directory="C:/synthetic/private-backup", previous_evaluator=copy.deepcopy(self.record["evaluation_provenance"]["evaluator"]),
+            retained_results_digest=export._retained_results_digest({case: self.progress["scene_results"][case] for case in retained}), retained_scenes=retained,
+            verification=dict(pmf_tree_at_both_commits="b" * 40, worker_source_change="Equivalent full-view code",
+                              full_view_regression="Full-view aggregation unchanged"))
+        self.events = [dict(event="running", event_id=f"{self.sid}:running:1", timestamp=self.start),
+            dict(event="failed", event_id=f"{self.sid}:terminal", timestamp=self.failed), self.recovery,
+            dict(event="running", event_id=f"{self.sid}:running:4", timestamp=self.start),
+            dict(event="operator_state_reconciliation", event_id=f"{self.sid}:repair", timestamp=self.repaired,
+                 previous_state="failed", current_state="done", result="repaired")]
+        self.record.update(worker_attempt=4, requeued_at=self.resumed, completed_at=self.completed, submission_events=self.events,
+                           metrics=dict(next(iter(self.progress["scene_results"].values()))["metrics"]))
+        self.record["evaluation_provenance"]["compatibility_recoveries"] = [copy.deepcopy(self.recovery)]
+        self.save_records()
+        tombstone = self.source / f"results/{TASKS[1]}/_deleted_submissions.jsonl"
+        row = json.loads(tombstone.read_bytes())
+        row["deleted_at"] = self.repaired
+        tombstone.write_bytes(json.dumps(row).encode() + b"\n")
+        receipt = json.loads((self.source / ("queue/deletions/complete/" + "f" * 16 + ".json")).read_bytes())
+        receipt["completed_at"] = self.repaired
+        write(self.source, "queue/deletions/complete/" + "f" * 16 + ".json", receipt)
+
+    def save_records(self):
+        write(self.source, self.path, self.record)
+        write(self.source, f"queue/progress/{self.sid}.json", self.progress)
+        history = self.source / f"results/{TASKS[1]}/_history.2026-10.jsonl"
+        combined, seen = [], set()
+        for event in self.events + self.progress["submission_events"]:
+            identity = event.get("event_id")
+            if identity not in seen:
+                combined.append(event)
+                seen.add(identity)
+        history.write_bytes(json.dumps(dict(self.record, submission_events=combined)).encode() + b"\n")
+
+    def validate_events(self, record=None, events=None):
+        record = clean(record or self.record, SCHEMAS["submission"])
+        events = clean(events or self.events, ("list", SCHEMAS["audit"]))
+        return export._events(events, self.sid, submission=record)
+
+    def test_full_103_case_export_verify_reconstruct_preserves_append_order(self):
+        before = {path: (export._hash_file(path), path.stat().st_mtime_ns) for path in self.source.rglob("*") if path.is_file()}
+        manifest = export.export(self.source, self.root / "export", attempts=1)
+        export.verify(self.root / "export")
+        export.reconstruct(self.root / "export", self.root / "reconstructed")
+        rows = [json.loads(raw) for raw in (self.root / "reconstructed/business-records.jsonl").read_bytes().splitlines()]
+        terminal = next(row["data"] for row in rows if row["source"] == self.path)
+        self.assertEqual(terminal["submission_events"], clean(self.events, ("list", SCHEMAS["audit"])))
+        self.assertEqual(terminal["score"], 0.5)
+        self.assertEqual(terminal["scenes_done"], 103)
+        self.assertEqual(before, {path: (export._hash_file(path), path.stat().st_mtime_ns) for path in before})
+        self.assertEqual(manifest["validation"]["historical_tombstones"]["policy"], "historical-deletion-evidence-v1")
+
+    def test_positive_recovery_keeps_occurrence_watermark(self):
+        self.assertEqual(len(self.validate_events()), 1)
+        events = copy.deepcopy(self.events)
+        events.append(dict(event="queued", event_id="synthetic-late-invalid", timestamp=self.failed))
+        with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+            self.validate_events(events=events)
+
+    def test_missing_or_contradictory_provenance_rejects_exception(self):
+        for change in (None, [], [dict(self.recovery, reason="Contradictory evidence")]):
+            with self.subTest(change=bool(change)):
+                record = copy.deepcopy(self.record)
+                record["evaluation_provenance"]["compatibility_recoveries"] = change
+                with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+                    self.validate_events(record=record)
+
+    def test_wrong_original_start_attempt_or_requeue_timestamp_rejected(self):
+        for key, value in (("started_at", self.failed), ("worker_attempt", 5), ("requeued_at", self.completed)):
+            with self.subTest(key=key):
+                record = copy.deepcopy(self.record)
+                record[key] = value
+                with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+                    self.validate_events(record=record)
+
+    def test_ordinary_and_case_timestamp_regressions_remain_rejected(self):
+        ordinary = [dict(event="queued", timestamp=self.resumed), dict(event="running", timestamp=self.start)]
+        with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+            export._events(ordinary, self.sid)
+        events = copy.deepcopy(self.events)
+        events[3]["case"] = "synthetic-case"
+        events.insert(3, dict(event="case_started", case="synthetic-case", timestamp=self.resumed, attempt=2))
+        with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+            self.validate_events(events=events)
+
+    def test_case_attempt_order_is_still_strict(self):
+        events = [dict(event="case_started", case="synthetic-case", timestamp=self.start, attempt=2),
+                  dict(event="case_completed", case="synthetic-case", timestamp=self.completed, attempt=1)]
+        with self.assertRaisesRegex(ExportError, "case_attempt_order_mismatch"):
+            self.validate_events(events=events)
+
+    def test_retained_results_digest_is_checked_against_progress(self):
+        retained = self.recovery["retained_scenes"][0]
+        self.progress["scene_results"][retained]["metrics"]["pmf_novel"] = 0.7
+        self.save_records()
+        with self.assertRaisesRegex(ExportError, "recovery_retained_results_mismatch"):
+            export.export(self.source, self.root / "export", attempts=1)
+
+    def test_failure_and_unknown_evidence_cannot_bypass_schema(self):
+        record = copy.deepcopy(self.record)
+        record["status"] = "failed"
+        with self.assertRaisesRegex(ExportError, "event_order_mismatch"):
+            self.validate_events(record=record)
+        events = copy.deepcopy(self.events)
+        events[2]["verification"]["unknown"] = "must refuse"
+        with self.assertRaisesRegex(ExportError, "unclassified_field"):
+            self.validate_events(events=events)
+
+    def test_reset_incomplete_case_start_evidence_is_typed_and_preserved(self):
+        event = copy.deepcopy(self.recovery)
+        event["reset_incomplete_case_starts"] = {"synthetic-case": self.start}
+        cleaned = clean(event, SCHEMAS["audit"])
+        self.assertEqual(cleaned["reset_incomplete_case_starts"], event["reset_incomplete_case_starts"])
+        event["reset_incomplete_case_starts"]["synthetic-case"] = "invalid-date"
+        with self.assertRaisesRegex(ExportError, "invalid_timestamp"):
+            clean(event, SCHEMAS["audit"])
+
+    def test_manifest_record_digest_does_not_replace_operator_audit_digest(self):
+        retained = self.recovery["retained_scenes"]
+        self.recovery["retained_results_digest"] = export.record_digest({case: self.progress["scene_results"][case] for case in retained})
+        self.record["evaluation_provenance"]["compatibility_recoveries"] = [copy.deepcopy(self.recovery)]
+        self.save_records()
+        with self.assertRaisesRegex(ExportError, "recovery_retained_results_mismatch"):
+            export.export(self.source, self.root / "export", attempts=1)
+
+    def add_case_reset(self):
+        case = list(self.progress["scene_results"])[80]
+        position = next(index for index, event in enumerate(self.progress["submission_events"])
+                        if event.get("case") == case and event.get("event") == "case_started")
+        current = self.progress["submission_events"][position]
+        current["event_id"] = f"{self.sid}:{case}:started"
+        previous = dict(current, timestamp=self.start, actor_worker_id="previous-worker")
+        recovery = dict(copy.deepcopy(self.recovery), reset_incomplete_case_starts={case: self.start})
+        self.progress["submission_events"].insert(position, recovery)
+        self.progress["submission_events"].insert(0, previous)
+        self.save_records()
+
+    def test_verified_case_start_reset_preserves_both_historical_occurrences(self):
+        self.add_case_reset()
+        export.export(self.source, self.root / "export", attempts=1)
+        export.verify(self.root / "export")
+        export.reconstruct(self.root / "export", self.root / "reconstructed")
+        rows = [json.loads(raw) for raw in (self.root / "reconstructed/business-records.jsonl").read_bytes().splitlines()]
+        progress = next(row["data"] for row in rows if row["source"] == f"queue/progress/{self.sid}.json")
+        self.assertEqual(progress["submission_events"], clean(self.progress["submission_events"], ("list", SCHEMAS["audit"])))
+
+    def test_case_start_identity_conflict_without_reset_remains_rejected(self):
+        self.add_case_reset()
+        for event in self.progress["submission_events"]:
+            if event.get("event") == "operator_compatibility_resume":
+                event.pop("reset_incomplete_case_starts")
+        self.save_records()
+        with self.assertRaisesRegex(ExportError, "conflicting_event_identity"):
+            export.export(self.source, self.root / "export", attempts=1)
+
+    def test_case_reset_requires_matching_terminal_recovery_evidence(self):
+        self.add_case_reset()
+        for event in self.progress["submission_events"]:
+            if event.get("event") == "operator_compatibility_resume":
+                event["reason"] = "Contradictory recovery record"
+        self.save_records()
+        with self.assertRaisesRegex(ExportError, "case_reset_recovery_evidence_mismatch"):
+            export.export(self.source, self.root / "export", attempts=1)
+
+    def test_verified_reset_does_not_allow_case_attempt_counter_regression(self):
+        self.add_case_reset()
+        case = list(self.progress["scene_results"])[80]
+        self.progress["submission_events"].insert(1, dict(event="case_retry", event_id=f"{self.sid}:synthetic-retry",
+                                                         case=case, timestamp=self.failed, attempt=2))
+        self.save_records()
+        with self.assertRaisesRegex(ExportError, "case_attempt_order_mismatch"):
+            export.export(self.source, self.root / "export", attempts=1)
+
+
+if __name__ == "__main__":
+    with patch.object(socket.socket, "connect", side_effect=AssertionError("Live network forbidden")):
+        unittest.main(verbosity=2)

src/submission/business_backup.py

@@ -1,6 +1,7 @@
 """Default-disabled, process-isolated business backup auxiliary; no Worker imports."""
 from contextlib import contextmanager
 import json
+import errno
 import logging
 import os
 from pathlib import Path
@@ -30,6 +31,19 @@ NAMESPACE = "physinone-business-backup"
 ATTRIBUTES = "* -text -filter -ident -working-tree-encoding\n"
 _stop = threading.Event()
 _thread = None
+DISCOVERY_CODES = frozenset({"unclassified_field", "unsupported_business_source", "invalid_json", "source_scan_failed",
+                            "unknown_runtime_heartbeat_format", "runtime_heartbeat_size_limit", "cancellation_transition_incomplete",
+                            "unstable_terminal_record", "conflicting_terminal_records", "incomplete_operator_resume_event",
+                            "invalid_operator_resume_event", "invalid_retained_scene_identity", "invalid_evidence_digest", "invalid_identifier",
+                            "source_missing", "source_permission_denied", "source_io_error", "source_unavailable", "schema_validation_failed"})
+
+
+def discovery_code(error):
+    if isinstance(error, OSError):
+        return {errno.ENOENT: "source_missing", errno.EACCES: "source_permission_denied", errno.EPERM: "source_permission_denied",
+                errno.EIO: "source_io_error"}.get(error.errno, "source_unavailable")
+    code = str(error)
+    return code if code in DISCOVERY_CODES else "schema_validation_failed"
 
 
 def configuration():
@@ -119,15 +133,13 @@ def collect(source, destination):
     total = sum((source / relative).stat().st_size for relative, _ in listing)
     if total > MAX_SOURCE_BYTES or len(listing) > 5000:
         refuse("snapshot_resource_limit")
-    hashes = {}
     destination.mkdir(mode=0o700)
     for relative, kind in listing:
         path = source / relative
-        digest = export._hash_file(path)
         target = destination / relative
         target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
         with target.open("xb") as handle:
-            for _, data in export.source_records(path, kind, export.MAX_SOURCE_BYTES, digest):
+            for _, data in export.source_records(path, kind, export.MAX_SOURCE_BYTES):
                 if path.suffix == ".jsonl":
                     encoder = json.JSONEncoder(ensure_ascii=False, separators=(",", ":"), allow_nan=False)
                     for fragment in encoder.iterencode(data):
@@ -135,9 +147,6 @@ def collect(source, destination):
                     handle.write(b"\n")
                 else:
                     handle.write(export.encoded(data))
-        hashes[relative] = digest
-    if export.inventory(source) != listing or any(export._hash_file(source / name) != digest for name, digest in hashes.items()):
-        refuse("source_changed_during_collection")
     return terminal_map(destination, listing)
 
 
@@ -164,6 +173,7 @@ class BackupSession:
     def __init__(self, source, directory, workspace, remote, clock=time.time):
         self.source, self.directory, self.workspace, self.remote, self.clock = source, directory, workspace, remote, clock
         self.path = directory / "backup.json"
+        self.discovery_diagnostic = None
 
     def load(self):
         try:
@@ -231,16 +241,22 @@ class BackupSession:
         return marker
 
     def run(self):
+        self.discovery_diagnostic = None
         state = self.load()
+        discovery_step = "inventory"
         try:
             if state["pending"] is not None and state["pending"].get("commit"):
                 terminals = state["pending"]["terminals"]
             else:
                 listing = export.inventory(self.source)
+                discovery_step = "terminal_validation"
                 terminals = terminal_map(self.source, listing, state["receipts"])
         except (ExportError, BackupError, OSError) as error:
             pending = state["pending"]
+            self.discovery_diagnostic = dict(step=discovery_step, failure=discovery_code(error),
+                                             attempts=pending["attempts"] if pending else 0, retry="attempt")
             if pending is not None and (self.clock() < pending.get("next_try", 0) or pending["attempts"] >= MAX_ATTEMPTS):
+                self.discovery_diagnostic["retry"] = "exhausted" if pending["attempts"] >= MAX_ATTEMPTS else "backoff"
                 return "discovery_deferred"
             if pending is None:
                 pending = dict(operation=uuid.uuid4().hex, trigger="terminal" if state["baseline"] else "baseline", terminals={},
@@ -251,6 +267,9 @@ class BackupSession:
             state["pending"] = pending
             state["last_result"] = dict(status="failed", failure=pending["failure"], at=self.clock())
             save_state(self.path, state)
+            self.discovery_diagnostic["attempts"] = pending["attempts"]
+            if pending["attempts"] >= MAX_ATTEMPTS:
+                self.discovery_diagnostic["retry"] = "exhausted"
             return pending["failure"]
         pending = state["pending"]
         if state["baseline"] and not (terminals.keys() - state["receipts"].keys()) and pending is None:
@@ -419,7 +438,7 @@ def run_child(workspace):
         session = BackupSession(source, directory, workspace, remote)
         result = session.run()
     notify_exhaustion(source, session)
-    print(json.dumps({"result": result}))
+    print(json.dumps({"result": result, "discovery": session.discovery_diagnostic}))
 
 
 def notify_exhaustion(source, session):
@@ -449,7 +468,14 @@ def _supervise(workspace):
                     refuse("backup_resource_or_shutdown_stop")
         if process.returncode != 0:
             refuse("backup_child_failed")
-        return read_json(output, 4096).get("result", "disabled")
+        payload = read_json(output, 4096)
+        diagnostic = payload.get("discovery")
+        if (isinstance(diagnostic, dict) and diagnostic.get("step") in {"inventory", "terminal_validation"}
+                and diagnostic.get("failure") in DISCOVERY_CODES and type(diagnostic.get("attempts")) is int
+                and 0 <= diagnostic["attempts"] <= MAX_ATTEMPTS and diagnostic.get("retry") in {"attempt", "backoff", "exhausted"}):
+            logger.warning("Business backup discovery unavailable (step=%s, failure=%s, attempts=%d, retry=%s)",
+                           diagnostic["step"], diagnostic["failure"], diagnostic["attempts"], diagnostic["retry"])
+        return payload.get("result", "disabled")
     finally:
         if process is not None:
             if os.name != "nt":

tools/business_export.py

@@ -222,32 +222,37 @@ def _hash_file(path):
     return digest.hexdigest()
 
 
-def source_records(path, kind, limit, expected_hash):
+def source_records(path, kind, limit, expected_hash=None):
     _no_symlink(path)
     digest = hashlib.sha256()
     with path.open("rb") as handle:
+        size = os.fstat(handle.fileno()).st_size
         if path.suffix == ".jsonl":
-            ordinal = 0
+            ordinal, consumed = 0, 0
             while raw := handle.readline(limit + 1):
                 if len(raw) > limit:
                     fail("source_record_size_limit")
                 digest.update(raw)
+                consumed += len(raw)
                 if not raw.strip():
                     fail("empty_jsonl_record")
                 check_parse_memory(raw)
                 yield ordinal, clean(parse(raw), SCHEMAS[kind])
                 ordinal += 1
+            if consumed < size:
+                fail("source_read_incomplete")
         else:
-            size = os.fstat(handle.fileno()).st_size
             if size > limit:
                 fail("source_record_size_limit")
             raw = handle.read(min(limit + 1, size + 1))
             if len(raw) > limit:
                 fail("source_record_size_limit")
+            if len(raw) != size:
+                fail("source_read_incomplete")
             digest.update(raw)
             check_parse_memory(raw)
             yield 0, clean(parse(raw), SCHEMAS[kind])
-    if digest.hexdigest() != expected_hash:
+    if expected_hash is not None and digest.hexdigest() != expected_hash:
         fail("source_changed")
 
 
@@ -407,14 +412,103 @@ def records(root, manifest):
             yield _read_export_record(root, entry)
 
 
-def _events(events, submission_id=None):
+def _retained_results_digest(value):
+    digest = hashlib.sha256()
+    encoder = json.JSONEncoder(ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
+    for fragment in encoder.iterencode(value):
+        digest.update(fragment.encode("utf-8"))
+    return digest.hexdigest()
+
+
+def _recovery_start_reuse(events, index, submission):
+    if not isinstance(submission, dict) or index < 2 or submission.get("status") != "done":
+        return None
+    running, recovery, failure = events[index], events[index - 1], events[index - 2]
+    identity = submission.get("submission_id")
+    if (not identity or running.get("event") != "running" or running.get("case") or running.get("case_id")
+            or recovery.get("event") != "operator_compatibility_resume" or recovery.get("submission_id") != identity
+            or failure.get("event") != "failed" or failure.get("event_id") != f"{identity}:terminal"
+            or running.get("timestamp") != submission.get("started_at")
+            or recovery.get("timestamp") != submission.get("requeued_at")
+            or type(submission.get("worker_attempt")) is not int
+            or running.get("event_id") != f"{identity}:running:{submission['worker_attempt']}"
+            or submission["worker_attempt"] <= 1
+            or type(submission.get("scenes_total")) is not int or submission["scenes_total"] <= 0
+            or submission.get("scenes_done") != submission.get("scenes_total") or submission.get("scenes_failed") != 0):
+        return None
+    originals = [event for event in events[:index - 2] if event.get("event") == "running"
+                 and event.get("event_id") == f"{identity}:running:1" and event.get("timestamp") == running["timestamp"]]
+    if len(originals) != 1:
+        return None
+    provenance = submission.get("evaluation_provenance") or {}
+    proofs = provenance.get("compatibility_recoveries") or []
+    if not isinstance(proofs, list) or any(not isinstance(event, dict) for event in proofs):
+        return None
+    matches = [event for event in proofs
+               if event.get("event_id") == recovery.get("event_id")]
+    if len(matches) != 1 or record_digest(matches[0]) != record_digest(recovery):
+        return None
+    try:
+        clean(recovery, ("operator_resume_record",))
+        if len(recovery["retained_scenes"]) > submission["scenes_total"]:
+            return None
+        start, failed, resumed = [datetime.fromisoformat(event["timestamp"].replace("Z", "+00:00"))
+                                  for event in (running, failure, recovery)]
+        if any(value.tzinfo is None for value in (start, failed, resumed)) or not start <= failed <= resumed:
+            return None
+    except (ExportError, ValueError, TypeError, KeyError):
+        return None
+    return recovery
+
+
+def _reset_case_start(events, index, previous, progress):
+    current = events[index]
+    if not isinstance(progress, dict):
+        return None
+    identity, case = progress.get("submission_id"), current.get("case")
+    if (not identity or not case or current.get("event") != "case_started" or previous.get("event") != "case_started"
+            or current.get("event_id") != f"{identity}:{case}:started" or previous.get("case") != case
+            or current.get("attempt") != 1 or previous.get("attempt") != 1
+            or progress.get("case_started_at", {}).get(case) != current.get("timestamp")
+            or case not in (progress.get("scene_results") or {}) or case in (progress.get("scene_failures") or {})
+            or any(previous.get(key) != current.get(key) for key in (previous.keys() | current.keys()) - {"timestamp", "actor_worker_id"})):
+        return None
+    matches = [event for event in events[:index] if event.get("event") == "operator_compatibility_resume"
+               and event.get("submission_id") == identity
+               and (event.get("reset_incomplete_case_starts") or {}).get(case) == previous.get("timestamp")]
+    if len(matches) != 1:
+        return None
+    recovery = matches[0]
+    originals = [position for position, event in enumerate(events[:index]) if event == previous]
+    if len(originals) != 1 or events.index(recovery) <= originals[0]:
+        return None
+    try:
+        clean(recovery, ("operator_resume_record",))
+        if case in recovery["retained_scenes"]:
+            return None
+        values = (previous["timestamp"], recovery["timestamp"], current["timestamp"], progress["scene_results"][case]["ts"])
+        if any(not isinstance(value, str) for value in values):
+            return None
+        times = [datetime.fromisoformat(value.replace("Z", "+00:00")) for value in values]
+        if any(value.tzinfo is None for value in times) or not times[0] < times[1] <= times[2] <= times[3]:
+            return None
+    except (ExportError, KeyError, ValueError, TypeError):
+        return None
+    return recovery
+
+
+def _events(events, submission_id=None, *, submission=None, progress=None):
     timestamps, ids, attempts = {}, {}, {}
-    for event in events or []:
+    accepted_recoveries = []
+    for index, event in enumerate(events or []):
         if submission_id and event.get("submission_id") not in {None, submission_id}:
             fail("event_submission_mismatch")
         identity = event.get("event_id")
         if identity and identity in ids and ids[identity] != event:
-            fail("conflicting_event_identity")
+            recovery = _reset_case_start(events, index, ids[identity], progress)
+            if recovery is None:
+                fail("conflicting_event_identity")
+            accepted_recoveries.append(recovery)
         if identity:
             ids[identity] = event
         timestamp = event.get("timestamp")
@@ -427,14 +521,19 @@ def _events(events, submission_id=None):
                 fail("invalid_event_timestamp")
             group = event.get("case") or event.get("case_id") or "_submission"
             if group in timestamps and parsed < timestamps[group]:
-                fail("event_order_mismatch")
-            timestamps[group] = parsed
+                recovery = _recovery_start_reuse(events, index, submission) if group == "_submission" else None
+                if recovery is None:
+                    fail("event_order_mismatch")
+                accepted_recoveries.append(recovery)
+            else:
+                timestamps[group] = parsed
         group = event.get("case") or event.get("case_id")
         attempt = event.get("attempt")
         if group and attempt is not None:
             if attempt < 1 or attempt < attempts.get(group, 1):
                 fail("case_attempt_order_mismatch")
             attempts[group] = attempt
+    return accepted_recoveries
 
 
 def validate(root, manifest):
@@ -443,6 +542,8 @@ def validate(root, manifest):
     tombstone_rows, deletion_receipts, deletion_audits = {}, defaultdict(list), defaultdict(list)
     parent_times = defaultdict(set)
     terminals, history_ids = set(), set()
+    recovery_checks = defaultdict(dict)
+    case_reset_checks = defaultdict(dict)
     for record in records(root, manifest):
         data, kind, selector = record["data"], record["kind"], record["selector"]
         if kind != "registry" or not selector:
@@ -517,7 +618,8 @@ def validate(root, manifest):
             if previous and any(previous[key] != summary[key] for key in previous.keys() & summary.keys()):
                 fail("conflicting_submission_records")
             submissions[submission_id] = {**(previous or {}), **summary}
-            _events(data.get("submission_events"), submission_id)
+            for recovery in _events(data.get("submission_events"), submission_id, submission=data):
+                recovery_checks[submission_id][recovery["event_id"]] = recovery
             primary = data.get("primary_metric") or (data.get("creation_snapshot") or {}).get("primary_metric")
             if data.get("status") == "done" and primary and primary in (data.get("metrics") or {}) and data.get("score") != data["metrics"][primary]:
                 fail("score_primary_metric_mismatch")
@@ -546,7 +648,8 @@ def validate(root, manifest):
             progresses[submission_id] = summary
             if summary["total"] is not None and len(results) + len(failures) > summary["total"]:
                 fail("case_count_exceeds_total")
-            _events(data.get("submission_events"), submission_id)
+            for recovery in _events(data.get("submission_events"), submission_id, progress=data):
+                case_reset_checks[submission_id][recovery["event_id"]] = recovery
             for case, retry in (data.get("scene_retries") or {}).items():
                 if not isinstance(retry, dict) or retry.get("status") not in {"pending", "running", "exhausted", "recovered"}:
                     fail("invalid_case_retry")
@@ -610,6 +713,29 @@ def validate(root, manifest):
                     fail("terminal_case_count_mismatch")
             if submission["status"] == "done" and (progress["failed"] or progress["total"] is not None and progress["done"] != progress["total"]):
                 fail("done_progress_incomplete")
+    for identity, proofs in case_reset_checks.items():
+        for event_id, proof in proofs.items():
+            terminal_proof = recovery_checks[identity].get(event_id)
+            if (terminal_proof is None
+                    or record_digest({key: value for key, value in proof.items() if key != "reset_incomplete_case_starts"})
+                    != record_digest({key: value for key, value in terminal_proof.items() if key != "reset_incomplete_case_starts"})
+                    or terminal_proof.get("reset_incomplete_case_starts") is not None
+                    and terminal_proof["reset_incomplete_case_starts"] != proof["reset_incomplete_case_starts"]):
+                fail("case_reset_recovery_evidence_mismatch")
+    if recovery_checks:
+        checked = set()
+        for record in records(root, manifest):
+            if record["kind"] != "progress" or record["data"].get("submission_id") not in recovery_checks:
+                continue
+            identity = record["data"]["submission_id"]
+            results = record["data"].get("scene_results") or {}
+            for recovery in recovery_checks[identity].values():
+                retained = recovery["retained_scenes"]
+                if not set(retained) <= results.keys() or _retained_results_digest({case: results[case] for case in retained}) != recovery["retained_results_digest"]:
+                    fail("recovery_retained_results_mismatch")
+            checked.add(identity)
+        if recovery_checks.keys() - checked:
+            fail("recovery_progress_missing")
     tombstone_counts = Counter(retained_parent=0, standalone_deletion_verified=0, standalone_provenance_incomplete=0)
     standalone = {}
     all_receipts = defaultdict(list)

tools/export_schema.py

@@ -61,8 +61,9 @@ def fields(names, rule="text"):
 
 
 PROFILE_FIELDS = fields("name role affiliation email supervisor_name saved_at")
-EVENT = {**fields("event_id event action actor actor_worker_id actor_hf_username submission_id case case_id", "id"),
+EVENT_FIELDS = {**fields("event_id event action actor actor_worker_id actor_hf_username submission_id case case_id", "id"),
          **fields("timestamp previous_state current_state result final_outcome reason requested_action"), "attempt": "int"}
+EVENT = ("submission_event", EVENT_FIELDS)
 PROFILE_REVISION = {**fields("timestamp actor_hf_username researcher_identity_id"),
                     "changes": ("profile_changes",)}
 RULE_ACCEPTANCE = fields("researcher_identity_id rules_version accepted_at actor_hf_username")
@@ -103,6 +104,11 @@ CONFIG = {**fields("IMAGE_METRIC_FRAME_BATCH_SIZE PROPERTY_ESTIMATION_FRAME_BATC
 PROVENANCE = {"schema_version": "int", "input": fields("repository_id repository_type subdirectory revision resolved_at"),
               "evaluator": {**fields("git_commit git_commit_source source_digest protocol_version manifest_digest metric_configuration_digest"),
                             "manifest_version": "version", "metric_configuration_version": "int", "metric_configuration": CONFIG}}
+OPERATOR_RESUME_EVENT = {**EVENT_FIELDS, "backup_directory": "text", "previous_evaluator": PROVENANCE["evaluator"],
+                         "retained_results_digest": "sha256", "retained_scenes": ("list", "id"),
+                         "reset_incomplete_case_starts": ("map", "timestamp"),
+                         "verification": {"pmf_tree_at_both_commits": "sha1", "worker_source_change": "text", "full_view_regression": "text"}}
+PROVENANCE["compatibility_recoveries"] = ("list", ("operator_resume_record",))
 SUBMISSION = {**fields("submission_id researcher_identity_id hf_username_snapshot team_id owner owner_username task_name track_id worker_id last_renewed_by source_submission_id submitter_orcid orcid_id", "id"),
               **fields("display_name team_name user_dataset repo_subdir primary_metric status submitted_at started_at reclaimed_at completed_at updated_at cancel_requested_at error_message quota_consumed_at quota_refunded_at quota_refund_reason evaluation_revision"),
               **fields("scenes_total scenes_done scenes_failed scenes_count worker_attempt", "int"), "deleted": "bool",
@@ -110,6 +116,7 @@ SUBMISSION = {**fields("submission_id researcher_identity_id hf_username_snapsho
               "creation_snapshot": SNAPSHOT, "submission_events": ("list", EVENT), "evaluation_provenance": PROVENANCE,
               "scene_failures": ("map", "text")}
 SUBMISSION["history_written_at"] = "timestamp"
+SUBMISSION["requeued_at"] = "timestamp"
 RETRY = {**fields("attempts", "int"), **fields("status error"), "infrastructure": "bool"}
 PROGRESS = {**fields("submission_id task_name last_worker current_scene", "id"),
             **fields("current_stage last_updated_at next_retry_at last_retry_at"),
@@ -157,6 +164,25 @@ def clean(value, rule, depth=0):
             result[key] = clean(item, rule[key], depth + 1)
         return result
     if isinstance(rule, tuple):
+        if rule[0] in {"submission_event", "operator_resume_record"}:
+            if rule[0] == "operator_resume_record" and (not isinstance(value, dict) or value.get("event") != "operator_compatibility_resume"):
+                fail("invalid_operator_resume_event")
+            if isinstance(value, dict) and value.get("event") == "operator_compatibility_resume":
+                required = {"event_id", "timestamp", "submission_id", "reason", "backup_directory", "previous_evaluator",
+                            "retained_results_digest", "retained_scenes", "verification"}
+                if not required <= value.keys():
+                    fail("incomplete_operator_resume_event")
+                result = clean(value, OPERATOR_RESUME_EVENT, depth + 1)
+                if any(result.get(key) is None for key in required) or not isinstance(result["verification"], dict):
+                    fail("incomplete_operator_resume_event")
+                if set(result["verification"]) != {"pmf_tree_at_both_commits", "worker_source_change", "full_view_regression"} or any(item is None for item in result["verification"].values()):
+                    fail("incomplete_operator_resume_event")
+                retained = result["retained_scenes"]
+                if not retained or len(set(retained)) != len(retained):
+                    fail("invalid_retained_scene_identity")
+                clean(result["timestamp"], "timestamp", depth + 1)
+                return result
+            return clean(value, rule[1], depth + 1)
         if rule[0] == "legacy_account":
             return clean(value, "id" if isinstance(value, str) else rule[1], depth + 1)
         if rule[0] == "list":
@@ -182,6 +208,10 @@ def clean(value, rule, depth=0):
         return clean(value, "int", depth + 1)
     if not isinstance(value, str):
         fail("expected_string")
+    if rule in {"sha1", "sha256"}:
+        if not re.fullmatch(r"[0-9a-f]{40}" if rule == "sha1" else r"[0-9a-f]{64}", value):
+            fail("invalid_evidence_digest")
+        return value
     if rule == "timestamp":
         try:
             if datetime.fromisoformat(value.replace("Z", "+00:00")).tzinfo is None: