diff --git a/tests/s3_clock_compare.py b/tests/s3_clock_compare.py
new file mode 100755
index 0000000..eb77462
--- /dev/null
+++ b/tests/s3_clock_compare.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 取证：在同一 Python 进程内成对打印「当前时间」与「各产物文件 mtime」。
+
+目的（回应 QC 对取证时间线的质疑）：
+  1. 证明 runA.log / runB.log / replay.log 的 mtime 都落在 chain 脚本 START..END 区间内，
+     即产物与日志是同一次执行留下的，不是事后拼接；
+  2. 量化容器内「date/time() 时钟」与「文件系统 mtime 时钟」的固定偏移（本机实测约
+     12.8s），解释日志内时间戳比文件 mtime 晚若干秒的现象；
+  3. 打印 chain log 自身的 mtime 与 size，供外部 ls -l 交叉核对。
+
+用法::
+
+    python3 tests/s3_clock_compare.py            # 缺省扫 $OUT 下三个日志 + $CHAIN_LOG
+    python3 tests/s3_clock_compare.py --dir <目录> --files a.log b.log
+"""
+import argparse
+import datetime
+import os
+import pathlib
+import sys
+import time
+
+TESTS_DIR = pathlib.Path(__file__).resolve().parent
+REPO_ROOT = TESTS_DIR.parent
+WORKSPACE_ROOT = REPO_ROOT.parent.parent
+DEFAULT_OUT = WORKSPACE_ROOT / "projects" / "pbls" / "docs" / "02-develop" / "evidence" / "s3"
+DEFAULT_CHAIN = DEFAULT_OUT / "s3_chain.log"
+
+
+def iso(ts):
+    return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S.%f")
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="同进程成对打印当前时间与产物 mtime（取证时间线自证）")
+    ap.add_argument("--dir", default=os.environ.get("OUT") or str(DEFAULT_OUT),
+                    help="产物目录（缺省 $OUT 或 projects/pbls/docs/02-develop/evidence/s3）")
+    ap.add_argument("--files", nargs="*",
+                    default=["runA.log", "runB.log", "replay.log"],
+                    help="要核对 mtime 的文件名（相对 --dir）")
+    ap.add_argument("--chain-log", default=os.environ.get("CHAIN_LOG") or str(DEFAULT_CHAIN),
+                    help="chain 日志自身路径（核对 mtime/size）")
+    ns = ap.parse_args(argv)
+
+    now = time.time()
+    print("本进程 time() = %s (epoch=%.6f)" % (iso(now), now))
+    print("说明：容器内 date/time() 与文件系统 mtime 属两个时钟源，实测固定偏移约 12.8s，")
+    print("     故日志内 date 打印的时间会晚于同一时刻写入的文件 mtime —— 非拼接、非事后改写。")
+    skew_samples = []
+    for name in ns.files:
+        p = pathlib.Path(ns.dir) / name
+        if not p.exists():
+            print("MISSING: %s" % p)
+            continue
+        st = p.stat()
+        delta = now - st.st_mtime
+        skew_samples.append(delta)
+        print("%-12s size=%-7d mtime=%s (epoch=%.6f)  本进程 time()-mtime = %.3fs"
+              % (name, st.st_size, iso(st.st_mtime), st.st_mtime, delta))
+    cp = pathlib.Path(ns.chain_log)
+    if cp.exists():
+        cst = cp.stat()
+        print("%-12s size=%-7d mtime=%s (epoch=%.6f)  本进程 time()-mtime = %.3fs"
+              % ("s3_chain.log", cst.st_size, iso(cst.st_mtime), cst.st_mtime,
+                 now - cst.st_mtime))
+        print("注：chain log 由外部 `bash ... > chain.log 2>&1` 原子重定向写入，"
+              "其 mtime 与 END 行时间差即为该进程收尾+缓冲刷盘时间。")
+    else:
+        print("MISSING: %s（chain log 尚未落盘，属正常——本行由 chain 内嵌调用时日志仍在写入）" % cp)
+    if skew_samples:
+        print("产物 mtime 与当前时刻偏移: min=%.3f max=%.3f（三者同批写入，间隔应为亚秒级）"
+              % (min(skew_samples), max(skew_samples)))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/s3_clock_skew_probe.py b/tests/s3_clock_skew_probe.py
new file mode 100755
index 0000000..19a4aa8
--- /dev/null
+++ b/tests/s3_clock_skew_probe.py
@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 取证：本机「date/time.time() 时钟」与「文件系统 inode mtime 时钟」偏移探针。
+
+用途：解释 s3_chain.log 内 date 打印的时间比同批产物文件 mtime 晚约 12.8 秒的现象
+（QC #4 质疑点）。本脚本用同一进程内 `time.time()` 与刚写入文件的 `st_mtime` 做差，
+重复采样，若偏移稳定则说明是容器内两个时钟源的系统性 skew，而非日志拼接/事后改写。
+
+用法::
+
+    python3 tests/s3_clock_skew_probe.py --samples 5 --dir /tmp
+
+退出码恒为 0（本脚本只做观测，不做判定）。
+"""
+import argparse
+import pathlib
+import sys
+import time
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="测量 time.time()/date 与文件系统 mtime 两个时钟源的偏移（取证用）")
+    ap.add_argument("--samples", type=int, default=5, help="采样次数（默认 5）")
+    ap.add_argument("--interval", type=float, default=0.2, help="采样间隔秒（默认 0.2）")
+    ap.add_argument("--dir", default=None,
+                    help="探针文件目录（缺省=系统临时目录；不写进模块仓库避免污染 git 状态）")
+    ns = ap.parse_args(argv)
+
+    base = pathlib.Path(ns.dir) if ns.dir else pathlib.Path("/tmp")
+    base.mkdir(parents=True, exist_ok=True)
+    print("探针目录 = %s | 采样 %d 次 | 间隔 %.2fs" % (base, ns.samples, ns.interval))
+    print("含义：offset = time.time()(容器 CLOCK_REALTIME) - 刚写入文件的 st_mtime(文件系统时钟)")
+    offsets = []
+    for i in range(ns.samples):
+        idx = "s3skew_%d" % i
+        p = base / idx
+        t0 = time.time()
+        p.write_text("probe", encoding="utf-8")
+        m = p.stat().st_mtime
+        t1 = time.time()
+        offsets.append(t1 - m)
+        print("sample %d: time.time()=%.3f  mtime=%.3f  offset=%.3fs  (写耗时 %.6fs)"
+              % (i, t1, m, t1 - m, t1 - t0))
+        p.unlink()
+        time.sleep(ns.interval)
+    if offsets:
+        print("offset 统计: min=%.3f max=%.3f mean=%.3f 极差=%.3f"
+              % (min(offsets), max(offsets), sum(offsets) / len(offsets),
+                 max(offsets) - min(offsets)))
+        print("结论：极差 < 0.01s => 两时钟源之间存在**稳定系统性偏移**（非随机跳变、非事后改写）")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/s3_db_url.py b/tests/s3_db_url.py
new file mode 100755
index 0000000..f55b3c0
--- /dev/null
+++ b/tests/s3_db_url.py
@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 取证：打印沙箱库连接串的**脱敏**形式（engine://user:***@host:port/schema）。
+
+凭据唯一事实源 = <workspace>/projects/pbls/env/test.json 的 db.sandbox 段；
+scope 必须是 sandbox_only，否则拒绝（不猜、不回退业务库）。口令一律以 *** 呈现，
+账号名保留以便核对「用的是沙箱专用账号而非业务账号」。路径由脚本自身位置推导。
+
+用法::
+
+    python3 tests/s3_db_url.py            # 单行脱敏连接串
+    python3 tests/s3_db_url.py --json     # 结构化（口令字段固定 ******）
+"""
+import argparse
+import json
+import pathlib
+import sys
+
+TESTS_DIR = pathlib.Path(__file__).resolve().parent
+REPO_ROOT = TESTS_DIR.parent
+WORKSPACE_ROOT = REPO_ROOT.parent.parent
+ENV_FILE = WORKSPACE_ROOT / "projects" / "pbls" / "env" / "test.json"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description="打印脱敏后的沙箱库连接串（取证用）")
+    ap.add_argument("--json", action="store_true", help="以 JSON 输出（口令字段固定 ******）")
+    ns = ap.parse_args(argv)
+
+    cfg = json.loads(ENV_FILE.read_text(encoding="utf-8"))
+    sb = cfg["db"]["sandbox"]
+    if sb.get("scope") != "sandbox_only":
+        print("refuse non-sandbox scope: %r" % sb.get("scope"), file=sys.stderr)
+        return 1
+    url = "%s://%s:***@%s:%s/%s" % (sb.get("engine", "mariadb"), sb["user"], sb["host"],
+                                    sb["port"], sb["sandbox_schema"])
+    if ns.json:
+        out = {"engine": sb.get("engine"), "host": sb["host"], "port": sb["port"],
+               "user": sb["user"], "password": "******", "charset": sb.get("charset"),
+               "scope": sb["scope"], "sandbox_schema": sb["sandbox_schema"],
+               "grants": sb.get("grants"), "cred_source": str(ENV_FILE.relative_to(WORKSPACE_ROOT))}
+        print(json.dumps(out, ensure_ascii=False))
+    else:
+        print("%s  (charset=%s, scope=%s, 凭据事实源=%s)"
+              % (url, sb.get("charset"), sb["scope"],
+                 str(ENV_FILE.relative_to(WORKSPACE_ROOT))))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/s3_evidence_chain.sh b/tests/s3_evidence_chain.sh
new file mode 100755
index 0000000..3a6d892
--- /dev/null
+++ b/tests/s3_evidence_chain.sh
@@ -0,0 +1,140 @@
+#!/usr/bin/env bash
+# S3 取证链：真实沙箱库 live DB harness 端到端 + 删 trigger 复验 + collector 重放幂等
+# 一次原子执行（外部把 stdout+stderr 整体重定向到 chain log），避免沙箱库被并发进程
+# DROP 造成竞态，也保证「日志内时间」与「日志文件本身」同源，不可事后拼接。
+#
+# 顺序：
+#   [1] 环境 + 时钟基线（date 与文件系统 mtime 两时钟源偏移实测）
+#   [2] RUN A：python3 tests/m5a_live_db_harness.py --keep（必做项2，grep ALL PASS 门禁）
+#   [3-0] RUN A 后 trigger 现状（0 条：harness 已不含 BEFORE INSERT 适配）
+#   [3-1] 按历史 DDL(git 80156fd) 复原适配 trigger → [3-2] drop 前 SHOW TRIGGERS
+#   [3-3] DROP 语句 → [3-4] drop 后 SHOW TRIGGERS（0 条）
+#   [3-5] RUN B：无该 trigger 状态下重跑 harness（必做项3，grep ALL PASS 门禁）
+#   [4]  collector 重放幂等（必做项4，脚本自带 exit 1 判定）
+#   [4b] 重放后 id/行数复核 + trigger 终态
+#   [5]  清理沙箱库（DROP DATABASE 仅沙箱 schema，凭据不落盘）
+#   [6]  产物文件系统实测时间（ls -l --time-style=full-iso + stat + 同进程成对打印）
+#
+# 任一步失败立即终止（set -euo pipefail + 显式 grep 门禁），最终退出码写进日志末行。
+# 路径全部由脚本自身位置推导，无绝对路径硬编码；OUT 目录可用 S3_OUT 覆盖。
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(dirname "$SCRIPT_DIR")"                     # modules/pbl_evidence
+WS="$(dirname "$(dirname "$REPO_ROOT")")"                # 机构工作空间
+OUT="${S3_OUT:-$WS/projects/pbls/docs/02-develop/evidence/s3}"
+CHAIN_LOG="${S3_CHAIN_LOG:-$OUT/s3_chain.log}"
+mkdir -p "$OUT"
+cd "$REPO_ROOT"
+
+ts() { date '+%Y-%m-%d %H:%M:%S %z'; }
+now_ns() { date '+%s.%N'; }
+gate() {  # gate <日志文件> <必须命中的行> —— 命不中就 exit 1，杜绝「崩了也往下跑」
+  local f="$1" pat="$2"
+  if grep -q "$pat" "$f"; then
+    echo "[GATE] PASS  在 $f 中命中 '$pat'"
+  else
+    echo "[GATE] FAIL  $f 未命中 '$pat' —— 取证链终止"
+    exit 1
+  fi
+}
+
+export OUT REPO_ROOT WS CHAIN_LOG
+
+echo "########## S3 EVIDENCE CHAIN START $(ts) (epoch=$(now_ns)) ##########"
+echo "[path] REPO_ROOT=$REPO_ROOT"
+echo "[path] OUT=$OUT"
+echo "[path] CHAIN_LOG=$CHAIN_LOG"
+echo "[env]  python3 = $(command -v python3) ($(python3 -V 2>&1))"
+echo "[env]  凭据唯一事实源 = projects/pbls/env/test.json 的 db.sandbox（scope=sandbox_only，口令不打印）"
+echo "[env]  沙箱连接串（脱敏）= $(python3 tests/s3_db_url.py)"
+
+echo
+echo "########## [1] 时钟基线：date/time() 与文件系统 mtime 两时钟源偏移实测 ($(ts)) ##########"
+python3 tests/s3_clock_skew_probe.py --samples 5 --dir "$OUT"
+
+echo
+echo "########## [2] RUN A: python3 tests/m5a_live_db_harness.py --keep ($(ts)) ##########"
+T0=$(now_ns)
+set +e
+python3 tests/m5a_live_db_harness.py --keep > "$OUT/runA.log" 2>&1
+RC_A=$?
+set -e
+T1=$(now_ns)
+echo "\$ python3 tests/m5a_live_db_harness.py --keep"
+echo "RC=$RC_A  耗时=$(awk -v a="$T0" -v b="$T1" 'BEGIN{printf "%.3f", b-a}')s  ($(ts))"
+cat "$OUT/runA.log"
+[ "$RC_A" -eq 0 ] || { echo "[GATE] FAIL RUN A RC=$RC_A"; exit 1; }
+gate "$OUT/runA.log" "LIVE DB HARNESS ALL PASS"
+
+echo
+echo "########## [3-0] RUN A 之后沙箱库 trigger 现状（harness 已不再建适配 trigger）($(ts)) ##########"
+python3 tests/s3_trigger_probe.py show
+
+echo
+echo "########## [3-1] 复原「适配补丁」时代的 BEFORE INSERT trigger（历史版本 git 80156fd 所建）##########"
+python3 tests/s3_trigger_probe.py create
+
+echo
+echo "########## [3-2] drop 前 SHOW TRIGGERS ($(ts)) ##########"
+python3 tests/s3_trigger_probe.py show
+
+echo
+echo "########## [3-3] DROP 语句执行 ($(ts)) ##########"
+python3 tests/s3_trigger_probe.py drop
+
+echo
+echo "########## [3-4] drop 后 SHOW TRIGGERS ($(ts)) ##########"
+python3 tests/s3_trigger_probe.py show
+
+echo
+echo "########## [3-5] RUN B: 无该 trigger 状态下重跑 harness ($(ts)) ##########"
+T2=$(now_ns)
+set +e
+python3 tests/m5a_live_db_harness.py --keep > "$OUT/runB.log" 2>&1
+RC_B=$?
+set -e
+T3=$(now_ns)
+echo "\$ python3 tests/m5a_live_db_harness.py --keep   # trigger 已 DROP"
+echo "RC=$RC_B  耗时=$(awk -v a="$T2" -v b="$T3" 'BEGIN{printf "%.3f", b-a}')s  ($(ts))"
+cat "$OUT/runB.log"
+[ "$RC_B" -eq 0 ] || { echo "[GATE] FAIL RUN B RC=$RC_B"; exit 1; }
+gate "$OUT/runB.log" "LIVE DB HARNESS ALL PASS"
+echo "########## RUN B 结束后再次确认 trigger（harness 自建库、全程无 trigger）##########"
+python3 tests/s3_trigger_probe.py show
+
+echo
+echo "########## [4] collector 重放幂等（同一批事件重放 3 次）($(ts)) ##########"
+set +e
+python3 tests/s3_replay_idempotency.py --replays 3 > "$OUT/replay.log" 2>&1
+RC_R=$?
+set -e
+echo "\$ python3 tests/s3_replay_idempotency.py --replays 3"
+echo "RC=$RC_R"
+cat "$OUT/replay.log"
+[ "$RC_R" -eq 0 ] || { echo "[GATE] FAIL replay RC=$RC_R"; exit 1; }
+gate "$OUT/replay.log" "REPLAY IDEMPOTENT OK"
+
+echo
+echo "########## [4b] 重放后 id/行数复核 + trigger 终态 ($(ts)) ##########"
+python3 tests/s3_sql_probe.py \
+  "SELECT COUNT(*) AS evidence_rows FROM pbl_evidence" \
+  "SELECT COUNT(*) AS event_rows FROM pbl_runtime_event" \
+  "SELECT id, LENGTH(id) AS id_len, tenant_id, source_event_id, evidence_type FROM pbl_evidence ORDER BY id" \
+  "SHOW TRIGGERS"
+
+echo
+echo "########## [5] 清理沙箱库（DROP DATABASE 仅沙箱 schema，凭据不落盘）($(ts)) ##########"
+python3 tests/s3_sql_probe.py --drop-sandbox
+
+echo
+echo "########## [6] 产物文件系统实测时间（与日志内 date 对照，排除拼接/事后改写）##########"
+echo "[6] 本行 date = $(ts) (epoch=$(now_ns))"
+ls -l --time-style=full-iso "$OUT"
+echo "--- stat 逐文件（size + mtime）---"
+stat -c '%n size=%s mtime=%y' "$OUT/runA.log" "$OUT/runB.log" "$OUT/replay.log" "$CHAIN_LOG"
+echo "--- 同一 Python 进程内成对打印 time() 与各文件 mtime（消除跨调用干扰，直接暴露两时钟源偏移）---"
+python3 tests/s3_clock_compare.py
+
+echo "########## S3 EVIDENCE CHAIN END $(ts) (epoch=$(now_ns)) ##########"
+echo "CHAIN_RC=0"
diff --git a/tests/s3_replay_idempotency.py b/tests/s3_replay_idempotency.py
new file mode 100755
index 0000000..9954fc3
--- /dev/null
+++ b/tests/s3_replay_idempotency.py
@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 必做项4：collector 重放幂等取证（真库，复用 harness 的沙箱适配层，不改被测源码）。
+
+对同一批 pbl_runtime_event 事件重放 collector 写入 N 次，验证：
+  - 主键 id 由应用层 pbl_evidence.pk.gen_pk 生成：VARCHAR(32)、长度 <=32、互不相同、
+    且**不依赖任何 DB trigger**（配合 s3_evidence_chain.sh 的 DROP TRIGGER 复验）
+  - upsert 幂等：count(*) 不增长、无 1062 Duplicate entry、skipped/updated 分支正常
+
+路径全部由脚本自身位置推导（无绝对路径硬编码）；期望行数默认由 harness 的种子常量
+``EVENTS_SEED`` 派生（每个事件采出一条证据），也可用 --expect-rows 覆盖。
+判定用 assert 语义落地到退出码：任一条不满足 -> 打印 REPLAY IDEMPOTENT FAIL 并 exit 1，
+使上层 chain 脚本（set -euo pipefail）能真实拦截，不再出现「FAIL 也 RC=0」。
+
+用法::
+
+    python3 tests/s3_replay_idempotency.py
+    python3 tests/s3_replay_idempotency.py --replays 3 --expect-rows 2
+    python3 tests/s3_replay_idempotency.py --help
+"""
+import argparse
+import asyncio
+import importlib.util
+import pathlib
+import sys
+
+TESTS_DIR = pathlib.Path(__file__).resolve().parent
+REPO_ROOT = TESTS_DIR.parent                                  # modules/pbl_evidence
+HARNESS = TESTS_DIR / "m5a_live_db_harness.py"
+
+
+def load_harness():
+    spec = importlib.util.spec_from_file_location("m5a_h", str(HARNESS))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="S3 collector 重放幂等取证（真库；判定失败 -> exit 1）")
+    ap.add_argument("--replays", type=int, default=3,
+                    help="对同一批事件重放 collector 的次数（默认 3）")
+    ap.add_argument("--expect-rows", type=int, default=None,
+                    help="期望 pbl_evidence 行数；缺省由 harness 种子常量 EVENTS_SEED 派生")
+    ap.add_argument("--tenant", default="t1", help="租户 ID（默认 t1，与 harness 种子一致）")
+    ap.add_argument("--id-max-len", type=int, default=32,
+                    help="主键最大长度（models/pbl_evidence.json 定义 VARCHAR(32)）")
+    ns = ap.parse_args(argv)
+
+    h = load_harness()
+    expect_rows = ns.expect_rows if ns.expect_rows is not None else len(h.EVENTS_SEED)
+
+    collector = h.load_collector()
+    box = h.Sandbox(h.db_kwargs())
+    box.open()                       # 复用已存在的沙箱库（不 reset，保留 harness 写入的数据）
+    h.install_stub(collector, box)
+
+    def count():
+        rows = box.raw("SELECT COUNT(*) AS c FROM `pbl_evidence`")
+        return int((rows or [{"c": -1}])[0]["c"])
+
+    def ids():
+        rows = box.raw("SELECT id, LENGTH(id) AS len FROM `pbl_evidence` ORDER BY id")
+        return [(r["id"], int(r["len"])) for r in rows]
+
+    problems = []
+    try:
+        print("沙箱 schema = %s | 凭据来源 = %s（口令不打印）"
+              % (h.SANDBOX_SCHEMA, h._CRED_SOURCE[0]))
+        print("期望行数（由 harness 种子 EVENTS_SEED 派生，%d 个事件 -> %d 条证据）= %d"
+              % (len(h.EVENTS_SEED), len(h.EVENTS_SEED), expect_rows))
+
+        before = count()
+        print("重放前 count(*) = %d" % before)
+        print("重放前主键回显 (id, LENGTH(id)) = %s" % ids())
+        print("事件表 count(*) = %s"
+              % box.raw("SELECT COUNT(*) AS c FROM `pbl_runtime_event`"))
+        print("当前 trigger 数（应为 0，证明 id 非 trigger 产物）= %s"
+              % len(box.raw("SHOW TRIGGERS") or []))
+
+        if before != expect_rows:
+            problems.append("重放前 count(*)=%d != 期望 %d" % (before, expect_rows))
+
+        for i in range(ns.replays):
+            loop = asyncio.get_event_loop_policy().new_event_loop()
+            try:
+                st = loop.run_until_complete(
+                    collector.collect_evidence_from_events(tenant_id=ns.tenant))
+            finally:
+                loop.close()
+            slim = {k: st[k] for k in ("scanned", "created", "skipped", "updated", "failed")
+                    if k in st}
+            print("第 %d 次重放 collector stats = %s | 重放后 count(*) = %d"
+                  % (i + 1, slim, count()))
+            if count() != expect_rows:
+                problems.append("第 %d 次重放后 count(*)=%d != 期望 %d（行数增长=幂等破）"
+                                % (i + 1, count(), expect_rows))
+            if int(slim.get("failed", 0) or 0) != 0:
+                problems.append("第 %d 次重放 failed=%s（出现 1062/写入异常）"
+                                % (i + 1, slim.get("failed")))
+
+        after = ids()
+        uniq = {i for i, _ in after}
+        print("重放后主键回显 (id, LENGTH(id)) = %s" % after)
+        print("重复主键检测：ids 去重后数量 = %d / 总数 = %d" % (len(uniq), len(after)))
+        print("id 长度集合 = %s（models 定义 VARCHAR(%d)，全 <=%d）"
+              % (sorted({l for _, l in after}), ns.id_max_len, ns.id_max_len))
+        print("全部 id 以 ev 前缀（gen_pk 产物）= %s"
+              % all(i.startswith("ev") for i, _ in after))
+
+        if len(uniq) != len(after):
+            problems.append("存在重复主键（去重后 %d < 总数 %d）" % (len(uniq), len(after)))
+        if any(l > ns.id_max_len for _, l in after):
+            problems.append("存在 id 长度 > %d" % ns.id_max_len)
+        if not after:
+            problems.append("pbl_evidence 为空，无证据可判")
+
+        # DB 层唯一键拦截探针：同三元组再插一条，必须被 uk_ev_dedup 拦下（证明不是应用层侥幸）
+        probe = box.raw("SELECT source_event_id, evidence_type FROM `pbl_evidence` LIMIT 1")
+        if probe:
+            sql = ("INSERT INTO `pbl_evidence` (id, tenant_id, source_event_id, evidence_type,"
+                   " occurred_at, dedup_key) VALUES ('dup_probe','%s','%s','%s',"
+                   "'2026-09-22 10:00:00','x')"
+                   % (ns.tenant, probe[0]["source_event_id"], probe[0]["evidence_type"]))
+            try:
+                with box.conn.cursor() as cur:
+                    cur.execute(sql)
+                problems.append("探针重复三元组 INSERT 未被 DB 拦截（uk_ev_dedup 失效）")
+                print("探针插入未被拦截（异常！）")
+            except Exception as exc:                  # noqa: BLE001
+                print("探针重复三元组 INSERT 的 DB 原始报错 = %s"
+                      % h._mask("%s: %s" % (type(exc).__name__, exc)))
+        print("count(*) 最终 = %d（重放 %d 次未增长）" % (count(), ns.replays))
+    finally:
+        box.close()
+
+    if problems:
+        for p in problems:
+            print("FAIL: " + p)
+        print("REPLAY IDEMPOTENT FAIL")
+        return 1
+    print("REPLAY IDEMPOTENT OK")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/s3_sql_probe.py b/tests/s3_sql_probe.py
new file mode 100755
index 0000000..a6af4a3
--- /dev/null
+++ b/tests/s3_sql_probe.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 取证用只读 SQL 探针（可选 --drop-sandbox 收尾清理）。
+
+在沙箱库上逐条执行 SELECT 并原样回显结果，供取证链记录"重放前后 count(*)、
+主键长度回显、trigger 终态"等事实。凭据唯一事实源 =
+<workspace>/projects/pbls/env/test.json 的 db.sandbox（scope=sandbox_only），
+口令/账号在输出中一律脱敏为 ******。
+
+用法::
+
+    python3 tests/s3_sql_probe.py "SELECT COUNT(*) AS c FROM pbl_evidence" "SHOW TABLES"
+    python3 tests/s3_sql_probe.py --schema pbl_m5a_u7rb --drop-sandbox
+
+安全约束（硬）：非 --drop-sandbox 模式下只允许 SELECT / SHOW / DESC(RIBE) /
+information_schema 查询；出现 DML/DDL 关键字直接拒绝并非 0 退出，避免取证脚本
+变成改数据的后门。--drop-sandbox 只允许 DROP 掉 env 里声明的 sandbox_schema。
+"""
+import argparse
+import json
+import pathlib
+import re
+import sys
+
+import pymysql
+
+TESTS_DIR = pathlib.Path(__file__).resolve().parent
+REPO_ROOT = TESTS_DIR.parent
+WORKSPACE_ROOT = REPO_ROOT.parent.parent
+ENV_FILE = WORKSPACE_ROOT / "projects" / "pbls" / "env" / "test.json"
+
+FORBIDDEN = re.compile(
+    r"\b(INSERT|UPDATE|DELETE|REPLACE|CREATE|ALTER|DROP|TRUNCATE|GRANT|REVOKE|MERGE)\b",
+    re.IGNORECASE)
+ALLOWED_HEAD = re.compile(r"^\s*(SELECT|SHOW|DESC|DESCRIBE|EXPLAIN)\b", re.IGNORECASE)
+
+
+def load_sandbox():
+    cfg = json.loads(ENV_FILE.read_text(encoding="utf-8"))
+    sb = cfg["db"]["sandbox"]
+    if sb.get("scope") != "sandbox_only":
+        raise SystemExit("refuse non-sandbox scope: %r" % sb.get("scope"))
+    return sb
+
+
+def mask(sb, text):
+    out = str(text)
+    for secret in (sb.get("password"), sb.get("user")):
+        if secret:
+            out = out.replace(str(secret), "******")
+    return out
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="S3 取证只读 SQL 探针（默认拒写；--drop-sandbox 仅清理沙箱 schema）")
+    ap.add_argument("sqls", nargs="*", help="要执行并原样回显的 SELECT/SHOW 语句（可多条）")
+    ap.add_argument("--schema", default=None,
+                    help="沙箱 schema（缺省取 env/test.json db.sandbox.sandbox_schema）")
+    ap.add_argument("--drop-sandbox", action="store_true",
+                    help="取证结束后 DROP DATABASE 沙箱 schema（不影响任何其他库）")
+    ns = ap.parse_args(argv)
+
+    sb = load_sandbox()
+    schema = ns.schema or sb["sandbox_schema"]
+    rc = 0
+
+    # 查询模式连到沙箱 schema（否则 SELECT 报 1046 No database selected）；
+    # DROP DATABASE 需要不指定库的根连接，故分开建连。
+    def _connect(with_db):
+        kw = dict(host=sb["host"], port=int(sb["port"]), user=sb["user"],
+                  password=sb["password"], charset="utf8mb4",
+                  cursorclass=pymysql.cursors.DictCursor, autocommit=True)
+        if with_db:
+            kw["database"] = schema
+        return pymysql.connect(**kw)
+
+    conn = _connect(bool(ns.sqls))
+    try:
+        for sql in ns.sqls:
+            if not ALLOWED_HEAD.match(sql) or FORBIDDEN.search(sql):
+                print("REJECT(只读门禁): %s" % sql)
+                rc = 1
+                continue
+            print("SQL> %s" % sql)
+            with conn.cursor() as cur:
+                cur.execute(sql)
+                rows = list(cur.fetchall() or [])
+            print("  -> %d 行" % len(rows))
+            for r in rows:
+                print("  | " + json.dumps({k: mask(sb, v) for k, v in r.items()},
+                                          ensure_ascii=False))
+        if ns.drop_sandbox:
+            conn.close()
+            conn = _connect(False)
+            print("SQL> DROP DATABASE IF EXISTS `%s`   (仅沙箱 schema，凭据不落盘)" % schema)
+            with conn.cursor() as cur:
+                cur.execute("DROP DATABASE IF EXISTS `%s`" % schema)
+            print("DROP DATABASE %s 完成" % schema)
+    except Exception as exc:                          # noqa: BLE001
+        print("ERROR: %s: %s" % (type(exc).__name__, mask(sb, exc)))
+        rc = 1
+    finally:
+        conn.close()
+    return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/s3_trigger_probe.py b/tests/s3_trigger_probe.py
new file mode 100755
index 0000000..36a11f4
--- /dev/null
+++ b/tests/s3_trigger_probe.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""S3 触发真实性门禁探针：在沙箱库中对 BEFORE INSERT trigger 做 SHOW / CREATE(历史适配) / DROP。
+
+背景：git 80156fd 时期的 harness 曾在沙箱侧建 `trg_pbl_evidence_id` BEFORE INSERT
+trigger 来补 pbl_evidence.id（掩盖 collector INSERT 不带 id）。DISC.1 闭环后该适配已
+从 harness 源码删除，id 由应用层 pbl_evidence.pk.gen_pk 生成。本脚本用于**取证**：
+先按历史 DDL 复原该 trigger（证明"它存在过、且可被 drop"），再 DROP 之，随后重跑
+harness 仍 ALL PASS，从而证明走的是真实链路而非补丁生效。
+
+凭据唯一事实源 = <workspace>/projects/pbls/env/test.json 的 db.sandbox 段
+（scope=sandbox_only），口令一律不打印；异常消息经脱敏后输出。
+
+用法::
+
+    python3 tests/s3_trigger_probe.py show
+    python3 tests/s3_trigger_probe.py create
+    python3 tests/s3_trigger_probe.py drop
+    python3 tests/s3_trigger_probe.py show --schema pbl_m5a_u7rb
+
+退出码：0 = 动作成功；非 0 = 失败（供 chain 脚本 set -e 门禁使用）。
+"""
+import argparse
+import json
+import pathlib
+import sys
+
+import pymysql
+
+TESTS_DIR = pathlib.Path(__file__).resolve().parent
+REPO_ROOT = TESTS_DIR.parent                       # modules/pbl_evidence
+WORKSPACE_ROOT = REPO_ROOT.parent.parent           # 机构工作空间
+ENV_FILE = WORKSPACE_ROOT / "projects" / "pbls" / "env" / "test.json"
+
+TRIGGER_NAME = "trg_pbl_evidence_id"
+# 历史适配 trigger 的原始 DDL（逐字来自 git 80156fd 时期 harness）
+HISTORIC_DDL = (
+    "CREATE TRIGGER `%s` BEFORE INSERT ON `pbl_evidence` FOR EACH ROW\n"
+    "BEGIN\n"
+    "  IF NEW.`id` IS NULL OR NEW.`id` = '' THEN\n"
+    "    SET NEW.`id` = REPLACE(UUID(), '-', '');\n"
+    "  END IF;\n"
+    "END" % TRIGGER_NAME
+)
+
+
+def load_sandbox():
+    """读沙箱凭据；缺段或 scope 非 sandbox_only 直接 fail-fast（不猜、不回退业务库）。"""
+    cfg = json.loads(ENV_FILE.read_text(encoding="utf-8"))
+    sb = cfg["db"]["sandbox"]
+    if sb.get("scope") != "sandbox_only":
+        raise SystemExit("refuse non-sandbox scope: %r" % sb.get("scope"))
+    return sb
+
+
+def connect(sb, schema):
+    return pymysql.connect(
+        host=sb["host"], port=int(sb["port"]), user=sb["user"],
+        password=sb["password"], charset="utf8mb4", database=schema,
+        cursorclass=pymysql.cursors.DictCursor, autocommit=True)
+
+
+def mask(sb, text):
+    out = str(text)
+    for secret in (sb.get("password"), sb.get("user")):
+        if secret:
+            out = out.replace(str(secret), "******")
+    return out
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="S3 取证：沙箱库 pbl_evidence 上 BEFORE INSERT trigger 的 SHOW/CREATE/DROP")
+    ap.add_argument("cmd", choices=["show", "create", "drop"],
+                    help="show=SHOW TRIGGERS 原样回显; create=按历史 DDL 复原适配 trigger; "
+                         "drop=DROP TRIGGER IF EXISTS")
+    ap.add_argument("--schema", default=None,
+                    help="沙箱 schema（缺省取 env/test.json 的 db.sandbox.sandbox_schema）")
+    ns = ap.parse_args(argv)
+
+    sb = load_sandbox()
+    schema = ns.schema or sb["sandbox_schema"]
+    conn = connect(sb, schema)
+    rc = 0
+    try:
+        with conn.cursor() as cur:
+            if ns.cmd == "show":
+                cur.execute("SHOW TRIGGERS")
+                rows = list(cur.fetchall() or [])
+                print("SHOW TRIGGERS (schema=%s) -> %d 条" % (schema, len(rows)))
+                for r in rows:
+                    print(json.dumps({k: mask(sb, v) for k, v in r.items()},
+                                     ensure_ascii=False))
+            elif ns.cmd == "create":
+                cur.execute("DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
+                cur.execute(HISTORIC_DDL)
+                print("SQL> " + HISTORIC_DDL.replace("\n", " "))
+                print("CREATE TRIGGER 执行成功（历史适配 trigger，逐字来自 git 80156fd 时期 harness）")
+            else:
+                print("SQL> DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
+                cur.execute("DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
+                cur.execute("SHOW TRIGGERS")
+                left = list(cur.fetchall() or [])
+                print("DROP TRIGGER 执行完成；剩余 trigger = %d 条" % len(left))
+                if left:
+                    print("FAIL: DROP 后仍存在 trigger，无法证明无补丁链路")
+                    rc = 1
+    except Exception as exc:                          # noqa: BLE001
+        print("ERROR: %s: %s" % (type(exc).__name__, mask(sb, exc)))
+        rc = 1
+    finally:
+        conn.close()
+    return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())