Bind review conclusions to evidence and reject stale submissions

README.md

@@ -68,6 +68,10 @@ uv run controltrace verify     # 校验 ZIP 文件哈希，并从包内源数据
 
 复核历史按解析后的 UTC 时间排列，保留小数秒精度；时间相同则按稳定复核 ID 排序，因此补录较早的记录不会替换较晚的结论。`verify` 同时检查每条复核的字段、复核人、时间、状态与结论：`pending` 不能带结论，`in_review` 可不填结论或标记待补证，`closed` 必须有最终结论和说明。即使文件哈希及 CSV、底稿内容相互一致，非法复核或逆序历史也会被拒绝。此检查仍不能证明复核人的身份或原始材料真实性。升级到新版本后，请重新导出旧版本底稿。
 
+**复核不能只跟着发现 ID 走。** 0.2.8 将每次新复核绑定到自动发现的完整证据指纹。相同证据重跑保留当前结论；同一 ID 的关联来源记录或判断口径变化后，原结论保留在历史中，当前状态返回待复核。界面保存时核对首次展示的编辑版本；其他人追加复核或证据发生变化后，须显式重新加载，防止旧表单覆盖新证据。版本检查、复核及指纹关联在一个事务内提交，并对同一发现的并发写入进行冲突检查。[复核完整性设计](docs/REVIEW_INTEGRITY.md)解释业务场景、Python API、迁移和可复现实验。
+
+新 ZIP 使用格式 3，增加 `review_evidence.json`，`verify` 根据重放的当前证据选择可适用的复核。旧数据库的复核历史继续保留，但因原证据版本未知，升级后需要重新评估当前证据。格式 2 / 0.2.7 底稿仍可核验，结果明确说明它没有证据版本关联；更早底稿请使用原版本。新包没有归档历史源快照，也不提供身份认证或签名。
+
 ## 数据字典与案例
 
 - [数据字典](docs/DATA_DICTIONARY.md)：各源表、主键、关联键、时间和来源字段。
@@ -87,7 +91,7 @@ python -m ruff check src tests
 python -m pytest -q
 ```
 
-**本机验证（2026-10-11，Windows / Python 3.12）：** 完整 pytest 为 **74 passed**，Ruff 与 git diff --check 通过；实际执行生成、规则测试、CSV/ZIP 导出及包内重放，仍有 12 条预置线索（9 条演示异常、3 条人工判断），验证 28 个源文件。新增回归确认截止日后创建的账号不造成离职/转岗误报，并覆盖 UTC 偏移及恰在截止时点创建的账号；缺少截止日前账号证据仍列为人工核实。既有测试覆盖日期边界、重复事件、非默认种子、缺失关联、复核时间和状态、界面筛选及重哈希后非法复核的拒绝。[GitHub Actions CI](https://github.com/dev-belly/ControlTrace/actions/workflows/ci.yml) 在 Ubuntu / Python 3.11、3.12 和 Windows / Python 3.12 运行完整 Ruff 与 pytest。规则版本为 0.2.7，旧版底稿应使用原版本校验，升级后重新导出。
+**本机验证（2026-10-12，Windows / Python 3.12.13）：** 完整 pytest 为 **90 passed**，Ruff 与 git diff --check 通过；实际 CSV/ZIP 导出及包内重放有 12 条预置线索（9 条演示异常、3 条人工判断），验证 28 个源文件。新增回归覆盖 CH002 审批时间修正后同一发现重新待复核、无关线索保留结论、两个复核者同版本并发、事务失败回滚、界面显式重新加载、旧数据库迁移、旧格式兼容及重哈希后的伪造当前结论拒绝。既有测试覆盖日期边界、缺失关联、复核 UTC 时间、合法状态和非法 JSON。[GitHub Actions CI](https://github.com/dev-belly/ControlTrace/actions/workflows/ci.yml) 配置在 Ubuntu / Python 3.11、3.12 和 Windows / Python 3.12 运行 Ruff 与完整 pytest。当前包版本为 0.2.8；格式 2 / 0.2.7 底稿可继续核验，新导出采用格式 3。
 
 ## 已知局限
 

docs/CASE_STUDY.md

@@ -2,6 +2,8 @@
 
 > 合成案例，非真实企业审计。本文记录一次真实运行 ControlTrace 演示程序并在界面保存示范复核的过程。`confirmed_exception` 仅指本合成数据符合演示控制的时间违例口径，不是对任何真实组织的审计结论。
 
+本文截图与示范底稿保存的是 0.2.7 的历史运行。0.2.8 新增复核与证据版本关联，以及旧表单冲突检查；旧数据库记录的原证据版本未知，须重新评估后记录新复核。该历史记录不自动获得新版本的证据关联保证。具体行为见[复核完整性设计](REVIEW_INTEGRITY.md)。
+
 ![D002 的发现详情、证据 ID 和规则依据](../assets/controltrace-finding.png)
 
 ## 1. 控制目标与总体

docs/REVIEW_INTEGRITY.md

@@ -0,0 +1,57 @@
+# 复核结论如何绑定证据
+
+> 合成案例；这里的关闭状态是线索处理结果，不是企业审计意见。
+
+## 业务问题：ID 没变，证据可能已经变了
+
+发现 ID 根据控制、原因代码和对象 ID 生成。它方便关联同一条线索，但不代表证据内容不变。例如 D002 的变更工单后来修正了审批时间，即使依然命中“审批晚于上线”，发现 ID 仍然相同。仅按发现 ID 取最后一条复核，会把基于原抽取的 `closed` 结论自动套到修正后的证据上。
+
+从 0.2.8 起，每次保存复核都会同时保存其自动发现指纹。指纹覆盖规则口径、关联源记录原值、时间线和自动判断。复核人、处理状态等界面附加字段不参与指纹，规则重跑的刷新时间也不参与。因此相同证据重跑保留结论，关联证据变化则返回 `pending`，原复核仍保留在历史中。
+
+当前结论是**在当前指纹下，按 UTC 复核时间和复核 ID 排序的最后一条记录**。历史补录沿用原来的时间排序规则。其他线索的复核不影响本线索的编辑版本。若完整证据恢复为已复核的同一个指纹，对应历史结论也会重新适用；这里比较的是确切内容，不推断来源的可信性。
+
+## 保存前核对编辑版本
+
+界面首次展示复核表单时保存 `review_revision`，它概括当前证据指纹及该线索的全部历史记录。另一位复核者保存了新记录，或规则重跑后证据变化，当前表单就必须显式重新加载，避免把旧表单中的结论写到新证据上。
+
+Python 调用者编辑之前读出的发现时，也应传入这个版本：
+
+```python
+from controltrace.store import get_finding, save_review
+
+finding = get_finding("data/controltrace.duckdb", "F-2C5FD7D5834D")
+assert finding is not None
+save_review(
+    "data/controltrace.duckdb",
+    finding["finding_id"],
+    "Demo Reviewer",
+    "closed",
+    "confirmed_exception",
+    "Rechecked the current synthetic approval and deployment records.",
+    expected_revision=finding["review_revision"],
+)
+```
+
+版本核对、复核记录、证据指纹关联在同一 DuckDB 事务中提交。事务还实际递增该发现行的写入计数，使两个同时读取同一版本的连接发生写入冲突；仅执行字段值不变的更新可能被数据库优化掉，不能充当并发保护。失败会回滚复核及关联，使用者需重新读取当前记录。
+
+`expected_revision` 是可选参数，用来保留旧 Python API；省略时，调用者明确按保存时的当前自动发现追加记录，不能获得旧表单检查。界面始终传入版本。这是本机同进程内多个连接的冲突控制，不是分布式协作或多人认证服务。DuckDB 的跨进程文件写入限制仍然适用。
+
+## 旧数据与可复现底稿
+
+旧数据库没有保存复核时的证据指纹。迁移保留所有原记录，但把它们视为证据版本未知，不从今天的源数据补造历史关联；使用者要重新评估当前证据后记录新结论。
+
+新底稿采用格式 3，保留原 `reviews.json` 字段，同时增加 `review_evidence.json`：按发现和复核 ID 保存关联指纹，未知历史为 `null`。CSV 标明当前证据指纹和不适用历史数量，Markdown 明确显示当前结论与全部历史。校验器从包内源记录重放自动发现，重算指纹并独立选择当前结论；即使 CSV、底稿和文件哈希被一致修改，旧证据结论冒充当前结论仍会被拒绝。
+
+格式 2、版本 0.2.7 的旧底稿继续支持原始重放；返回 `review_evidence_bound=false`，命令行也说明旧格式没有证据版本关联。更早版本仍需其原版本校验器。这种兼容不会把旧底稿升级成拥有新保证的证据。
+
+指纹和包内清单不能证明复核者身份、源系统真实性或防御同时重写所有内容的攻击者。新包只带当前源快照，保留的历史指纹并不等于保存了旧源快照。认证、签名和历史源版本归档仍是实际业务接入前需要单独解决的工作。
+
+## 可以现场展示的验证
+
+[回归测试](../tests/test_review_evidence.py)直接修改同一发现关联源行，验证发现 ID 不变、指纹改变、旧关闭记录保留、当前状态重新待复核；还覆盖两人同版本并发、绑定失败回滚、旧数据库迁移和重新计算哈希后的伪造关闭底稿。[界面测试](../tests/test_app.py)验证旧表单禁用保存，并在使用者主动重新加载后展示当前复核记录。
+
+```bash
+uv run --extra dev pytest tests/test_review_evidence.py tests/test_app.py -q
+```
+
+讨论项目时可以从这个具体问题展开：稳定业务 ID 与证据版本不同；追加历史与当前有效结论不同；事务原子性与读取后写入的并发冲突不同。每个判断都有代码路径和失败场景，而不是依赖合成数据的命中数。

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "controltrace"
-version = "0.2.7"
+version = "0.2.8"
 description = "Synthetic IT audit evidence and internal control testing workbench"
 readme = "README.md"
 requires-python = ">=3.11"

src/controltrace/__init__.py

@@ -1,3 +1,3 @@
 """ControlTrace synthetic IT audit demonstration."""
 
-__version__ = "0.2.7"
+__version__ = "0.2.8"

src/controltrace/app.py

@@ -52,8 +52,7 @@ def _display(value: Any) -> str:
 
 
 def _load_findings() -> list[dict[str, Any]]:
-    if not DB_PATH.exists():
-        initialize_db(DB_PATH)
+    initialize_db(DB_PATH)
     findings = list_findings(DB_PATH)
     if not findings:
         run_tests(DB_PATH)
@@ -241,22 +240,38 @@ def _load_findings() -> list[dict[str, Any]]:
 
     st.subheader("Human review")
     current = selected.get("latest_review") or {}
+    if selected["review_stale_count"]:
+        st.warning(
+            f"{selected['review_stale_count']} historical review(s) concern different or unknown "
+            "evidence. Assess the current source records before recording a new decision."
+        )
     st.caption(
         f"Current: {current.get('status', 'pending')} · "
         f"{current.get('conclusion') or 'no conclusion'}"
     )
     finding_id = selected["finding_id"]
+    revision_key = f"review_base_{finding_id}"
+    if revision_key not in st.session_state:
+        st.session_state[revision_key] = selected["review_revision"]
+    base_revision = st.session_state[revision_key]
+    stale_form = base_revision != selected["review_revision"]
+    if stale_form:
+        st.warning("This finding's evidence or review changed while you were editing. Reload it.")
+        if st.button("Reload current finding", key=f"reload_{finding_id}"):
+            st.session_state[revision_key] = selected["review_revision"]
+            st.rerun()
+    input_key = f"{finding_id}_{base_revision}"
     reviewer = st.text_input(
         "Reviewer name or initials",
         value=current.get("reviewer", ""),
-        key=f"reviewer_{finding_id}",
+        key=f"reviewer_{input_key}",
     )
     status_options = ["pending", "in_review", "closed"]
     status = st.selectbox(
         "Processing status",
         status_options,
         index=status_options.index(current.get("status", "pending")),
-        key=f"review_status_{finding_id}",
+        key=f"review_status_{input_key}",
     )
     conclusion_options = {
         "pending": [None],
@@ -273,23 +288,27 @@ def _load_findings() -> list[dict[str, Any]]:
             else 0
         ),
         format_func=lambda value: value.replace("_", " ") if value else "No conclusion",
-        key=f"review_conclusion_{finding_id}_{status}",
+        key=f"review_conclusion_{input_key}_{status}",
     )
     notes = st.text_area(
         "Review notes / exception basis",
         value=current.get("notes", ""),
-        key=f"review_notes_{finding_id}",
+        key=f"review_notes_{input_key}",
     )
-    submitted = st.button("Save review", type="primary", width="stretch")
+    submitted = st.button("Save review", type="primary", width="stretch", disabled=stale_form)
     if submitted:
         if not reviewer.strip() or not notes.strip():
             st.error("Reviewer and notes are required to preserve a usable review trail.")
         else:
             try:
-                save_review(DB_PATH, finding_id, reviewer.strip(), status, conclusion, notes.strip())
+                save_review(
+                    DB_PATH, finding_id, reviewer.strip(), status, conclusion, notes.strip(),
+                    expected_revision=base_revision,
+                )
             except ValueError as error:
                 st.error(str(error))
             else:
+                del st.session_state[revision_key]
                 st.rerun()
     if selected.get("review_history"):
         with st.expander(f"Decision history ({len(selected['review_history'])})"):

src/controltrace/cli.py

@@ -43,6 +43,8 @@ def main(argv: list[str] | None = None) -> int:
             f"{result['replayed_findings']} observations "
             f"({result['exported_findings']} in this {result['scope']} export)."
         )
+        if not result["review_evidence_bound"]:
+            print("Legacy bundle: review history is valid but was not bound to an evidence snapshot.")
         return 0
     from controltrace.rules import run_tests
     from controltrace.store import initialize_db

src/controltrace/exports.py

@@ -51,6 +51,8 @@ def exception_csv(findings: list[dict[str, Any]]) -> bytes:
         "occurred_at", "audit_cutoff", "entity_type", "entity_id", "title", "summary", "rationale",
         "evidence_ids", "reviewer", "reviewed_at", "status", "conclusion", "notes",
     ]
+    if any("evidence_sha256" in finding for finding in findings):
+        fields += ["evidence_sha256", "review_stale_count"]
     rows = []
     for finding in findings:
         review = finding.get("latest_review") or {}
@@ -68,7 +70,9 @@ def exception_csv(findings: list[dict[str, Any]]) -> bytes:
     return _csv_bytes(rows, fields)
 
 
-def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str:
+def finding_workpaper(
+    finding: dict[str, Any], dataset_sha256: str, *, version: str = __version__,
+) -> str:
     """A self-contained trail from control objective to source evidence and human decision."""
     rule = finding.get("rule") or RULES[finding["control_id"]]
     review = finding.get("latest_review") or {}
@@ -79,7 +83,7 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str:
         "",
         f"- Audit cutoff: `{DEMO_CUTOFF}`",
         f"- Dataset SHA-256: `{dataset_sha256}`",
-        f"- ControlTrace version: `{__version__}`",
+        f"- ControlTrace version: `{version}`",
         f"- Control: `{finding['control_id']}` — {finding['title']}",
         f"- Classification: `{finding['classification']}`; risk: `{finding['risk']}`",
         f"- System: `{finding['system']}`; subject: `{finding['entity_type']}:{finding['entity_id']}`",
@@ -100,6 +104,8 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str:
         if isinstance(value, list):
             value = "; ".join(str(item) for item in value)
         lines.append(f"- **{label}:** {value}")
+    if "evidence_sha256" in finding:
+        lines.append(f"- Finding evidence SHA-256: `{finding['evidence_sha256']}`")
     lines += ["", "## Automated observation", "", finding["summary"], "", finding["rationale"]]
     lines += ["", "## Event timeline", ""]
     for event in finding.get("timeline", []):
@@ -121,6 +127,15 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str:
         "### Review history",
         "",
     ]
+    if "review_bindings" in finding:
+        lines += [
+            "Current decision applies only to this finding's evidence SHA-256. "
+            f"Historical decisions on different or unknown evidence: {finding['review_stale_count']}. "
+            "Their original source snapshots are not included in this bundle.",
+            "",
+            "    " + _json({"review_evidence_sha256": finding["review_bindings"]}),
+            "",
+        ]
     for item in finding.get("review_history", []):
         lines.append("    " + _json(item))
     lines += [
@@ -171,6 +186,9 @@ def workpaper_zip(db_path: str | Path, finding_ids: set[str] | None = None) -> b
         "reviews.json": _json({
             finding["finding_id"]: finding["review_history"] for finding in findings
         }).encode("utf-8"),
+        "review_evidence.json": _json({
+            finding["finding_id"]: finding["review_bindings"] for finding in findings
+        }).encode("utf-8"),
         "rules.json": _json(RULES).encode("utf-8"),
         "README.txt": (
             "Synthetic ControlTrace workpapers. Check manifest.json hashes before using "
@@ -189,7 +207,7 @@ def workpaper_zip(db_path: str | Path, finding_ids: set[str] | None = None) -> b
     })
     manifest = {
         "notice": "Synthetic case study; no real enterprise audit data",
-        "bundle_format_version": 2,
+        "bundle_format_version": 3,
         "audit_cutoff": DEMO_CUTOFF,
         "generator_seed": int(meta.get("seed", DEFAULT_SEED)),
         "controltrace_version": __version__,

src/controltrace/reviews.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 from datetime import datetime
 
 from controltrace.data import parse_utc
@@ -11,6 +13,42 @@
 REVIEW_FIELDS = {
     "review_id", "finding_id", "reviewer", "reviewed_at", "status", "conclusion", "notes",
 }
+REVIEW_ANNOTATIONS = {
+    "latest_review", "review_history", "review_status", "review_bindings",
+    "evidence_sha256", "review_revision", "review_stale_count",
+}
+
+
+def finding_fingerprint(finding: dict) -> str:
+    """Identify the exact automated observation, rule and linked source values."""
+    payload = {key: value for key, value in finding.items() if key not in REVIEW_ANNOTATIONS}
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, allow_nan=False).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def bind_review_history(finding: dict, history: list[dict], bindings: dict) -> dict:
+    """Only a decision made on the current evidence can become the current verdict.
+
+    Unbound legacy reviews remain in history, but cannot claim to have assessed
+    the present evidence. Their original evidence is unknown.
+    """
+    fingerprint = finding_fingerprint(finding)
+    applicable = [item for item in history if bindings.get(item["review_id"]) == fingerprint]
+    latest = applicable[-1] if applicable else None
+    revision = hashlib.sha256(json.dumps({
+        "evidence": fingerprint,
+        "reviews": [(item["review_id"], bindings.get(item["review_id"])) for item in history],
+    }, sort_keys=True).encode("utf-8")).hexdigest()
+    return {
+        **finding,
+        "evidence_sha256": fingerprint,
+        "review_history": history,
+        "review_bindings": {item["review_id"]: bindings.get(item["review_id"]) for item in history},
+        "latest_review": latest,
+        "review_status": latest["status"] if latest else "pending",
+        "review_stale_count": len(history) - len(applicable),
+        "review_revision": revision,
+    }
 
 
 def validate_review_decision(