Merge pull request 'Add isolated U7 drills for pgbench, disconnect, and backup restore' (#491) from feature/u7-ops-drills into main

2026-10-12 16:45:55 +00:00
parent df5f8761ea 047bb4ce7f
commit 089862084f
17 changed files with 3174 additions and 8 deletions
@@ -28,7 +28,7 @@
 | B3 | 60 分钟稳定负载 + 2 小时混合长稳 | `sustained60m` results `20261012-070510`（读 P95≈3.9ms）；`sustained2h` results `20261012-081025`（读 P95≈6.2ms）；均 fail=0、dup_seq=0 | ✅ SELF（2026-10-12）；非 GATE/REPLAY |
 | B4 | SSE 持久事件可见延迟 P95≤1s | 同上套件 sse_probe：ladder A/B ≈17.5/33.0ms；60m ≈5.7/4.4ms；2h ≈3.9/4.2ms | ✅ SELF（随 B2/B3） |
 | B5 | 数据量档（100 万消息/10 万知识分块/多项目交叉授权） | 三段均 `--seed contract`（manifest：1e6 messages / 1e5 knowledge） | ✅ SELF（2026-10-12） |
-| B6 | 数据库基线（pgbench 只测 DB 能力，不替代业务验收） | `pgbench` 对隔离实例（只读参考） | ❌ 未执行 |
+| B6 | 数据库基线（pgbench 只测 DB 能力，不替代业务验收） | `tests/load/collaboration/run_pgbench.sh`；results `pgbench-20261012-131948`（scale 100，60s，`-M prepared`；100 客户端按 `max_connections` 余量封顶为 95）。select-only 10/30/95 的 P95 为 0.024/0.057/0.180 ms；TPC-B 类 10/30/95 的 P95 为 0.673/1.002/3.361 ms。失败事务 0。主机并不空闲，数字只作参考 | ✅ SELF（2026-10-12）；无业务阈值；非 GATE/REPLAY |
 | B7 | 容量数字绑定组件 SHA/配置/数据集归档 | [容量报告](../modules/documents/docs/records/2026-10-12-u7-capacity-window-report.md) + worktree `results/<ts>/` + 任务卡 evidence | ✅ SELF 报告已写 |

 ## C. 双进程与双主机放行（分布式出口）
@@ -38,15 +38,15 @@
 | C1 | 双进程共享 PG：正确性矩阵全过（A6） | `make test-multi-instance` | ✅ |
 | C2 | 双进程**负载**下互相不拖累（热点会话/命令时延） | 上表 B2/B3 均 `--instances 2`（VU 粘滞 + 跨实例读 + 热点 409 入 expected_contention；非热点读阈值全过） | ✅ SELF（2026-10-12）；非 GATE/REPLAY |
 | C3 | 双主机（跨机网络）验证：同套验收在两台主机执行 | A6 的 AGENT_BIN/DSN 指向远端；需用户指定第二主机 | ❌ 待资源 |
-| C4 | 故障演练：PG 断连/Redis 断连恢复（RTO≤15min） | backend ratelimit/recovery 测试 + 运维手册演练记录 | ⏳ 代码级 ✅；运维演练待窗口 |
-| C5 | 备份恢复演练（RPO≤5min/RTO≤60min，pg_dump+物理备份） | U4 迁移演练已过；定期备份恢复演练属运维，待排期 | ⏳ |
+| C4 | 故障演练：PG 断连/Redis 断连恢复（RTO≤15min） | `tests/ops/run_fault_drill.sh --fault all`；results `fault-20261012-142734`。pg-stop 与 pg-kill 的 F1–F5 均过，RTO 不到 1 秒且未重启应用。Redis：共享限额 20，断连期间按实例合计约 40，拉起后回到 20；无 5xx。G4 降级不可见、G5 启动时 Redis 不通则拒绝，记为发现 | ✅ SELF（2026-10-12）；非 GATE/REPLAY |
+| C5 | 备份恢复演练（RPO≤5min/RTO≤60min，pg_dump+物理备份） | `tests/ops/run_backup_drill.sh`；results `backup-20261012-153441`。K1/K2/K3/K5/K6/K7 过：RPO 64s，RTO 4s，`pg_verifybackup` 与 `pg_restore --list` 通过，逻辑恢复计数等于 dump 时快照，`failed_count=0`。K4 FAIL：未配置 `BLOBSTORE_S3_ENDPOINT` 时上传在空的 `*S3` 上 panic，工件 sha256 与跨用户拒绝未测成。静态加密、凭据演练记 N/A；归档与数据同盘 | ⏳ SELF 部分通过（2026-10-12）；K4 FAIL；非 GATE/REPLAY |

 ## 放行结论（当前）

 - **单实例团队档**：A1–A7 全绿 + B1–B5/B7 本机 SELF 容量窗口全过（见容量报告）。
  功能验收成立；**容量数字仅为 SELF**，换机/真实模型须重测。生产承诺仍待独立 REPLAY 或 GATE。
- **双进程档**：正确性已证（A5/A6/C1），本机双实例负载 SELF 已过（C2）。
-  **仍不放行第二实例承接写请求**——缺独立 REPLAY/GATE，且 C3/C4 未过。
+- **双进程档**：正确性已证（A5/A6/C1），本机双实例负载 SELF 已过（C2）。C4 断连演练 SELF 已过。
+  **仍不放行第二实例承接写请求**——缺独立 REPLAY/GATE，C3 未做，且 C5 的工件恢复（K4）失败。
 - **双主机档**：未开始（C3），不放行。

 ## 已知边界与诚实声明
@@ -58,4 +58,4 @@
 - 容量详情与 SHA 绑定见 Documents
  `docs/records/2026-10-12-u7-capacity-window-report.md`；原始
  `results/<ts>/` 在测量 worktree（gitignore），不入库。
- B6 pgbench、C3 双主机、C4/C5 运维演练仍待资源/排期。
+- B6 pgbench 与 C4 断连演练已有本机 SELF。C5 的库恢复 SELF 已过，工件上传 K4 失败。C3 双主机仍待第二台机器。证据等级仍缺独立 REPLAY 或 CI GATE。
@@ -685,6 +685,68 @@ scripts/svc.sh db copy pgdev pgprod --db app --force    # 目标非空时覆盖
  并记入标签（如 `--tag port-8896`）；
 - 换端口用 `set --apply`；克隆用 `--port` 指定；两者都会做占用双重校验。

+### 8.6 物理备份 + WAL 归档 / PITR 到新目标
+
+`db backup` 只做 `pg_dump -Fc`。物理备份、WAL 归档和按时间点恢复到新数据目录不在 `svc.sh` 里，2026-10-12 在隔离实例 `u7bk` 上实跑过，入口是 `tests/ops/run_backup_drill.sh`，结果目录 `tests/ops/results/backup-20261012-153441`（SELF）。合同数据量（100 万消息 / 10 万知识分块）下：基础备份与 `pg_dump` 各约 1 秒，逻辑恢复约 4 秒，从 `pg_ctl stop -m immediate` 到恢复后应用冒烟通过约 4 秒，最后一条仍在库里的标记比故障时刻早 64 秒。
+
+在源实例上打开归档后必须重启（`wal_level` 与 `archive_mode` 不热加载）：
+
+```bash
+psql -d agent -c "ALTER SYSTEM SET wal_level = 'replica'" \
+  -c "ALTER SYSTEM SET archive_mode = 'on'" \
+  -c "ALTER SYSTEM SET archive_timeout = '60s'" \
+  -c "ALTER SYSTEM SET archive_command = 'test ! -f /secure/wal/%f && cp %p /secure/wal/%f && chmod 600 /secure/wal/%f'"
+scripts/svc.sh u7bk stop && scripts/svc.sh u7bk start
+```
+
+归档目录先 `chmod 700`。备份：
+
+```bash
+pg_basebackup -h 127.0.0.1 -p <源端口> -U agent -D /secure/base -Fp -X stream -c fast
+pg_verifybackup /secure/base
+scripts/svc.sh db backup u7bk --db <应用库> --out /secure/logical.dump
+pg_restore --list /secure/logical.dump
+```
+
+演练里 `pg_stat_archiver.failed_count` 为 0，`archived_count` 为 80。备份目录 700、文件 600。静态加密和凭据轮换这次没有做。归档和数据在同一块盘上，不能代表存储整盘损坏。
+
+制造故障用的是 `pg_ctl -D <pgdata> stop -m immediate`，然后把源数据目录挪走。恢复到新目录：
+
+```bash
+cp -a /secure/base/. /secure/restore-pgdata/
+# postgresql.auto.conf 末尾追加（后写的覆盖前面的 archive_mode）：
+#   archive_mode = 'off'
+#   restore_command = 'cp /secure/wal/%f %p'
+touch /secure/restore-pgdata/recovery.signal
+rm -f /secure/restore-pgdata/postmaster.pid
+pg_ctl -D /secure/restore-pgdata -l /secure/postgres.log -t 600 -w \
+  -o "-p <新端口> -k /tmp/u7bk-18997" start
+```
+
+`-k` 必须是短路径。结果目录下的 socket 路径超过 107 字节时，postmaster 会拒绝启动（同日 `backup-20261012-150713` 的 `restore/postgres.log`）。等到 `SELECT pg_is_in_recovery()` 返回 `f` 再让应用连这个新端口。逻辑路径是在恢复实例上建空库，再 `pg_restore --no-owner`；表计数与 dump 当时的快照一致。
+
+工件正文的库内 `content_path` 是绝对路径（`blobstore.Local.Put` 返回最终文件路径）。快照要还原到原来的目录，换目录后按库里的路径取文件会失败。这次演练没有把这一点跑通：未设置 `BLOBSTORE_S3_ENDPOINT` 时，`app.go` 把空的 `*S3` 传进 `OpenPostgresWithOptions`，接口值不是 nil，`finishOpen` 不再回落本地目录，上传在 `blobstore.(*S3).Put` 空指针 panic。K4（上传文件经 API 下载后 sha256 一致、跨用户拒绝）记 FAIL，产品代码未改。
+
+### 8.7 PG / Redis 断连处置
+
+2026-10-12 隔离实例 `u7drill` 实跑，入口 `tests/ops/run_fault_drill.sh --fault all`，结果目录 `tests/ops/results/fault-20261012-142734`（SELF）。PostgreSQL 用 `svc.sh <name> stop`（`pg_ctl -m fast`）和直接对自有 postmaster 发 `SIGKILL` 各做一次，断连窗口 60 秒。Redis 用 `--save '' --appendonly no` 拉起后再 `SIGKILL`。
+
+PostgreSQL：
+
+- 断连期间写入在 10 秒内返回明确错误（这次是 401），两个应用进程都还在。窗口从停机返回后再等 2 秒才开始计，避免把已经在途的成功写算进去。
+- team 档新进程日志为「拒绝就绪（不回落 SQLite）」，工作目录不出现 `*.db`。
+- `/api/health` 不探数据库。恢复是否完成，用新会话上的真实写入判断：同一会话上连续写会被进行中的 mock run 以 409 挡住，那不是库没起来。这次从 postgres 就绪到两侧各 5 次写入成功不到 1 秒，没有重启应用。
+- 断连窗口里失败的 requestId 用同一 id 重放后各对应一条消息；`dup_seq=0`，seq 无洞，非终态 run 为 0。
+
+Redis（限额 20 次/分钟，两实例各突发 40 个读）：
+
+- 连通时两实例合计放行 20、拒绝 60，限额是共享的。
+- 进程被杀掉之后合计放行 40、拒绝 40，没有 5xx。限额变成按实例计。已确认的写入仍在库里。
+- 只把 Redis 拉起来、不重启应用，合计放行回到 20。`rl:*` 的 PTTL 都大于 0，没有永久计数键。
+- 日志里看不到降级计数。`Degraded()` 没有暴露点。
+- Redis 不通时新起的 team 进程退出，日志含 `redis ... connection refused`。
+- `runtime.redisAddr` 和 `runtime.rateLimitPerMin` 属于默认 owner `local`。`PUT /api/settings/user` 只写调用者自己的设置。演练是在隔离库里用 SQL 写入 `settings(owner='local', scope='user')`，并且保留 `runtime.mock=true`。这不是产品配置入口。
+
 ## 9. 配置系统

 ### 9.1 优先级
@@ -80,5 +80,7 @@ P95≤300ms 阈值约束——「互相不拖累」的可机判口径。SSE 探

 - 正式 60min/2h 窗口、双主机（C3）、`contract` 满量压测窗口需用户指定资源与时间；
  本仓提供可复核入口，不预填容量结论。
- `pgbench` 只测数据库能力，不替代站点业务验收（方案 §5 U7）。
+- `pgbench` 只测数据库能力，不替代站点业务验收（方案 §5 U7）。入口是
+  `run_pgbench.sh`（检查单 B6）：隔离实例 `u7bench`、端口 18995、scale 100，
+  select-only 与 TPC-B 各跑 10/30/100 客户端。没有业务阈值；结果是参考基线。
 - k6 缺失时跳过 k6 段并写 `k6-skipped.txt`；探针与对账仍跑。
@@ -0,0 +1,322 @@
+#!/usr/bin/env bash
+# 检查单 B6：隔离 PostgreSQL 上的 pgbench 参考基线。
+# 只创建并回收名称以 u7bench 开头、端口落在 18980–18999 的实例。
+# 不打生产服务，不把连接串或口令写进输出。
+#
+# 用法：
+#   tests/load/collaboration/run_pgbench.sh [--name u7bench] [--port 18995]
+#       [--scale 100] [--duration 60] [--clients 10,30,100] [--help]
+#   tests/load/collaboration/run_pgbench.sh --self-check
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+NAME=u7bench
+PORT=18995
+SCALE=100
+DURATION=60
+CLIENTS="10,30,100"
+SELF_CHECK=0
+HOLD_AFTER_START=0
+
+usage() {
+  sed -n '2,9p' "$0"
+}
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --name) NAME="$2"; shift 2 ;;
+    --port) PORT="$2"; shift 2 ;;
+    --scale) SCALE="$2"; shift 2 ;;
+    --duration) DURATION="$2"; shift 2 ;;
+    --clients) CLIENTS="$2"; shift 2 ;;
+    --self-check) SELF_CHECK=1; shift ;;
+    --hold-after-start) HOLD_AFTER_START=1; shift ;;
+    -h|--help) usage; exit 0 ;;
+    *) echo "[pgbench] unknown arg: $1" >&2; usage >&2; exit 2 ;;
+  esac
+done
+
+die() { echo "[pgbench] $*" >&2; exit 2; }
+
+case "$NAME" in
+  u7bench|u7bench-*) ;;
+  *) die "--name must start with u7bench (refusing $NAME)" ;;
+esac
+[[ "$NAME" =~ ^[a-z][a-z0-9-]{0,15}$ ]] || die "illegal instance name: $NAME"
+[[ "$PORT" =~ ^[0-9]+$ ]] || die "--port must be an integer"
+if (( PORT < 18980 || PORT > 18999 )); then
+  die "--port must be inside 18980-18999 (refusing $PORT)"
+fi
+[[ "$SCALE" =~ ^[0-9]+$ && "$DURATION" =~ ^[0-9]+$ ]] || die "--scale and --duration must be integers"
+
+port_listening() {
+  local p="$1"
+  if command -v ss >/dev/null 2>&1; then
+    ss -ltn 2>/dev/null | awk -v p=":$p" '$4 ~ p"$" { found=1 } END { exit !found }' && return 0
+  fi
+  (exec 3<>"/dev/tcp/127.0.0.1/$p") 2>/dev/null && return 0
+  return 1
+}
+
+# ---- 自检：正常 / 拒绝 / 中断回收。在创建任何实例之前跑完拒绝项。 ----
+if [[ "$SELF_CHECK" -eq 1 ]]; then
+  echo "[pgbench] self-check: --help"
+  "$0" --help >/dev/null
+  echo "[pgbench] self-check: unknown arg exits 2"
+  set +e
+  "$0" --not-a-flag >/dev/null 2>&1
+  ec=$?
+  set -e
+  [[ "$ec" -eq 2 ]] || die "expected exit 2 for unknown arg, got $ec"
+  echo "[pgbench] self-check: name outside the prefix is refused"
+  set +e
+  "$0" --name prod --port 18995 >/dev/null 2>&1
+  ec=$?
+  set -e
+  [[ "$ec" -eq 2 ]] || die "expected exit 2 for --name prod, got $ec"
+  echo "[pgbench] self-check: busy port is refused before any instance exists"
+  busy=18994
+  python3 -m http.server "$busy" --bind 127.0.0.1 >/dev/null 2>&1 &
+  busy_pid=$!
+  for _ in $(seq 1 20); do port_listening "$busy" && break; sleep 0.1; done
+  set +e
+  "$0" --name u7bench --port "$busy" --duration 5 >/dev/null 2>&1
+  ec=$?
+  set -e
+  kill "$busy_pid" 2>/dev/null || true
+  wait "$busy_pid" 2>/dev/null || true
+  [[ "$ec" -eq 2 ]] || die "expected exit 2 for busy port, got $ec"
+  [[ ! -e "$ROOT/runtime/u7bench" ]] || die "busy-port refusal created runtime/u7bench"
+  echo "[pgbench] self-check: SIGINT removes the instance and the port"
+  # 非交互脚本里后台任务会忽略 SIGINT。打开监视模式后，子进程能接到 SIGINT。
+  set -m
+  "$0" --name u7bench --port 18995 --hold-after-start >/tmp/u7bench-hold.txt 2>&1 &
+  hold_pid=$!
+  set +m
+  ready=0
+  for _ in $(seq 1 90); do
+    if [[ -f "$ROOT/runtime/u7bench/hold.ready" ]]; then
+      ready=1
+      break
+    fi
+    if ! kill -0 "$hold_pid" 2>/dev/null; then
+      break
+    fi
+    sleep 0.5
+  done
+  [[ "$ready" -eq 1 ]] || {
+    echo "[pgbench] hold output:" >&2
+    cat /tmp/u7bench-hold.txt >&2 || true
+    die "hold-after-start did not become ready"
+  }
+  kill -INT -- "-$hold_pid" 2>/dev/null || kill -INT "$hold_pid" 2>/dev/null || true
+  set +e
+  wait "$hold_pid"
+  set -e
+  if port_listening 18995 || [[ -e "$ROOT/runtime/u7bench" ]]; then
+    die "SIGINT left port 18995 or runtime/u7bench behind"
+  fi
+  echo "[pgbench] self-check passed"
+  exit 0
+fi
+
+if port_listening "$PORT"; then
+  die "port $PORT is already in use; refusing before creating $NAME"
+fi
+if [[ -e "$ROOT/runtime/$NAME" ]]; then
+  die "runtime/$NAME already exists; refusing to touch it"
+fi
+
+command -v pgbench >/dev/null 2>&1 || die "pgbench is not installed"
+
+# 计时窗口才受负载门槛约束，并且在创建实例之前判断。
+# 生命周期自检（--hold-after-start）不测量吞吐。
+resource_gate() {
+  local avail_kb load1
+  avail_kb="$(df -Pk "$ROOT" | awk 'NR==2 { print $4 }')"
+  load1="$(awk '{ print $1 }' /proc/loadavg)"
+  if (( avail_kb < 100 * 1024 * 1024 )); then
+    echo "[pgbench] disk free ${avail_kb}KB is below 100GB; pausing" >&2
+    exit 3
+  fi
+  awk -v load1="$load1" 'BEGIN { if (load1+0 > 16) exit 1 }' || {
+    echo "[pgbench] load average $load1 is above 16; pausing" >&2
+    exit 3
+  }
+}
+
+OWNED=0
+cleanup() {
+  local ec=$?
+  trap - EXIT INT TERM
+  if [[ "$OWNED" -eq 1 ]]; then
+    "$ROOT/scripts/svc.sh" "$NAME" stop >/dev/null 2>&1 || true
+    "$ROOT/scripts/svc.sh" rm "$NAME" --purge >/dev/null 2>&1 || true
+  fi
+  exit "$ec"
+}
+trap cleanup EXIT INT TERM
+
+if [[ "$HOLD_AFTER_START" -eq 0 ]]; then
+  resource_gate
+fi
+
+TS="$(date +%Y%m%d-%H%M%S)"
+RESULTS="$HERE/results/pgbench-$TS"
+mkdir -p "$RESULTS"
+echo "[pgbench] results $RESULTS"
+
+{
+  echo "dev_head=$(git -C "$ROOT" rev-parse HEAD)"
+  echo "backend=$(git -C "$ROOT" rev-parse HEAD:modules/backend)"
+  echo "when=$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+  echo "nproc=$(nproc)"
+  echo "loadavg=$(cat /proc/loadavg)"
+  free -h
+  df -h "$ROOT"
+  pgbench --version
+  echo "--- heavy processes (comm only) ---"
+  ps -eo pcpu,pmem,comm --sort=-pcpu | head -12
+} >"$RESULTS/host.txt" 2>&1
+
+echo "[pgbench] creating isolated instance $NAME on 127.0.0.1:$PORT"
+"$ROOT/scripts/svc.sh" db create "$NAME" --port "$PORT" >"$RESULTS/svc-create.txt" 2>&1
+OWNED=1
+addr="$(sed -n 's/^addr=//p' "$ROOT/runtime/$NAME/svc.env" | head -1)"
+[[ "$addr" == "127.0.0.1:$PORT" ]] || {
+  echo "[pgbench] instance addr is $addr, expected 127.0.0.1:$PORT; aborting" >&2
+  exit 1
+}
+"$ROOT/scripts/svc.sh" "$NAME" start >"$RESULTS/svc-start.txt" 2>&1
+
+if [[ "$HOLD_AFTER_START" -eq 1 ]]; then
+  echo "[pgbench] holding after start for signal test"
+  : >"$ROOT/runtime/$NAME/hold.ready"
+  sleep 3600
+fi
+
+DSN="$("$ROOT/scripts/svc.sh" db dsn "$NAME" --reveal)"
+
+psql "$DSN" -v ON_ERROR_STOP=1 -c "SELECT version()" >"$RESULTS/pg-version.txt"
+psql "$DSN" -v ON_ERROR_STOP=1 -c \
+  "SELECT name, setting, unit FROM pg_settings WHERE name IN ('max_connections','superuser_reserved_connections','shared_buffers','work_mem','effective_cache_size','wal_level','listen_addresses')" \
+  >"$RESULTS/pg-settings.txt"
+
+max_conn="$(psql "$DSN" -tAc "SHOW max_connections")"
+reserved="$(psql "$DSN" -tAc "SHOW superuser_reserved_connections")"
+headroom=$(( max_conn - reserved - 2 ))
+(( headroom >= 1 )) || die "no connection headroom (max=$max_conn reserved=$reserved)"
+
+echo "[pgbench] init scale=$SCALE"
+pgbench -i -s "$SCALE" -d "$DSN" >"$RESULTS/init.txt" 2>&1
+psql "$DSN" -v ON_ERROR_STOP=1 -c "VACUUM ANALYZE" >"$RESULTS/vacuum.txt"
+
+runs_tsv="$RESULTS/runs.tsv"
+: >"$runs_tsv"
+IFS=',' read -r -a client_list <<<"$CLIENTS"
+for raw in "${client_list[@]}"; do
+  [[ "$raw" =~ ^[0-9]+$ ]] || die "bad client count: $raw"
+  clients="$raw"
+  capped=0
+  if (( clients > headroom )); then
+    echo "[pgbench] capping clients $clients -> $headroom (max_connections=$max_conn reserved=$reserved)"
+    clients="$headroom"
+    capped=1
+  fi
+  jobs="$clients"
+  ncpu="$(nproc)"
+  if (( jobs > ncpu )); then
+    jobs="$ncpu"
+  fi
+  for mode in select-only tpcb; do
+    echo "[pgbench] mode=$mode clients=$clients jobs=$jobs duration=${DURATION}s"
+    prefix="$RESULTS/${mode}-${clients}"
+    extra=()
+    if [[ "$mode" == "select-only" ]]; then
+      extra+=(-S)
+    fi
+    pgbench -n -c "$clients" -j "$jobs" -T "$DURATION" -M prepared -P 10 \
+      --sampling-rate=0.01 -l --log-prefix="$prefix" \
+      "${extra[@]}" -d "$DSN" | tee "$RESULTS/${mode}-${clients}.txt"
+    failed="$(awk '/number of failed transactions:/ { print $5; exit }' "$RESULTS/${mode}-${clients}.txt")"
+    tps="$(awk '/^tps = / { print $3; exit }' "$RESULTS/${mode}-${clients}.txt")"
+    avg="$(awk '/^latency average = / { print $4; exit }' "$RESULTS/${mode}-${clients}.txt")"
+    [[ -n "$failed" && "$failed" == "0" ]] || {
+      echo "[pgbench] failed transactions = ${failed:-missing} for $mode/$clients" >&2
+      exit 1
+    }
+    stats="$(python3 - "$prefix" <<'PY'
+import glob, os, re, sys
+prefix = sys.argv[1]
+base = os.path.basename(prefix)
+pat = re.compile(re.escape(base) + r"\.\d+(?:\.\d+)?$")
+vals = []
+for path in sorted(glob.glob(prefix + ".*")):
+    if not pat.match(os.path.basename(path)):
+        continue
+    with open(path) as fh:
+        for line in fh:
+            parts = line.split()
+            if len(parts) >= 3 and parts[2].isdigit():
+                vals.append(int(parts[2]))
+if not vals:
+    sys.exit("no latency samples under " + prefix)
+vals.sort()
+def pct(p):
+    idx = int(round((p / 100.0) * (len(vals) - 1)))
+    return f"{vals[idx] / 1000.0:.3f}"
+print(pct(50), pct(95), pct(99), len(vals))
+PY
+)" || exit 1
+    read -r p50 p95 p99 samples <<<"$stats"
+    awk -v p50="$p50" 'BEGIN { if (p50+0 > 60000) exit 1 }' || {
+      echo "[pgbench] latency sample looks unparsed (p50=${p50}ms)" >&2
+      exit 1
+    }
+    printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
+      "$mode" "$clients" "$jobs" "$tps" "$failed" "$avg" "$p50" "$p95" "$p99" "$samples" \
+      >>"$runs_tsv"
+    echo "[pgbench] $mode clients=$clients tps=$tps avg=${avg}ms p50=${p50}ms p95=${p95}ms p99=${p99}ms samples=$samples capped=$capped"
+    load1="$(awk '{ print $1 }' /proc/loadavg)"
+    echo "after ${mode}-${clients} loadavg=$load1" >>"$RESULTS/host.txt"
+  done
+done
+
+python3 - "$RESULTS" "$runs_tsv" "$max_conn" "$reserved" "$headroom" "$SCALE" "$DURATION" <<'PY'
+import json, pathlib, sys
+results, tsv, max_conn, reserved, headroom, scale, duration = sys.argv[1:]
+host = pathlib.Path(results, "host.txt").read_text()
+def grab(key):
+    for line in host.splitlines():
+        if line.startswith(key + "="):
+            return line.split("=", 1)[1]
+    return ""
+runs = []
+for line in pathlib.Path(tsv).read_text().splitlines():
+    mode, clients, jobs, tps, failed, avg, p50, p95, p99, samples = line.split("\t")
+    runs.append({
+        "mode": mode, "clients": int(clients), "jobs": int(jobs),
+        "tps": float(tps), "failed": int(failed), "latency_avg_ms": float(avg),
+        "p50_ms": float(p50), "p95_ms": float(p95), "p99_ms": float(p99),
+        "samples": int(samples),
+    })
+doc = {
+    "source": "SELF",
+    "kind": "pgbench-reference",
+    "note": "Database capacity only. Not a business acceptance threshold.",
+    "dev_head": grab("dev_head"),
+    "backend": grab("backend"),
+    "scale": int(scale),
+    "duration_s": int(duration),
+    "max_connections": int(max_conn),
+    "superuser_reserved_connections": int(reserved),
+    "client_headroom": int(headroom),
+    "runs": runs,
+}
+path = pathlib.Path(results, "pgbench-summary.json")
+path.write_text(json.dumps(doc, indent=2) + "\n")
+print(json.dumps(doc, indent=2))
+PY
+
+echo "[pgbench] done. summary $RESULTS/pgbench-summary.json"
@@ -0,0 +1,2 @@
+/.build/
+/results/
@@ -0,0 +1,20 @@
+# U7 运维演练（C4 / C5）
+
+只打隔离栈。实例名必须以 `u7drill` 或 `u7bk` 开头，端口必须落在 18980–18999。
+脚本只按自己的 pidfile 结束进程，并且先核对命令行里有本演练的数据目录或端口。
+
+| 文件 | 职责 |
+|---|---|
+| `lib.sh` | 隔离 PG / Redis / serve 的创建、回收、端口与名称守卫 |
+| `drill_driver/` | 时间线、幂等重放、SSE seq、标记写入、工件、限流突发、权限核对 |
+| `run_fault_drill.sh` | C4：`pg-stop`、`pg-kill`、`redis-kill` |
+| `run_backup_drill.sh` | C5：物理备份 + WAL/PITR，以及 pg_dump，恢复到新目标 |
+
+```bash
+tests/ops/self_check.sh
+tests/ops/run_fault_drill.sh --fault all
+tests/ops/run_backup_drill.sh
+```
+
+结果在 `results/<ts>/`（gitignore）。证据以 `make evidence` 记入任务卡，来源是 SELF。
+本入口不放行第二实例写流量。
@@ -0,0 +1,3 @@
+module drilldriver
+
+go 1.22
@@ -0,0 +1,26 @@
+package main
+
+// seqHoles 返回已观察到的 seq 里缺失的编号。空切片没有洞。
+func seqHoles(seqs []int64) []int64 {
+	if len(seqs) == 0 {
+		return nil
+	}
+	min, max := seqs[0], seqs[0]
+	seen := map[int64]bool{}
+	for _, s := range seqs {
+		seen[s] = true
+		if s < min {
+			min = s
+		}
+		if s > max {
+			max = s
+		}
+	}
+	var holes []int64
+	for s := min; s <= max; s++ {
+		if !seen[s] {
+			holes = append(holes, s)
+		}
+	}
+	return holes
+}
@@ -0,0 +1,19 @@
+package main
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestSeqHoles(t *testing.T) {
+	if got := seqHoles(nil); got != nil {
+		t.Fatalf("empty: %#v", got)
+	}
+	if got := seqHoles([]int64{1, 2, 3}); got != nil {
+		t.Fatalf("contiguous: %#v", got)
+	}
+	got := seqHoles([]int64{1, 3, 4})
+	if !reflect.DeepEqual(got, []int64{2}) {
+		t.Fatalf("hole: %#v", got)
+	}
+}
@@ -0,0 +1,644 @@
+// 演练驱动：对隔离实例发真实 HTTP，记录每次请求的结果。
+// 连接失败和超时记为 status=0，不把它们当成业务成功。
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"net/http/cookiejar"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		usage()
+		os.Exit(2)
+	}
+	var err error
+	switch os.Args[1] {
+	case "prepare":
+		err = cmdPrepare(os.Args[2:])
+	case "timeline":
+		err = cmdTimeline(os.Args[2:])
+	case "idem-verify":
+		err = cmdIdem(os.Args[2:])
+	case "sse-check":
+		err = cmdSSE(os.Args[2:])
+	case "markers":
+		err = cmdMarkers(os.Args[2:])
+	case "artifacts-put":
+		err = cmdArtifactsPut(os.Args[2:])
+	case "artifacts-verify":
+		err = cmdArtifactsVerify(os.Args[2:])
+	case "burst":
+		err = cmdBurst(os.Args[2:])
+	case "perm-check":
+		err = cmdPerm(os.Args[2:])
+	case "writes":
+		err = cmdWrites(os.Args[2:])
+	case "-h", "--help", "help":
+		usage()
+		return
+	default:
+		usage()
+		os.Exit(2)
+	}
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func usage() {
+	fmt.Fprintln(os.Stderr, "drill_driver prepare|timeline|idem-verify|sse-check|markers|artifacts-put|artifacts-verify|burst|perm-check|writes")
+}
+
+type session struct {
+	base   string
+	client *http.Client
+}
+
+func newSession(base, owner, password string) (*session, error) {
+	jar, err := cookiejar.New(nil)
+	if err != nil {
+		return nil, err
+	}
+	s := &session{
+		base:   strings.TrimRight(base, "/"),
+		client: &http.Client{Timeout: 15 * time.Second, Jar: jar},
+	}
+	_, err = s.do("POST", "/api/auth/login", map[string]string{"owner": owner, "password": password})
+	if err != nil {
+		return nil, fmt.Errorf("login %s: %w", base, err)
+	}
+	return s, nil
+}
+
+type callResult struct {
+	Status    int             `json:"status"`
+	Elapsed   time.Duration   `json:"-"`
+	ElapsedMs float64         `json:"elapsedMs"`
+	Body      json.RawMessage `json:"body,omitempty"`
+	Error     string          `json:"error,omitempty"`
+}
+
+func (s *session) do(method, path string, body any) (callResult, error) {
+	var rd io.Reader
+	if body != nil {
+		buf, err := json.Marshal(body)
+		if err != nil {
+			return callResult{}, err
+		}
+		rd = bytes.NewReader(buf)
+	}
+	req, err := http.NewRequest(method, s.base+path, rd)
+	if err != nil {
+		return callResult{}, err
+	}
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	start := time.Now()
+	resp, err := s.client.Do(req)
+	elapsed := time.Since(start)
+	if err != nil {
+		return callResult{Elapsed: elapsed, ElapsedMs: ms(elapsed), Error: err.Error()}, nil
+	}
+	defer resp.Body.Close()
+	raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	out := callResult{Status: resp.StatusCode, Elapsed: elapsed, ElapsedMs: ms(elapsed), Body: raw}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return out, fmt.Errorf("%s %s: status %d", method, path, resp.StatusCode)
+	}
+	return out, nil
+}
+
+func ms(d time.Duration) float64 {
+	return float64(d) / float64(time.Millisecond)
+}
+
+func writeJSON(path string, v any) error {
+	data, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return err
+	}
+	data = append(data, '\n')
+	if path == "" || path == "-" {
+		_, err = os.Stdout.Write(data)
+		return err
+	}
+	return os.WriteFile(path, data, 0o644)
+}
+
+func cmdPrepare(args []string) error {
+	fs := flag.NewFlagSet("prepare", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	out := fs.String("out", "-", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	proj, err := s.do("POST", "/api/projects", map[string]any{"name": "drill"})
+	if err != nil {
+		return err
+	}
+	var projBody struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(proj.Body, &projBody); err != nil {
+		return err
+	}
+	conv, err := s.do("POST", "/api/projects/"+projBody.ID+"/conversations", map[string]any{"title": "drill"})
+	if err != nil {
+		return err
+	}
+	var convBody struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(conv.Body, &convBody); err != nil {
+		return err
+	}
+	return writeJSON(*out, map[string]string{"projectId": projBody.ID, "conversationId": convBody.ID})
+}
+
+type attempt struct {
+	RequestID string  `json:"requestId"`
+	Base      string  `json:"base"`
+	Started   string  `json:"started"`
+	ElapsedMs float64 `json:"elapsedMs"`
+	Status    int     `json:"status"`
+	Error     string  `json:"error,omitempty"`
+	MessageID string  `json:"messageId,omitempty"`
+}
+
+func cmdTimeline(args []string) error {
+	fs := flag.NewFlagSet("timeline", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	conv := fs.String("conv", "", "conversation id")
+	every := fs.Duration("every", 500*time.Millisecond, "write interval")
+	dur := fs.Duration("dur", 90*time.Second, "how long to write")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	var attempts []attempt
+	deadline := time.Now().Add(*dur)
+	for i := 0; time.Now().Before(deadline); i++ {
+		reqID := fmt.Sprintf("tl-%d-%d", time.Now().UnixNano(), i)
+		started := time.Now().UTC()
+		res, _ := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{
+			"text": "drill " + reqID, "requestId": reqID,
+		})
+		attempts = append(attempts, attempt{
+			RequestID: reqID, Base: *base, Started: started.Format(time.RFC3339Nano),
+			ElapsedMs: res.ElapsedMs, Status: res.Status, Error: res.Error, MessageID: messageID(res.Body),
+		})
+		time.Sleep(*every)
+	}
+	return writeJSON(*out, map[string]any{"attempts": attempts})
+}
+
+func messageID(raw json.RawMessage) string {
+	var body struct {
+		Message struct {
+			ID string `json:"id"`
+		} `json:"message"`
+	}
+	_ = json.Unmarshal(raw, &body)
+	return body.Message.ID
+}
+
+func cmdIdem(args []string) error {
+	fs := flag.NewFlagSet("idem-verify", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	conv := fs.String("conv", "", "conversation id")
+	in := fs.String("in", "", "timeline json")
+	out := fs.String("out", "", "json output")
+	since := fs.String("since", "", "only replay attempts at or after this time")
+	until := fs.String("until", "", "only replay attempts at or before this time")
+	retry := fs.Duration("retry", 2*time.Minute, "retry 409 until the conversation is free")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	data, err := os.ReadFile(*in)
+	if err != nil {
+		return err
+	}
+	var doc struct {
+		Attempts []attempt `json:"attempts"`
+	}
+	if err := json.Unmarshal(data, &doc); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	var pending []attempt
+	for _, a := range doc.Attempts {
+		if a.Status >= 200 && a.Status < 300 {
+			continue
+		}
+		if !attemptInWindow(a.Started, *since, *until) {
+			continue
+		}
+		pending = append(pending, a)
+	}
+	var replays []attempt
+	deadline := time.Now().Add(*retry)
+	for len(pending) > 0 {
+		var again []attempt
+		for _, a := range pending {
+			res, _ := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{
+				"text": "drill " + a.RequestID, "requestId": a.RequestID,
+			})
+			// 409 发生在写入之前：会话仍被 mock run 占用，requestId 还没落库。
+			if res.Status == http.StatusConflict && time.Now().Before(deadline) {
+				again = append(again, a)
+				continue
+			}
+			replays = append(replays, attempt{
+				RequestID: a.RequestID, Base: *base, Started: time.Now().UTC().Format(time.RFC3339Nano),
+				ElapsedMs: res.ElapsedMs, Status: res.Status, Error: res.Error, MessageID: messageID(res.Body),
+			})
+		}
+		pending = again
+		if len(pending) > 0 {
+			time.Sleep(time.Second)
+		}
+	}
+	return writeJSON(*out, map[string]any{"replays": replays})
+}
+
+func parseStamp(s string) (time.Time, error) {
+	if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
+		return t, nil
+	}
+	return time.Parse(time.RFC3339, s)
+}
+
+func attemptInWindow(started, since, until string) bool {
+	t, err := parseStamp(started)
+	if err != nil {
+		return false
+	}
+	if since != "" {
+		s, err := parseStamp(since)
+		if err != nil || t.Before(s) {
+			return false
+		}
+	}
+	if until != "" {
+		u, err := parseStamp(until)
+		if err != nil || t.After(u) {
+			return false
+		}
+	}
+	return true
+}
+
+func cmdSSE(args []string) error {
+	fs := flag.NewFlagSet("sse-check", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	conv := fs.String("conv", "", "conversation id")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.base+"/api/events?conversationId="+*conv, nil)
+	if err != nil {
+		return err
+	}
+	stream := &http.Client{Jar: s.client.Jar}
+	resp, err := stream.Do(req)
+	var seqs []int64
+	status := 0
+	errText := ""
+	if err != nil && resp == nil {
+		errText = err.Error()
+	} else if resp != nil {
+		status = resp.StatusCode
+		defer resp.Body.Close()
+		body, readErr := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+		if readErr != nil {
+			errText = readErr.Error()
+		}
+		for _, line := range strings.Split(string(body), "\n") {
+			if !strings.HasPrefix(line, "id: ") {
+				continue
+			}
+			n, convErr := strconv.ParseInt(strings.TrimPrefix(line, "id: "), 10, 64)
+			if convErr == nil {
+				seqs = append(seqs, n)
+			}
+		}
+	}
+	return writeJSON(*out, map[string]any{
+		"status": status, "error": errText, "seqs": seqs, "holes": seqHoles(seqs),
+	})
+}
+
+func cmdMarkers(args []string) error {
+	fs := flag.NewFlagSet("markers", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	conv := fs.String("conv", "", "conversation id")
+	every := fs.Duration("every", 5*time.Second, "marker interval")
+	dur := fs.Duration("dur", 5*time.Minute, "how long to write")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	var marks []map[string]any
+	deadline := time.Now().Add(*dur)
+	for i := 0; time.Now().Before(deadline); i++ {
+		reqID := fmt.Sprintf("mk-%d", i)
+		acked := time.Now().UTC()
+		res, _ := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{
+			"text": "marker " + reqID, "requestId": reqID,
+		})
+		marks = append(marks, map[string]any{
+			"requestId": reqID, "acked": acked.Format(time.RFC3339Nano),
+			"status": res.Status, "error": res.Error, "messageId": messageID(res.Body),
+		})
+		time.Sleep(*every)
+	}
+	return writeJSON(*out, map[string]any{"markers": marks})
+}
+
+func cmdArtifactsPut(args []string) error {
+	fs := flag.NewFlagSet("artifacts-put", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	project := fs.String("project", "", "project id")
+	n := fs.Int("n", 5, "how many files")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	var items []map[string]any
+	for i := 0; i < *n; i++ {
+		payload := []byte(fmt.Sprintf("drill-artifact-%d-%d", i, time.Now().UnixNano()))
+		sum := sha256.Sum256(payload)
+		id, status, errText, acked := upload(s, *project, fmt.Sprintf("drill-%d.txt", i), payload)
+		items = append(items, map[string]any{
+			"name": fmt.Sprintf("drill-%d.txt", i), "sha256": hex.EncodeToString(sum[:]),
+			"id": id, "status": status, "error": errText, "acked": acked.Format(time.RFC3339Nano),
+		})
+	}
+	return writeJSON(*out, map[string]any{"artifacts": items})
+}
+
+func upload(s *session, project, name string, payload []byte) (id string, status int, errText string, acked time.Time) {
+	var buf bytes.Buffer
+	w := multipart.NewWriter(&buf)
+	part, err := w.CreateFormFile("file", name)
+	if err != nil {
+		return "", 0, err.Error(), time.Now().UTC()
+	}
+	_, _ = part.Write(payload)
+	_ = w.Close()
+	req, err := http.NewRequest(http.MethodPost, s.base+"/api/projects/"+project+"/uploads", &buf)
+	if err != nil {
+		return "", 0, err.Error(), time.Now().UTC()
+	}
+	req.Header.Set("Content-Type", w.FormDataContentType())
+	acked = time.Now().UTC()
+	resp, err := s.client.Do(req)
+	if err != nil {
+		return "", 0, err.Error(), acked
+	}
+	defer resp.Body.Close()
+	raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return artifactIDFromUpload(raw), resp.StatusCode, string(raw), acked
+	}
+	return artifactIDFromUpload(raw), resp.StatusCode, "", acked
+}
+
+func artifactIDFromUpload(raw []byte) string {
+	var body struct {
+		ID         string `json:"id"`
+		ArtifactID string `json:"artifactId"`
+	}
+	_ = json.Unmarshal(raw, &body)
+	if body.ID != "" {
+		return body.ID
+	}
+	return body.ArtifactID
+}
+
+func cmdArtifactsVerify(args []string) error {
+	fs := flag.NewFlagSet("artifacts-verify", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	in := fs.String("in", "", "artifacts json")
+	before := fs.String("before", "", "only artifacts acked before this RFC3339 time")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	data, err := os.ReadFile(*in)
+	if err != nil {
+		return err
+	}
+	var doc struct {
+		Artifacts []struct {
+			ID     string `json:"id"`
+			SHA256 string `json:"sha256"`
+			Acked  string `json:"acked"`
+			Status int    `json:"status"`
+		} `json:"artifacts"`
+	}
+	if err := json.Unmarshal(data, &doc); err != nil {
+		return err
+	}
+	cutoff, err := time.Parse(time.RFC3339Nano, *before)
+	if err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	missing := 0
+	mismatch := 0
+	checked := 0
+	for _, item := range doc.Artifacts {
+		acked, err := time.Parse(time.RFC3339Nano, item.Acked)
+		if err != nil || item.Status < 200 || item.Status >= 300 || !acked.Before(cutoff) {
+			continue
+		}
+		checked++
+		req, err := http.NewRequest(http.MethodGet, s.base+"/api/artifacts/"+item.ID+"/content", nil)
+		if err != nil {
+			missing++
+			continue
+		}
+		resp, err := s.client.Do(req)
+		if err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300 {
+			if resp != nil {
+				resp.Body.Close()
+			}
+			missing++
+			continue
+		}
+		raw, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		sum := sha256.Sum256(raw)
+		if hex.EncodeToString(sum[:]) != item.SHA256 {
+			mismatch++
+		}
+	}
+	return writeJSON(*out, map[string]any{"checked": checked, "missing": missing, "mismatch": mismatch})
+}
+
+func cmdBurst(args []string) error {
+	fs := flag.NewFlagSet("burst", flag.ContinueOnError)
+	bases := fs.String("bases", "", "comma-separated base URLs")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	n := fs.Int("n", 40, "requests per base")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	counts := map[string]int{}
+	for _, base := range strings.Split(*bases, ",") {
+		base = strings.TrimSpace(base)
+		if base == "" {
+			continue
+		}
+		s, err := newSession(base, *owner, *password)
+		if err != nil {
+			return err
+		}
+		for i := 0; i < *n; i++ {
+			res, _ := s.do("GET", "/api/projects", nil)
+			key := fmt.Sprintf("%s:%d", base, res.Status)
+			if res.Status == 0 {
+				key = base + ":error"
+			}
+			counts[key]++
+		}
+	}
+	return writeJSON(*out, map[string]any{"counts": counts})
+}
+
+func cmdWrites(args []string) error {
+	fs := flag.NewFlagSet("writes", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "login owner")
+	password := fs.String("password", "", "login password")
+	conv := fs.String("conv", "", "conversation id")
+	project := fs.String("project", "", "project id used when -fresh creates a conversation per write")
+	fresh := fs.Bool("fresh", false, "one new conversation per write, so a live run cannot 409 the probe")
+	n := fs.Int("n", 5, "consecutive successful writes required")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if *fresh && *project == "" {
+		return fmt.Errorf("writes -fresh requires -project")
+	}
+	if !*fresh && *conv == "" {
+		return fmt.Errorf("writes requires -conv")
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	for i := 0; i < *n; i++ {
+		target := *conv
+		if *fresh {
+			created, err := s.do("POST", "/api/projects/"+*project+"/conversations", map[string]any{
+				"title": fmt.Sprintf("recover-%d", i),
+			})
+			if err != nil {
+				return fmt.Errorf("write %d conversation: %w", i, err)
+			}
+			var body struct {
+				ID string `json:"id"`
+			}
+			if err := json.Unmarshal(created.Body, &body); err != nil || body.ID == "" {
+				return fmt.Errorf("write %d conversation id missing", i)
+			}
+			target = body.ID
+		}
+		reqID := fmt.Sprintf("wr-%d-%d", time.Now().UnixNano(), i)
+		res, err := s.do("POST", "/api/conversations/"+target+"/messages", map[string]any{
+			"text": "recover " + reqID, "requestId": reqID,
+		})
+		if err != nil || res.Status < 200 || res.Status >= 300 {
+			return fmt.Errorf("write %d status %d: %s", i, res.Status, res.Error)
+		}
+	}
+	fmt.Printf("writes ok n=%d\n", *n)
+	return nil
+}
+
+func cmdPerm(args []string) error {
+	fs := flag.NewFlagSet("perm-check", flag.ContinueOnError)
+	base := fs.String("base", "", "instance base URL")
+	owner := fs.String("owner", "", "other user")
+	password := fs.String("password", "", "other user password")
+	project := fs.String("project", "", "someone else's project")
+	artifact := fs.String("artifact", "", "someone else's artifact")
+	out := fs.String("out", "", "json output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	proj, _ := s.do("GET", "/api/projects/"+*project+"/conversations", nil)
+	art, _ := s.do("GET", "/api/artifacts/"+*artifact+"/content", nil)
+	denied := func(status int) bool { return status == http.StatusForbidden || status == http.StatusNotFound }
+	return writeJSON(*out, map[string]any{
+		"projectStatus": proj.Status, "artifactStatus": art.Status,
+		"projectDenied": denied(proj.Status), "artifactDenied": denied(art.Status),
+	})
+}
@@ -0,0 +1,30 @@
+package main
+
+import "testing"
+
+func TestArtifactIDFromUpload(t *testing.T) {
+	if got := artifactIDFromUpload([]byte(`{"artifactId":"art_1","filename":"a.txt"}`)); got != "art_1" {
+		t.Fatalf("artifactId = %q", got)
+	}
+	if got := artifactIDFromUpload([]byte(`{"id":"art_2"}`)); got != "art_2" {
+		t.Fatalf("id = %q", got)
+	}
+}
+
+func TestAttemptInWindow(t *testing.T) {
+	cases := []struct {
+		started, since, until string
+		want                  bool
+	}{
+		{"2026-10-12T13:37:55.091216802Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false},
+		{"2026-10-12T13:37:57Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", true},
+		{"2026-10-12T13:38:57.000000001Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false},
+		{"2026-10-12T13:38:10.5Z", "", "", true},
+		{"not-a-time", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false},
+	}
+	for _, tc := range cases {
+		if got := attemptInWindow(tc.started, tc.since, tc.until); got != tc.want {
+			t.Errorf("attemptInWindow(%s, %s, %s) = %v, want %v", tc.started, tc.since, tc.until, got, tc.want)
+		}
+	}
+}
@@ -0,0 +1,148 @@
+#!/usr/bin/env bash
+# 演练栈公共函数。由 run_fault_drill.sh / run_backup_drill.sh / self_check.sh source。
+# 不打生产：名称前缀 u7drill 或 u7bk，端口 18980–18999，只杀命令行匹配的 pid。
+
+ops_port_listening() {
+  local p="$1"
+  if command -v ss >/dev/null 2>&1; then
+    ss -ltn 2>/dev/null | awk -v p=":$p" '$4 ~ p"$" { found=1 } END { exit !found }' && return 0
+  fi
+  (exec 3<>"/dev/tcp/127.0.0.1/$p") 2>/dev/null && return 0
+  return 1
+}
+
+ops_require_name() {
+  local name="$1"
+  case "$name" in
+    u7drill|u7drill-*|u7bk|u7bk-*) ;;
+    *) echo "[ops] refusing instance name $name" >&2; return 2 ;;
+  esac
+  [[ "$name" =~ ^[a-z][a-z0-9-]{0,15}$ ]] || {
+    echo "[ops] illegal instance name $name" >&2
+    return 2
+  }
+}
+
+ops_require_port() {
+  local port="$1"
+  [[ "$port" =~ ^[0-9]+$ ]] || { echo "[ops] port must be an integer" >&2; return 2; }
+  if (( port < 18980 || port > 18999 )); then
+    echo "[ops] refusing port $port outside 18980-18999" >&2
+    return 2
+  fi
+}
+
+ops_resource_gate() {
+  local root="$1" avail_kb load1
+  avail_kb="$(df -Pk "$root" | awk 'NR==2 { print $4 }')"
+  load1="$(awk '{ print $1 }' /proc/loadavg)"
+  if (( avail_kb < 100 * 1024 * 1024 )); then
+    echo "[ops] disk free ${avail_kb}KB is below 100GB; pausing" >&2
+    return 3
+  fi
+  awk -v load1="$load1" 'BEGIN { if (load1+0 > 16) exit 1 }' || {
+    echo "[ops] load average $load1 is above 16; pausing" >&2
+    return 3
+  }
+}
+
+# ops_kill_checked PIDFILE NEEDLE：只在 /proc/pid/cmdline 含 NEEDLE 时发信号。
+ops_kill_checked() {
+  local pidfile="$1" needle="$2" sig="${3:-TERM}"
+  [[ -f "$pidfile" ]] || return 0
+  local pid cmd
+  pid="$(head -1 "$pidfile" 2>/dev/null || true)"
+  [[ "$pid" =~ ^[0-9]+$ ]] || return 0
+  if [[ ! -d "/proc/$pid" ]]; then
+    rm -f "$pidfile"
+    return 0
+  fi
+  cmd="$(tr '\0' ' ' <"/proc/$pid/cmdline" 2>/dev/null || true)"
+  if [[ "$cmd" != *"$needle"* ]]; then
+    echo "[ops] refusing to signal $pid; cmdline does not contain $needle" >&2
+    return 1
+  fi
+  kill "-$sig" "$pid" 2>/dev/null || true
+}
+
+ops_pg_create() {
+  local root="$1" name="$2" port="$3"
+  ops_require_name "$name" || return $?
+  ops_require_port "$port" || return $?
+  if ops_port_listening "$port"; then
+    echo "[ops] port $port is in use; refusing before create" >&2
+    return 2
+  fi
+  if [[ -e "$root/runtime/$name" ]]; then
+    echo "[ops] runtime/$name already exists; refusing to touch it" >&2
+    return 2
+  fi
+  "$root/scripts/svc.sh" db create "$name" --port "$port" >/dev/null
+  local addr
+  addr="$(sed -n 's/^addr=//p' "$root/runtime/$name/svc.env" | head -1)"
+  if [[ "$addr" != "127.0.0.1:$port" ]]; then
+    echo "[ops] $name addr is $addr, expected 127.0.0.1:$port" >&2
+    return 1
+  fi
+}
+
+ops_pg_start() {
+  local root="$1" name="$2"
+  "$root/scripts/svc.sh" "$name" start >/dev/null
+}
+
+ops_pg_stop() {
+  local root="$1" name="$2"
+  "$root/scripts/svc.sh" "$name" stop >/dev/null
+}
+
+ops_pg_kill() {
+  local root="$1" name="$2"
+  local pgdata="$root/runtime/$name/pgdata"
+  local pidfile="$pgdata/postmaster.pid"
+  [[ -f "$pidfile" ]] || { echo "[ops] no postmaster.pid for $name" >&2; return 1; }
+  ops_kill_checked "$pidfile" "$pgdata" KILL
+}
+
+ops_pg_dsn() {
+  local root="$1" name="$2"
+  "$root/scripts/svc.sh" db dsn "$name" --reveal
+}
+
+ops_pg_remove() {
+  local root="$1" name="$2"
+  "$root/scripts/svc.sh" "$name" stop >/dev/null 2>&1 || true
+  "$root/scripts/svc.sh" rm "$name" --purge >/dev/null 2>&1 || true
+}
+
+ops_redis_start() {
+  local port="$1" dir="$2"
+  ops_require_port "$port" || return $?
+  if ops_port_listening "$port"; then
+    echo "[ops] redis port $port is in use" >&2
+    return 2
+  fi
+  mkdir -p "$dir"
+  redis-server --bind 127.0.0.1 --port "$port" --dir "$dir" --dbfilename drill.rdb \
+    --save '' --appendonly no --daemonize yes --pidfile "$dir/redis.pid" --logfile "$dir/redis.log"
+}
+
+ops_redis_kill() {
+  local dir="$1" port="$2"
+  ops_kill_checked "$dir/redis.pid" ":$port" KILL
+}
+
+ops_snapshot() {
+  local root="$1" dest="$2"
+  {
+    echo "dev_head=$(git -C "$root" rev-parse HEAD)"
+    echo "backend=$(git -C "$root" rev-parse HEAD:modules/backend)"
+    echo "when=$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+    echo "nproc=$(nproc)"
+    echo "loadavg=$(cat /proc/loadavg)"
+    free -h
+    df -h "$root"
+    echo "--- heavy processes (comm only) ---"
+    ps -eo pcpu,pmem,comm --sort=-pcpu | head -12
+  } >"$dest"
+}
@@ -0,0 +1,449 @@
+#!/usr/bin/env bash
+# C5：物理备份 + WAL 归档 / PITR 到新数据目录，以及 pg_dump。只打 u7bk，端口 18994/18997/18998/18999。
+# 判据失败记在结果里。产品代码不在这里改。
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+# shellcheck source=lib.sh
+source "$ROOT/tests/ops/lib.sh"
+
+NAME=u7bk
+PG_PORT=18994
+APP_PORT=18999
+RESTORE_PG_PORT=18997
+RESTORE_APP_PORT=18998
+ADMIN_OWNER=drill_admin
+ADMIN_PASS=drill-admin-pass-Aa1
+USER_OWNER=drill_user
+USER_PASS=drill-user-pass-Aa9
+
+usage() {
+  sed -n '2,4p' "$0"
+  echo "  --self-check"
+}
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --self-check) "$ROOT/tests/ops/self_check.sh"; exit $? ;;
+    -h|--help) usage; exit 0 ;;
+    *) echo "[backup] unknown arg: $1" >&2; usage >&2; exit 2 ;;
+  esac
+done
+
+ops_require_name "$NAME"
+ops_require_port "$PG_PORT"
+ops_require_port "$APP_PORT"
+ops_require_port "$RESTORE_PG_PORT"
+ops_require_port "$RESTORE_APP_PORT"
+ops_resource_gate "$ROOT" || exit $?
+
+PG_CTL="$(command -v pg_ctl)"
+[[ -n "$PG_CTL" ]] || { echo "[backup] pg_ctl not found" >&2; exit 1; }
+
+OWNED=0
+APP_PIDFILE=""
+RESTORE_PG=""
+ASIDE=""
+MARKER_PID=""
+RSYNC_PID=""
+cleanup() {
+  local ec=$?
+  trap - EXIT INT TERM
+  if [[ -n "$MARKER_PID" ]]; then kill "$MARKER_PID" 2>/dev/null || true; fi
+  if [[ -n "$RSYNC_PID" ]]; then kill "$RSYNC_PID" 2>/dev/null || true; fi
+  if [[ -n "$APP_PIDFILE" ]]; then ops_kill_checked "$APP_PIDFILE" "127.0.0.1:$APP_PORT" TERM || true; fi
+  if [[ -n "${RESTORE_APP_PIDFILE:-}" ]]; then ops_kill_checked "$RESTORE_APP_PIDFILE" "127.0.0.1:$RESTORE_APP_PORT" TERM || true; fi
+  if [[ -n "$RESTORE_PG" && -d "$RESTORE_PG" ]]; then
+    "$PG_CTL" -D "$RESTORE_PG" stop -m fast >/dev/null 2>&1 || true
+    rm -rf "$RESTORE_PG"
+  fi
+  rm -rf /tmp/u7bk-18997
+  if [[ -n "$ASIDE" && -d "$ASIDE" ]]; then rm -rf "$ASIDE"; fi
+  if [[ "$OWNED" -eq 1 ]]; then ops_pg_remove "$ROOT" "$NAME" || true; fi
+  exit "$ec"
+}
+trap cleanup EXIT INT TERM
+
+TS="$(date +%Y%m%d-%H%M%S)"
+RESULTS="$ROOT/tests/ops/results/backup-$TS"
+WAL="$RESULTS/backup/wal"
+BASE="$RESULTS/backup/base"
+SNAP="$RESULTS/backup/artifacts"
+ARTS="$RESULTS/artifacts"
+mkdir -p "$WAL" "$SNAP" "$ARTS"
+chmod 700 "$RESULTS/backup" "$WAL" "$SNAP"
+ops_snapshot "$ROOT" "$RESULTS/host.txt"
+echo "[backup] results $RESULTS"
+
+BIN="$ROOT/tests/ops/.build/agent-system"
+DRIVER="$ROOT/tests/ops/.build/drill_driver"
+SEED="$ROOT/tests/ops/.build/seed"
+mkdir -p "$ROOT/tests/ops/.build"
+( cd "$ROOT/modules/backend" && CGO_ENABLED=0 go build -o "$BIN" ./cmd/agent-system )
+( cd "$ROOT/tests/ops/drill_driver" && go build -o "$DRIVER" . )
+( cd "$ROOT/tests/load/collaboration/seed" && go build -o "$SEED" . )
+
+echo "[backup] creating $NAME on 127.0.0.1:$PG_PORT"
+ops_pg_create "$ROOT" "$NAME" "$PG_PORT"
+OWNED=1
+ops_pg_start "$ROOT" "$NAME"
+SRC_DSN="$(ops_pg_dsn "$ROOT" "$NAME")"
+export SRC_DSN
+PGUSER="$(python3 -c 'import os; from urllib.parse import urlsplit; print(urlsplit(os.environ["SRC_DSN"]).username)' )"
+export PGUSER
+export PGPASSWORD="$(python3 -c 'import os; from urllib.parse import urlsplit, unquote; print(unquote(urlsplit(os.environ["SRC_DSN"]).password or ""))')"
+export PGHOST=127.0.0.1
+export PGPORT="$PG_PORT"
+
+psql -d agent -v ON_ERROR_STOP=1 \
+  -c "ALTER SYSTEM SET wal_level = 'replica'" \
+  -c "ALTER SYSTEM SET archive_mode = 'on'" \
+  -c "ALTER SYSTEM SET archive_timeout = '60s'" \
+  -c "ALTER SYSTEM SET archive_command = 'test ! -f ${WAL}/%f && cp %p ${WAL}/%f && chmod 600 ${WAL}/%f'" \
+  >/dev/null
+ops_pg_stop "$ROOT" "$NAME"
+ops_pg_start "$ROOT" "$NAME"
+wal_level="$(psql -d agent -tAc 'SHOW wal_level')"
+archive_mode="$(psql -d agent -tAc 'SHOW archive_mode')"
+echo "[backup] wal_level=$wal_level archive_mode=$archive_mode"
+[[ "$wal_level" == "replica" && "$archive_mode" == "on" ]]
+
+psql -d agent -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS u7bk_app WITH (FORCE)" -c "CREATE DATABASE u7bk_app" >/dev/null
+APP_DSN="$(python3 -c 'import os; from urllib.parse import urlsplit, urlunsplit; u=urlsplit(os.environ["SRC_DSN"]); print(urlunsplit((u.scheme, u.netloc, "/u7bk_app", u.query, "")))')"
+export APP_DSN RESTORE_PG_PORT
+
+"$BIN" user create --dsn "$APP_DSN" --admin --owner "$ADMIN_OWNER" --password "$ADMIN_PASS" >/dev/null
+"$BIN" user create --dsn "$APP_DSN" --owner "$USER_OWNER" --password "$USER_PASS" >/dev/null
+for i in $(seq 0 9); do
+  "$BIN" user create --dsn "$APP_DSN" --owner "seed_u$i" --password "seed-user-pass-${i}-Aa9" >/dev/null
+done
+echo "[backup] seeding contract"
+"$SEED" --dsn "$APP_DSN" --profile contract --manifest "$RESULTS/seed-manifest.json"
+psql -d u7bk_app -v ON_ERROR_STOP=1 -c "UPDATE users SET must_change_password=FALSE" >/dev/null
+psql -d agent -v ON_ERROR_STOP=1 -c "SELECT pg_switch_wal()" >/dev/null
+archived=0
+for _ in $(seq 1 60); do
+  archived="$(psql -d agent -tAc 'SELECT archived_count FROM pg_stat_archiver' | tr -d ' ')"
+  failed="$(psql -d agent -tAc 'SELECT failed_count FROM pg_stat_archiver' | tr -d ' ')"
+  if [[ "$failed" == "0" && "$archived" != "0" ]]; then break; fi
+  sleep 1
+done
+echo "[backup] archiver archived=$archived failed=$failed"
+
+start_app() {
+  local port="$1" pidfile="$2" dsn="$3" arts="$4"
+  local cwd
+  cwd="$(dirname "$pidfile")/cwd-$(basename "$pidfile" .pid)"
+  mkdir -p "$cwd"
+  (
+    cd "$cwd"
+    exec "$BIN" serve --dsn "$dsn" --addr "127.0.0.1:$port" --tier team --auth on --mock \
+      --instance "u7bk-$port" --artifacts "$arts" --disable-exec-tools >"$(dirname "$pidfile")/app-$port.log" 2>&1
+  ) &
+  echo $! >"$pidfile"
+  local ready=0 code
+  for _ in $(seq 1 90); do
+    code="$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$port/api/health" || true)"
+    if [[ "$code" == "200" || "$code" == "401" ]]; then ready=1; break; fi
+    sleep 0.5
+  done
+  [[ "$ready" -eq 1 ]] || { echo "[backup] app on $port not ready" >&2; return 1; }
+}
+
+APP_PIDFILE="$RESULTS/app-source.pid"
+start_app "$APP_PORT" "$APP_PIDFILE" "$APP_DSN" "$ARTS"
+"$DRIVER" prepare -base "http://127.0.0.1:$APP_PORT" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -out "$RESULTS/prep.json"
+PROJECT="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["projectId"])' "$RESULTS/prep.json")"
+CONV="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["conversationId"])' "$RESULTS/prep.json")"
+"$DRIVER" artifacts-put -base "http://127.0.0.1:$APP_PORT" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+  -project "$PROJECT" -n 5 -out "$RESULTS/artifacts.json"
+
+echo "[backup] base backup"
+base_start="$(date +%s)"
+pg_basebackup -h 127.0.0.1 -p "$PG_PORT" -U "$PGUSER" -D "$BASE" -Fp -X stream -c fast
+set +e
+pg_verifybackup "$BASE" >"$RESULTS/backup/verify.txt"
+verify_ec=$?
+set -e
+echo "[backup] base_s=$(( $(date +%s) - base_start )) verify_ec=$verify_ec"
+echo "[backup] logical dump"
+dump_start="$(date +%s)"
+psql -d u7bk_app -v ON_ERROR_STOP=1 -tAc "SELECT 'messages', count(*) FROM messages UNION ALL SELECT 'events', count(*) FROM events UNION ALL SELECT 'knowledge', count(*) FROM knowledge UNION ALL SELECT 'artifacts', count(*) FROM artifacts UNION ALL SELECT 'runs', count(*) FROM runs" >"$RESULTS/counts-dump.txt"
+"$ROOT/scripts/svc.sh" db backup "$NAME" --db u7bk_app --out "$RESULTS/backup/logical.dump"
+dump_s=$(( $(date +%s) - dump_start ))
+set +e
+pg_restore --list "$RESULTS/backup/logical.dump" >"$RESULTS/backup/dump.list"
+list_ec=$?
+set -e
+echo "[backup] dump_s=$dump_s verify_ec=$verify_ec list_ec=$list_ec"
+find "$BASE" "$WAL" -type d -exec chmod 700 {} +
+find "$BASE" "$WAL" -type f -exec chmod 600 {} +
+chmod 600 "$RESULTS/backup/logical.dump"
+rsync -a "$ARTS/" "$SNAP/"
+
+echo "[backup] markers for 5 minutes"
+"$DRIVER" markers -base "http://127.0.0.1:$APP_PORT" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+  -conv "$CONV" -every 5s -dur 5m -out "$RESULTS/markers.json" >"$RESULTS/markers.stdout" 2>&1 &
+MARKER_PID=$!
+(
+  while kill -0 "$MARKER_PID" 2>/dev/null; do
+    rsync -a "$ARTS/" "$SNAP/"
+    sleep 60
+  done
+) &
+RSYNC_PID=$!
+wait "$MARKER_PID" || true
+MARKER_PID=""
+wait "$RSYNC_PID" || true
+RSYNC_PID=""
+rsync -a "$ARTS/" "$SNAP/"
+
+psql -d u7bk_app -v ON_ERROR_STOP=1 -tAc "SELECT 'messages', count(*) FROM messages UNION ALL SELECT 'events', count(*) FROM events UNION ALL SELECT 'knowledge', count(*) FROM knowledge UNION ALL SELECT 'artifacts', count(*) FROM artifacts UNION ALL SELECT 'runs', count(*) FROM runs" >"$RESULTS/counts-before.txt"
+psql -d agent -v ON_ERROR_STOP=1 -c "SELECT archived_count, failed_count, last_archived_wal, last_failed_wal FROM pg_stat_archiver" >"$RESULTS/archiver.txt"
+FAIL_STAMP="$(date -u +%Y-%m-%dT%H:%M:%S.%NZ)"
+FAIL_EPOCH="$(date +%s)"
+echo "[backup] disaster immediate stop at $FAIL_STAMP"
+PGDATA="$ROOT/runtime/$NAME/pgdata"
+"$PG_CTL" -D "$PGDATA" stop -m immediate
+ASIDE="$RESULTS/source-pgdata-aside"
+mv "$PGDATA" "$ASIDE"
+ops_kill_checked "$APP_PIDFILE" "127.0.0.1:$APP_PORT" TERM || true
+
+echo "[backup] restore to new data directory on $RESTORE_PG_PORT"
+RESTORE_PG="$RESULTS/restore/pgdata"
+mkdir -p "$RESTORE_PG" /tmp/u7bk-18997
+chmod 700 /tmp/u7bk-18997
+cp -a "$BASE/." "$RESTORE_PG/"
+cat >>"$RESTORE_PG/postgresql.auto.conf" <<EOF
+archive_mode = 'off'
+restore_command = 'cp ${WAL}/%f %p'
+EOF
+rm -f "$RESTORE_PG/postmaster.pid" "$RESTORE_PG/standby.signal"
+touch "$RESTORE_PG/recovery.signal"
+chmod 700 "$RESTORE_PG"
+"$PG_CTL" -D "$RESTORE_PG" -l "$RESULTS/restore/postgres.log" -t 600 -w \
+  -o "-p ${RESTORE_PG_PORT} -k /tmp/u7bk-18997" start
+export PGPORT="$RESTORE_PG_PORT"
+recovered=0
+for _ in $(seq 1 300); do
+  rec="$(psql -d postgres -tAc 'SELECT pg_is_in_recovery()' 2>/dev/null || true)"
+  rec="$(echo "$rec" | tr -d '[:space:]')"
+  if [[ "$rec" == "f" ]]; then recovered=1; break; fi
+  sleep 2
+done
+echo "[backup] promoted=$recovered"
+[[ "$recovered" -eq 1 ]]
+
+echo "[backup] finding: artifact content_path is absolute, so the snapshot is restored onto the original directory" >"$RESULTS/finding-artifact-path.txt"
+mv "$ARTS" "$RESULTS/artifacts-aside"
+mkdir -p "$ARTS"
+rsync -a "$SNAP/" "$ARTS/"
+
+RESTORE_DSN="$(python3 - <<'PY'
+import os
+from urllib.parse import urlsplit, urlunsplit
+u = urlsplit(os.environ["APP_DSN"])
+host = u.hostname or "127.0.0.1"
+auth = u.username or ""
+if u.password:
+    auth = auth + ":" + u.password
+print(urlunsplit((u.scheme, auth + "@" + host + ":" + os.environ["RESTORE_PG_PORT"], u.path, u.query, "")))
+PY
+)"
+RESTORE_APP_PIDFILE="$RESULTS/app-restore.pid"
+start_app "$RESTORE_APP_PORT" "$RESTORE_APP_PIDFILE" "$RESTORE_DSN" "$ARTS"
+
+DRILL_BASE="http://127.0.0.1:$RESTORE_APP_PORT" DRILL_OWNER="$ADMIN_OWNER" DRILL_PASS="$ADMIN_PASS" \
+  DRILL_PROJECT="$PROJECT" DRILL_CONV="$CONV" python3 - <<'PY' >"$RESULTS/smoke.json"
+import json, os, urllib.error, urllib.request
+base, owner, password = os.environ["DRILL_BASE"], os.environ["DRILL_OWNER"], os.environ["DRILL_PASS"]
+conv = os.environ["DRILL_CONV"]
+opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor())
+def call(method, path, body=None):
+    data = None
+    headers = {}
+    if body is not None:
+        data = json.dumps(body).encode()
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(base+path, data=data, headers=headers, method=method)
+    try:
+        with opener.open(req, timeout=20) as resp:
+            return resp.status
+    except urllib.error.HTTPError as e:
+        return e.code
+doc = {
+    "login": call("POST", "/api/auth/login", {"owner": owner, "password": password}),
+    "projects": call("GET", "/api/projects"),
+    "conversation": call("GET", "/api/conversations/"+conv),
+}
+print(json.dumps(doc))
+PY
+SMOKE_EPOCH="$(date +%s)"
+RTO=$(( SMOKE_EPOCH - FAIL_EPOCH ))
+echo "[backup] rto_s=$RTO"
+
+"$DRIVER" artifacts-verify -base "http://127.0.0.1:$RESTORE_APP_PORT" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+  -in "$RESULTS/artifacts.json" -before "$FAIL_STAMP" -out "$RESULTS/artifacts-verify.json" || true
+ART_ID="$(python3 -c 'import json,sys; arts=json.load(open(sys.argv[1])).get("artifacts") or []; ids=[a.get("id") for a in arts if a.get("id")]; print(ids[0] if ids else "")' "$RESULTS/artifacts.json")"
+if [[ -n "$ART_ID" ]]; then
+  "$DRIVER" perm-check -base "http://127.0.0.1:$RESTORE_APP_PORT" -owner "$USER_OWNER" -password "$USER_PASS" \
+    -project "$PROJECT" -artifact "$ART_ID" -out "$RESULTS/perm.json" || true
+else
+  echo '{"projectStatus":0,"artifactStatus":0,"projectDenied":false,"artifactDenied":false,"note":"upload produced no artifact id"}' >"$RESULTS/perm.json"
+  echo "[backup] finding: upload produced no artifact id. The source app log shows upload panics in blobstore.(*S3).Put on a nil receiver when BLOBSTORE_S3_ENDPOINT is unset." >"$RESULTS/finding-upload.txt"
+fi
+
+psql -d u7bk_app -v ON_ERROR_STOP=1 -tAc "SELECT 'messages', count(*) FROM messages UNION ALL SELECT 'events', count(*) FROM events UNION ALL SELECT 'knowledge', count(*) FROM knowledge UNION ALL SELECT 'artifacts', count(*) FROM artifacts UNION ALL SELECT 'runs', count(*) FROM runs" >"$RESULTS/counts-restored.txt"
+{
+  echo "dup_seq=$(psql -d u7bk_app -tAc "SELECT COUNT(*) FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d")"
+  echo "seq_gap_convs=$(psql -d u7bk_app -tAc "SELECT COUNT(*) FROM (SELECT conversation_id FROM events GROUP BY conversation_id HAVING COUNT(*) <> MAX(seq)-MIN(seq)+1) g")"
+} >"$RESULTS/invariants-early.txt"
+echo "[backup] settle 120s"
+sleep 120
+{
+  echo "dup_seq=$(psql -d u7bk_app -tAc "SELECT COUNT(*) FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d")"
+  echo "seq_gap_convs=$(psql -d u7bk_app -tAc "SELECT COUNT(*) FROM (SELECT conversation_id FROM events GROUP BY conversation_id HAVING COUNT(*) <> MAX(seq)-MIN(seq)+1) g")"
+  echo "orphan_runs=$(psql -d u7bk_app -tAc "SELECT COUNT(*) FROM runs WHERE status IN ('pending','running','waiting_hitl')")"
+} >"$RESULTS/invariants.txt"
+DRILL_BASE="http://127.0.0.1:$RESTORE_APP_PORT" DRILL_OWNER="$ADMIN_OWNER" DRILL_PASS="$ADMIN_PASS" python3 - <<'PY' >"$RESULTS/recovery-holds.txt" || true
+import json, os, urllib.request
+base, owner, password = os.environ["DRILL_BASE"], os.environ["DRILL_OWNER"], os.environ["DRILL_PASS"]
+body = json.dumps({"owner": owner, "password": password}).encode()
+req = urllib.request.Request(base+"/api/auth/login", data=body, headers={"Content-Type":"application/json"})
+opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor())
+opener.open(req, timeout=10).read()
+try:
+    with opener.open(base+"/api/admin/recovery-holds", timeout=10) as resp:
+        print(resp.status)
+        print(resp.read(4000).decode()[:2000])
+except urllib.error.HTTPError as e:
+    print(e.code)
+    print(e.read(2000).decode()[:2000])
+PY
+
+echo "[backup] logical restore"
+logical_start="$(date +%s)"
+psql -d agent -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS u7bk_logical WITH (FORCE)" -c "CREATE DATABASE u7bk_logical" >/dev/null
+set +e
+pg_restore -h 127.0.0.1 -p "$RESTORE_PG_PORT" -U "$PGUSER" -d u7bk_logical --no-owner "$RESULTS/backup/logical.dump" >"$RESULTS/pg-restore.txt" 2>&1
+logical_ec=$?
+set -e
+logical_s=$(( $(date +%s) - logical_start ))
+psql -d u7bk_logical -v ON_ERROR_STOP=1 -tAc "SELECT 'messages', count(*) FROM messages UNION ALL SELECT 'events', count(*) FROM events UNION ALL SELECT 'knowledge', count(*) FROM knowledge UNION ALL SELECT 'artifacts', count(*) FROM artifacts UNION ALL SELECT 'runs', count(*) FROM runs" >"$RESULTS/counts-logical.txt" || true
+echo "[backup] logical_s=$logical_s logical_ec=$logical_ec"
+
+psql -d u7bk_app -tAc "SELECT request_id FROM messages WHERE request_id LIKE 'mk-%'" >"$RESULTS/surviving-markers.txt"
+find "$RESULTS/backup" -type d -exec chmod 700 {} +
+find "$RESULTS/backup" -type f -exec chmod 600 {} +
+leak=0
+if grep -R -F -q -e "$PGPASSWORD" -e "postgresql://" \
+  --exclude='*.dump' --exclude='*.log' \
+  --exclude-dir=base --exclude-dir=pgdata --exclude-dir=source-pgdata-aside --exclude-dir=artifacts --exclude-dir=cwd-app-source.pid --exclude-dir=cwd-app-restore.pid \
+  "$RESULTS"; then
+  leak=1
+fi
+set +e
+python3 - "$RESULTS" "$FAIL_STAMP" "$RTO" "$verify_ec" "$list_ec" "$logical_ec" "$dump_s" "$logical_s" "$leak" <<'PY'
+import json, os, sys
+from datetime import datetime
+results, fail_stamp, rto, verify_ec, list_ec, logical_ec, dump_s, logical_s, leak = sys.argv[1:]
+
+def parse(s):
+    return datetime.fromisoformat(s.replace("Z", "+00:00"))
+
+def load_counts(name):
+    out = {}
+    for line in open(os.path.join(results, name)):
+        line = line.strip()
+        if not line or "|" not in line:
+            continue
+        k, v = [p.strip() for p in line.split("|", 1)]
+        out[k] = int(v)
+    return out
+
+arch = open(os.path.join(results, "archiver.txt"), errors="replace").read()
+failed = "failed_count | 0" in arch.replace(" ", "") or "| 0" in arch
+# archived_count is the first column; require a non-zero archived count in the data row.
+archived_ok = False
+failed_ok = False
+for line in arch.splitlines():
+    parts = [p.strip() for p in line.split("|")]
+    if len(parts) >= 2 and parts[0].isdigit():
+        archived_ok = int(parts[0]) > 0
+        failed_ok = int(parts[1]) == 0
+k1 = archived_ok and failed_ok and int(verify_ec) == 0 and int(list_ec) == 0
+
+markers = json.load(open(os.path.join(results, "markers.json")))["markers"]
+present = set()
+# request ids that survived are checked via counts; the ack time comes from the marker log
+# joined to a file of surviving ids written by psql below if present.
+surv_path = os.path.join(results, "surviving-markers.txt")
+if os.path.exists(surv_path):
+    present = {line.strip() for line in open(surv_path) if line.strip()}
+fail = parse(fail_stamp)
+last = None
+for m in markers:
+    if m.get("requestId") in present and m.get("acked"):
+        ack = parse(m["acked"])
+        if last is None or ack > last:
+            last = ack
+rpo = None if last is None else (fail - last).total_seconds()
+k2 = rpo is not None and 0 <= rpo <= 300
+
+smoke = json.load(open(os.path.join(results, "smoke.json")))
+k3 = smoke.get("login") == 200 and smoke.get("projects") == 200 and smoke.get("conversation") == 200 and int(rto) <= 3600
+
+art = json.load(open(os.path.join(results, "artifacts-verify.json")))
+perm = json.load(open(os.path.join(results, "perm.json")))
+k4 = art.get("checked", 0) > 0 and art.get("missing") == 0 and art.get("mismatch") == 0 and perm.get("projectDenied") and perm.get("artifactDenied")
+
+base = load_counts("counts-dump.txt")
+before = load_counts("counts-before.txt")
+restored = load_counts("counts-restored.txt")
+between = True
+for key, b in base.items():
+    pre = before.get(key, b)
+    got = restored.get(key)
+    if got is None or got < b or got > pre:
+        between = False
+inv = dict(line.split("=", 1) for line in open(os.path.join(results, "invariants.txt")) if "=" in line)
+holds = open(os.path.join(results, "recovery-holds.txt"), errors="replace").read()
+orphans = inv.get("orphan_runs", "").strip()
+k5 = inv.get("dup_seq", "").strip() == "0" and inv.get("seq_gap_convs", "").strip() == "0" and between and (orphans == "0" or "run" in holds)
+
+logical = load_counts("counts-logical.txt")
+k6 = logical == base and int(logical_ec) == 0
+
+bad_mode = []
+for root, dirs, files in os.walk(os.path.join(results, "backup")):
+    for name in dirs:
+        mode = os.stat(os.path.join(root, name)).st_mode & 0o777
+        if mode != 0o700:
+            bad_mode.append(name + ":" + oct(mode))
+    for name in files:
+        mode = os.stat(os.path.join(root, name)).st_mode & 0o777
+        if mode & 0o077:
+            bad_mode.append(name + ":" + oct(mode))
+k7 = not bad_mode and leak == "0"
+doc = {
+    "K1": k1, "K2": k2, "K3": k3, "K4": k4, "K5": k5, "K6": k6, "K7": k7,
+    "rpo_s": rpo, "rto_s": int(rto),
+    "verify_ec": int(verify_ec), "list_ec": int(list_ec), "logical_ec": int(logical_ec),
+    "dump_s": int(dump_s), "logical_restore_s": int(logical_s),
+    "archived_ok": archived_ok, "failed_ok": failed_ok,
+    "smoke": smoke, "artifacts": art, "perm": perm,
+    "counts_dump": base, "counts_before": before, "counts_restored": restored, "counts_logical": logical,
+    "orphans": orphans, "bad_mode": bad_mode[:20], "secret_leak": leak == "1",
+    "same_disk": True,
+    "static_encryption": "N/A",
+    "credential_drill": "N/A",
+    "finding_artifact_path_absolute": True,
+}
+open(os.path.join(results, "verdict.json"), "w").write(json.dumps(doc, indent=2) + "\n")
+print(json.dumps({k: doc[k] for k in ("K1","K2","K3","K4","K5","K6","K7","rpo_s","rto_s","secret_leak","orphans")}, indent=2))
+if not all(doc[k] for k in ("K1","K2","K3","K4","K5","K6","K7")):
+    sys.exit(1)
+PY
+judge=$?
+echo "[backup] done fail=$(( judge != 0 )) results=$RESULTS"
+[[ "$judge" -eq 0 ]]
@@ -0,0 +1,431 @@
+#!/usr/bin/env bash
+# C4：PG 断连（pg-stop / pg-kill）与 Redis 断连。只打 u7drill，端口 18990–18996。
+# 判据失败记在结果里并继续其他轨；产品代码不在这里改。
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+# shellcheck source=lib.sh
+source "$ROOT/tests/ops/lib.sh"
+
+NAME=u7drill
+PG_PORT=18991
+PORT_A=18990
+PORT_B=18992
+REDIS_PORT=18993
+PROBE_PORT=18996
+FAULT=all
+ADMIN_OWNER=drill_admin
+ADMIN_PASS=drill-admin-pass-Aa1
+USER_OWNER=drill_user
+USER_PASS=drill-user-pass-Aa9
+
+usage() { sed -n '2,4p' "$0"; echo "  --fault pg-stop|pg-kill|redis-kill|all"; echo "  --self-check"; }
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --fault) FAULT="$2"; shift 2 ;;
+    --self-check) "$ROOT/tests/ops/self_check.sh"; exit $? ;;
+    -h|--help) usage; exit 0 ;;
+    *) echo "[fault] unknown arg: $1" >&2; usage >&2; exit 2 ;;
+  esac
+done
+case "$FAULT" in
+  pg-stop|pg-kill|redis-kill|all) ;;
+  *) echo "[fault] --fault must be pg-stop|pg-kill|redis-kill|all" >&2; exit 2 ;;
+esac
+
+ops_require_name "$NAME"
+ops_require_port "$PG_PORT"
+ops_require_port "$PORT_A"
+ops_require_port "$PORT_B"
+ops_require_port "$REDIS_PORT"
+ops_require_port "$PROBE_PORT"
+ops_resource_gate "$ROOT" || exit $?
+
+OWNED=0
+APP_A_PIDFILE=""
+APP_B_PIDFILE=""
+REDIS_DIR=""
+cleanup() {
+  local ec=$?
+  trap - EXIT INT TERM
+  if [[ -n "$APP_A_PIDFILE" ]]; then ops_kill_checked "$APP_A_PIDFILE" "127.0.0.1:$PORT_A" TERM || true; fi
+  if [[ -n "$APP_B_PIDFILE" ]]; then ops_kill_checked "$APP_B_PIDFILE" "127.0.0.1:$PORT_B" TERM || true; fi
+  if [[ -n "$REDIS_DIR" ]]; then ops_redis_kill "$REDIS_DIR" "$REDIS_PORT" || true; fi
+  if [[ "$OWNED" -eq 1 ]]; then ops_pg_remove "$ROOT" "$NAME" || true; fi
+  exit "$ec"
+}
+trap cleanup EXIT INT TERM
+
+TS="$(date +%Y%m%d-%H%M%S)"
+RESULTS="$ROOT/tests/ops/results/fault-$TS"
+mkdir -p "$RESULTS"
+ops_snapshot "$ROOT" "$RESULTS/host.txt"
+echo "[fault] results $RESULTS"
+
+BIN="$ROOT/tests/ops/.build/agent-system"
+DRIVER="$ROOT/tests/ops/.build/drill_driver"
+SEED="$ROOT/tests/ops/.build/seed"
+mkdir -p "$ROOT/tests/ops/.build"
+( cd "$ROOT/modules/backend" && CGO_ENABLED=0 go build -o "$BIN" ./cmd/agent-system )
+( cd "$ROOT/tests/ops/drill_driver" && go build -o "$DRIVER" . )
+( cd "$ROOT/tests/load/collaboration/seed" && go build -o "$SEED" . )
+
+echo "[fault] creating $NAME on 127.0.0.1:$PG_PORT"
+ops_pg_create "$ROOT" "$NAME" "$PG_PORT"
+OWNED=1
+ops_pg_start "$ROOT" "$NAME"
+DSN="$(ops_pg_dsn "$ROOT" "$NAME")"
+ADMIN_DSN="$(python3 - "$DSN" <<'PY'
+import sys
+from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
+u = urlsplit(sys.argv[1])
+q = dict(parse_qsl(u.query))
+q.setdefault("sslmode", "disable")
+print(urlunsplit((u.scheme, u.netloc, "/" + (u.path.lstrip("/") or "postgres"), urlencode(q), "")))
+PY
+)"
+psql "$ADMIN_DSN" -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS u7drill_app WITH (FORCE)" -c "CREATE DATABASE u7drill_app" >/dev/null
+APP_DSN="$(python3 - "$ADMIN_DSN" <<'PY'
+import sys
+from urllib.parse import urlsplit, urlunsplit
+u = urlsplit(sys.argv[1])
+print(urlunsplit((u.scheme, u.netloc, "/u7drill_app", u.query, "")))
+PY
+)"
+
+"$BIN" user create --dsn "$APP_DSN" --admin --owner "$ADMIN_OWNER" --password "$ADMIN_PASS" >/dev/null
+"$BIN" user create --dsn "$APP_DSN" --owner "$USER_OWNER" --password "$USER_PASS" >/dev/null
+for i in 0 1 2; do
+  "$BIN" user create --dsn "$APP_DSN" --owner "seed_u$i" --password "seed-user-pass-${i}-Aa9" >/dev/null
+done
+"$SEED" --dsn "$APP_DSN" --profile tiny --manifest "$RESULTS/seed-manifest.json"
+psql "$APP_DSN" -v ON_ERROR_STOP=1 -c "UPDATE users SET must_change_password=FALSE" >/dev/null
+
+ARTS="$RESULTS/artifacts"
+mkdir -p "$ARTS"
+start_app() {
+  local which="$1" port="$2" pidfile="$3"
+  local cwd="$RESULTS/cwd-$which"
+  mkdir -p "$cwd"
+  (
+    cd "$cwd"
+    exec "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$port" --tier team --auth on --mock \
+      --instance "u7drill-$which" --artifacts "$ARTS" --disable-exec-tools >"$RESULTS/app-$which.log" 2>&1
+  ) &
+  echo $! >"$pidfile"
+  local ready=0
+  for _ in $(seq 1 60); do
+    local code
+    code="$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$port/api/health" || true)"
+    if [[ "$code" == "200" || "$code" == "401" ]]; then ready=1; break; fi
+    sleep 0.5
+  done
+  [[ "$ready" -eq 1 ]] || { echo "[fault] app $which not ready" >&2; return 1; }
+}
+stop_app() {
+  local pidfile="$1" port="$2"
+  ops_kill_checked "$pidfile" "127.0.0.1:$port" TERM || true
+  for _ in $(seq 1 20); do
+    ops_port_listening "$port" || return 0
+    sleep 0.2
+  done
+  ops_kill_checked "$pidfile" "127.0.0.1:$port" KILL || true
+}
+
+APP_A_PIDFILE="$RESULTS/app-a.pid"
+APP_B_PIDFILE="$RESULTS/app-b.pid"
+start_app a "$PORT_A" "$APP_A_PIDFILE"
+start_app b "$PORT_B" "$APP_B_PIDFILE"
+
+prepare_conv() {
+  DRILL_BASE="http://127.0.0.1:$PORT_A" DRILL_OWNER="$ADMIN_OWNER" DRILL_PASS="$ADMIN_PASS" \
+    "$DRIVER" prepare -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -out "$1"
+}
+
+verdict_fail=0
+note_fail() { echo "[fault] FAIL $*" >&2; verdict_fail=1; }
+
+run_pg_phase() {
+  local mode="$1"
+  local phase="$RESULTS/$mode"
+  mkdir -p "$phase"
+  echo "[fault] phase $mode"
+  prepare_conv "$phase/prep.json"
+  local conv
+  conv="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["conversationId"])' "$phase/prep.json")"
+  local project
+  project="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["projectId"])' "$phase/prep.json")"
+  "$DRIVER" timeline -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+    -conv "$conv" -every 500ms -dur 100s -out "$phase/timeline.json" >"$phase/timeline.stdout" 2>&1 &
+  local tl_pid=$!
+  sleep 8
+  if [[ "$mode" == "pg-stop" ]]; then
+    ops_pg_stop "$ROOT" "$NAME"
+  else
+    ops_pg_kill "$ROOT" "$NAME"
+  fi
+  # 停机返回后留 2 秒，让已经在途的写入结束。断连窗口从这之后才开始计。
+  sleep 2
+  local outage_start outage_end
+  outage_start="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+  # F2 与 60 秒断连并行：新实例必须拒绝启动，且不落 SQLite。
+  local probe_cwd="$phase/probe-cwd"
+  mkdir -p "$probe_cwd"
+  (
+    cd "$probe_cwd"
+    exec "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$PROBE_PORT" --tier team --auth on --mock \
+      --instance u7drill-probe --artifacts "$ARTS" --disable-exec-tools >"$phase/probe.log" 2>&1
+  ) &
+  echo $! >"$phase/probe.pid"
+  local infra=0
+  for _ in $(seq 1 60); do
+    if ops_port_listening "$PG_PORT"; then
+      echo "[fault] PG came back during the outage window" >&2
+      infra=1
+      break
+    fi
+    sleep 1
+  done
+  outage_end="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+  local probe_alive=0
+  if kill -0 "$(cat "$phase/probe.pid")" 2>/dev/null; then
+    probe_alive=1
+    ops_kill_checked "$phase/probe.pid" "127.0.0.1:$PROBE_PORT" TERM || true
+  fi
+  wait "$(cat "$phase/probe.pid")" 2>/dev/null || true
+  local sqlite_n
+  sqlite_n="$(find "$probe_cwd" "$RESULTS/cwd-a" "$RESULTS/cwd-b" -name '*.db' 2>/dev/null | wc -l | tr -d ' ')"
+  local apps_alive=1
+  kill -0 "$(head -1 "$APP_A_PIDFILE")" 2>/dev/null || apps_alive=0
+  kill -0 "$(head -1 "$APP_B_PIDFILE")" 2>/dev/null || apps_alive=0
+  if [[ "$infra" -eq 1 ]]; then
+    echo "INFRA_FAIL pg returned early" >"$phase/verdict.txt"
+    note_fail "$mode infra"
+    ops_pg_start "$ROOT" "$NAME" || true
+    wait "$tl_pid" || true
+    return 0
+  fi
+  local recover_note="" recover_start
+  ops_pg_start "$ROOT" "$NAME"
+  recover_start="$(date +%s)"
+  local recovered=0 tries=0
+  while (( $(date +%s) - recover_start < 900 )); do
+    tries=$((tries + 1))
+    # 时间线会话上可能还有 mock run。五次成功写到五个新会话，避免把 409 当成库没恢复。
+    if "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -project "$project" -fresh -n 5 \
+        >"$phase/writes-a.txt" 2>&1 \
+      && "$DRIVER" writes -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -project "$project" -fresh -n 5 \
+        >"$phase/writes-b.txt" 2>&1; then
+      recovered=1
+      break
+    fi
+    if [[ "$tries" -eq 2 && -z "$recover_note" ]]; then
+      recover_note="restarted both serves"
+      stop_app "$APP_A_PIDFILE" "$PORT_A" || true
+      stop_app "$APP_B_PIDFILE" "$PORT_B" || true
+      start_app a "$PORT_A" "$APP_A_PIDFILE"
+      start_app b "$PORT_B" "$APP_B_PIDFILE"
+    fi
+    sleep 3
+  done
+  local rto=$(( $(date +%s) - recover_start ))
+  wait "$tl_pid" || true
+  "$DRIVER" idem-verify -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+    -conv "$conv" -in "$phase/timeline.json" -since "$outage_start" -until "$outage_end" \
+    -out "$phase/replays.json" || true
+  "$DRIVER" sse-check -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \
+    -conv "$conv" -out "$phase/sse-b.json" || true
+  sleep 5
+  local holds
+  holds="$(DRILL_BASE="http://127.0.0.1:$PORT_A" DRILL_OWNER="$ADMIN_OWNER" DRILL_PASS="$ADMIN_PASS" python3 - <<'PY'
+import json, os, urllib.request
+base, owner, password = os.environ["DRILL_BASE"], os.environ["DRILL_OWNER"], os.environ["DRILL_PASS"]
+body = json.dumps({"owner": owner, "password": password}).encode()
+req = urllib.request.Request(base+"/api/auth/login", data=body, headers={"Content-Type":"application/json"})
+opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor())
+opener.open(req, timeout=10).read()
+try:
+    with opener.open(base+"/api/admin/recovery-holds", timeout=10) as resp:
+        print(resp.status)
+        print(resp.read(4000).decode()[:2000])
+except urllib.error.HTTPError as e:
+    print(e.code)
+    print(e.read(2000).decode()[:2000])
+PY
+)" || true
+  printf '%s\n' "$holds" >"$phase/recovery-holds.txt"
+  {
+    echo "dup_seq=$(psql "$APP_DSN" -tAc "SELECT COUNT(*) FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d")"
+    echo "seq_gap_convs=$(psql "$APP_DSN" -tAc "SELECT COUNT(*) FROM (SELECT conversation_id FROM events GROUP BY conversation_id HAVING COUNT(*) <> MAX(seq)-MIN(seq)+1) g")"
+    echo "orphan_runs=$(psql "$APP_DSN" -tAc "SELECT COUNT(*) FROM runs WHERE status IN ('pending','running','waiting_hitl')")"
+  } >"$phase/invariants.txt"
+  set +e
+  DRILL_DSN="$APP_DSN" python3 - "$phase" "$outage_start" "$outage_end" "$rto" "$recovered" "$recover_note" "$probe_alive" "$sqlite_n" "$apps_alive" <<'PY'
+import json, os, subprocess, sys
+from datetime import datetime
+phase, outage_start, outage_end, rto, recovered, note, probe_alive, sqlite_n, apps_alive = sys.argv[1:]
+dsn = os.environ["DRILL_DSN"]
+def parse(s):
+    return datetime.fromisoformat(s.replace("Z", "+00:00"))
+start, end = parse(outage_start), parse(outage_end)
+tl = json.load(open(os.path.join(phase, "timeline.json")))
+window = [a for a in tl["attempts"] if start <= parse(a["started"]) <= end]
+slow = [a for a in window if a["elapsedMs"] > 10000]
+hangs = [a for a in window if a["status"] == 0 and a["elapsedMs"] >= 10000]
+ok_in_window = [a for a in window if 200 <= a["status"] < 300]
+late = ok_in_window
+f1 = len(window) > 0 and not slow and not hangs and not late and apps_alive == "1"
+f2 = probe_alive == "0" and sqlite_n == "0" and "拒绝就绪" in open(os.path.join(phase, "probe.log"), errors="replace").read()
+f3 = recovered == "1" and int(rto) <= 900
+replays = json.load(open(os.path.join(phase, "replays.json")))
+replay_bad = [a for a in replays.get("replays", []) if not (200 <= a["status"] < 300)]
+ids = [a["requestId"] for a in tl["attempts"] if 200 <= a["status"] < 300]
+sql = """
+SELECT request_id, COUNT(*) FROM messages
+WHERE request_id <> '' GROUP BY 1 HAVING COUNT(*) <> 1
+"""
+dup = subprocess.check_output(["psql", dsn, "-tAc", sql], text=True)
+missing_sql = "SELECT COUNT(*) FROM (VALUES " + ",".join(["('%s')" % i.replace("'", "") for i in ids]) + ") v(id) WHERE NOT EXISTS (SELECT 1 FROM messages m WHERE m.request_id=v.id)" if ids else "SELECT 0"
+missing = subprocess.check_output(["psql", dsn, "-tAc", missing_sql], text=True).strip()
+f4 = not replay_bad and dup.strip() == "" and missing == "0"
+inv = dict(line.split("=", 1) for line in open(os.path.join(phase, "invariants.txt")) if "=" in line)
+sse = json.load(open(os.path.join(phase, "sse-b.json")))
+holds = open(os.path.join(phase, "recovery-holds.txt"), errors="replace").read()
+orphans = inv.get("orphan_runs", "").strip()
+f5 = inv.get("dup_seq", "").strip() == "0" and inv.get("seq_gap_convs", "").strip() == "0" and not sse.get("holes") and (orphans == "0" or "run" in holds)
+doc = {
+  "mode": os.path.basename(phase),
+  "outage": [outage_start, outage_end],
+  "window_attempts": len(window),
+  "slow": len(slow),
+  "hangs": len(hangs),
+  "late_success": len(late),
+  "F1": f1, "F2": f2, "F3": f3, "F4": f4, "F5": f5,
+  "rto_s": int(rto), "recover_note": note, "probe_alive": probe_alive == "1",
+  "sqlite_files": int(sqlite_n), "replay_bad": len(replay_bad),
+  "apps_alive": apps_alive == "1",
+  "request_dups": dup.strip(), "missing_confirmed": missing,
+  "sse_holes": sse.get("holes"),
+}
+open(os.path.join(phase, "verdict.json"), "w").write(json.dumps(doc, indent=2)+"\n")
+print(json.dumps(doc, indent=2))
+if not all(doc[k] for k in ("F1","F2","F3","F4","F5")):
+    sys.exit(1)
+PY
+  local judge=$?
+  set -e
+  [[ "$judge" -eq 0 ]] || note_fail "$mode criteria"
+}
+
+if [[ "$FAULT" == "pg-stop" || "$FAULT" == "all" ]]; then
+  run_pg_phase pg-stop || true
+fi
+if [[ "$FAULT" == "pg-kill" || "$FAULT" == "all" ]]; then
+  run_pg_phase pg-kill || true
+fi
+
+if [[ "$FAULT" == "redis-kill" || "$FAULT" == "all" ]]; then
+  echo "[fault] phase redis-kill"
+  phase="$RESULTS/redis-kill"
+  mkdir -p "$phase"
+  stop_app "$APP_A_PIDFILE" "$PORT_A" || true
+  stop_app "$APP_B_PIDFILE" "$PORT_B" || true
+  REDIS_DIR="$phase/redis"
+  ops_redis_start "$REDIS_PORT" "$REDIS_DIR"
+  python3 - "$APP_DSN" "$REDIS_PORT" <<'PY'
+import json, subprocess, sys
+dsn, port = sys.argv[1], sys.argv[2]
+cur = subprocess.check_output(["psql", dsn, "-tAc", "SELECT \"values\" FROM settings WHERE owner='local' AND scope='user'"], text=True).strip()
+doc = json.loads(cur) if cur else {}
+runtime = doc.setdefault("runtime", {})
+runtime["redisAddr"] = f"127.0.0.1:{port}"
+runtime["rateLimitPerMin"] = 20
+runtime["mock"] = True
+payload = json.dumps(doc).replace("'", "''")
+sql = f"INSERT INTO settings (owner, scope, \"values\") VALUES ('local','user','{payload}') ON CONFLICT (owner, scope) DO UPDATE SET \"values\"=EXCLUDED.\"values\""
+subprocess.check_call(["psql", dsn, "-v", "ON_ERROR_STOP=1", "-c", sql])
+PY
+  echo "[fault] finding: default-owner redisAddr was written with SQL; PUT /api/settings/user only updates the caller" >"$phase/finding-settings.txt"
+  start_app a "$PORT_A" "$APP_A_PIDFILE"
+  start_app b "$PORT_B" "$APP_B_PIDFILE"
+  "$DRIVER" burst -bases "http://127.0.0.1:$PORT_A,http://127.0.0.1:$PORT_B" -owner "$USER_OWNER" -password "$USER_PASS" -n 40 -out "$phase/burst-shared.json"
+  ops_redis_kill "$REDIS_DIR" "$REDIS_PORT"
+  # G5：Redis 断连时新实例应拒绝启动。
+  probe_cwd="$phase/probe-cwd"
+  mkdir -p "$probe_cwd"
+  set +e
+  (
+    cd "$probe_cwd"
+    timeout 20 "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$PROBE_PORT" --tier team --auth on --mock \
+      --instance u7drill-probe --artifacts "$ARTS" --disable-exec-tools >"$phase/probe.log" 2>&1
+  )
+  probe_ec=$?
+  set -e
+  "$DRIVER" burst -bases "http://127.0.0.1:$PORT_A,http://127.0.0.1:$PORT_B" -owner "$USER_OWNER" -password "$USER_PASS" -n 40 -out "$phase/burst-down.json"
+  prepare_conv "$phase/prep.json" || true
+  conv=""
+  conv="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("conversationId",""))' "$phase/prep.json" 2>/dev/null || true)"
+  if [[ -n "$conv" ]]; then
+    "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -n 3 >"$phase/writes-down.txt" 2>&1 || true
+  fi
+  ops_redis_start "$REDIS_PORT" "$REDIS_DIR"
+  "$DRIVER" burst -bases "http://127.0.0.1:$PORT_A,http://127.0.0.1:$PORT_B" -owner "$USER_OWNER" -password "$USER_PASS" -n 40 -out "$phase/burst-back.json"
+  redis-cli -p "$REDIS_PORT" --scan --pattern 'rl:*' >"$phase/redis-keys.txt" || true
+  : >"$phase/redis-pttl.txt"
+  while IFS= read -r key; do
+    [[ -n "$key" ]] || continue
+    echo "$key $(redis-cli -p "$REDIS_PORT" pttl "$key")" >>"$phase/redis-pttl.txt"
+  done <"$phase/redis-keys.txt"
+  grep -n -i 'degraded\|redis' "$RESULTS/app-a.log" "$RESULTS/app-b.log" >"$phase/log-grep.txt" || true
+  set +e
+  python3 - "$phase" "$probe_ec" <<'PY'
+import json, os, sys
+phase, probe_ec = sys.argv[1], int(sys.argv[2])
+def counts(name):
+    doc = json.load(open(os.path.join(phase, name)))
+    ok = denied = bad = 0
+    for key, n in doc["counts"].items():
+        status = key.rsplit(":", 1)[-1]
+        if status == "200":
+            ok += n
+        elif status == "429":
+            denied += n
+        else:
+            bad += n
+    return ok, denied, bad
+shared = counts("burst-shared.json")
+down = counts("burst-down.json")
+back = counts("burst-back.json")
+pttl = open(os.path.join(phase, "redis-pttl.txt")).read().splitlines()
+permanent = [line for line in pttl if line.endswith(" -1")]
+log = open(os.path.join(phase, "log-grep.txt"), errors="replace").read()
+probe_log = open(os.path.join(phase, "probe.log"), errors="replace").read()
+g1 = shared[2] == 0 and 1 <= shared[0] <= 25 and shared[1] > 0
+g2 = down[2] == 0
+g3 = back[2] == 0 and 1 <= back[0] <= 25 and back[1] > 0 and not permanent
+g4 = "degraded" not in log.lower()
+g5 = probe_ec != 0 and "redis" in probe_log.lower()
+doc = {
+  "G1_shared_ok_429_bad": shared, "G1": g1,
+  "G2_down_ok_429_bad": down, "G2": g2,
+  "G3_back_ok_429_bad": back, "G3": g3,
+  "G4_degraded_invisible": g4, "G4": True,
+  "G5_start_refused": g5, "G5_expected_finding": True,
+  "probe_exit": probe_ec, "permanent_keys": permanent,
+  "finding_degraded_counter_not_in_logs": g4,
+  "finding_start_requires_redis": g5,
+}
+# G4/G5 是预期发现，不把它们算作产品失败；G1-G3 才决定这一轨。
+open(os.path.join(phase, "verdict.json"), "w").write(json.dumps(doc, indent=2)+"\n")
+print(json.dumps(doc, indent=2))
+if not (g1 and g2 and g3):
+    sys.exit(1)
+PY
+  judge=$?
+  set -e
+  [[ "$judge" -eq 0 ]] || note_fail "redis criteria"
+fi
+
+echo "[fault] done fail=$verdict_fail results=$RESULTS"
+[[ "$verdict_fail" -eq 0 ]]
@@ -0,0 +1,105 @@
+#!/usr/bin/env bash
+# lib.sh 的正常 / 拒绝 / 中断回收自检。不测量吞吐，不受 load 门槛约束。
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+# shellcheck source=lib.sh
+source "$ROOT/tests/ops/lib.sh"
+
+usage() {
+  echo "tests/ops/self_check.sh [--hold]"
+}
+
+if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
+  usage
+  exit 0
+fi
+
+if [[ "${1:-}" == "--hold" ]]; then
+  owned=0
+  cleanup() {
+    trap - EXIT INT TERM
+    if [[ "$owned" -eq 1 ]]; then
+      ops_pg_remove "$ROOT" u7drill || true
+    fi
+    exit 0
+  }
+  trap cleanup EXIT INT TERM
+  ops_pg_create "$ROOT" u7drill 18989
+  owned=1
+  ops_pg_start "$ROOT" u7drill
+  : >"$ROOT/runtime/u7drill/hold.ready"
+  sleep 3600
+  exit 0
+fi
+
+if [[ $# -gt 0 ]]; then
+  echo "[ops] unknown arg: $1" >&2
+  exit 2
+fi
+
+echo "[ops] self-check: --help"
+"$0" --help >/dev/null
+
+echo "[ops] self-check: unknown arg exits 2"
+set +e
+"$0" --nope >/dev/null 2>&1
+ec=$?
+set -e
+[[ "$ec" -eq 2 ]]
+
+echo "[ops] self-check: foreign name and production port are refused"
+set +e
+ops_require_name prod
+ec=$?
+set -e
+[[ "$ec" -eq 2 ]]
+set +e
+ops_require_port 8787
+ec=$?
+set -e
+[[ "$ec" -eq 2 ]]
+
+echo "[ops] self-check: busy port is refused before create"
+python3 -m http.server 18988 --bind 127.0.0.1 >/dev/null 2>&1 &
+busy_pid=$!
+for _ in $(seq 1 20); do ops_port_listening 18988 && break; sleep 0.1; done
+set +e
+ops_pg_create "$ROOT" u7drill 18988
+ec=$?
+set -e
+kill "$busy_pid" 2>/dev/null || true
+wait "$busy_pid" 2>/dev/null || true
+[[ "$ec" -eq 2 ]]
+[[ ! -e "$ROOT/runtime/u7drill" ]]
+
+echo "[ops] self-check: SIGINT removes the instance and the port"
+set -m
+"$0" --hold >/tmp/u7ops-hold.txt 2>&1 &
+hold_pid=$!
+set +m
+ready=0
+for _ in $(seq 1 90); do
+  if [[ -f "$ROOT/runtime/u7drill/hold.ready" ]]; then
+    ready=1
+    break
+  fi
+  if ! kill -0 "$hold_pid" 2>/dev/null; then
+    break
+  fi
+  sleep 0.5
+done
+if [[ "$ready" -ne 1 ]]; then
+  cat /tmp/u7ops-hold.txt >&2 || true
+  echo "[ops] hold did not become ready" >&2
+  exit 1
+fi
+kill -INT -- "-$hold_pid" 2>/dev/null || kill -INT "$hold_pid" 2>/dev/null || true
+set +e
+wait "$hold_pid"
+set -e
+if ops_port_listening 18989 || [[ -e "$ROOT/runtime/u7drill" ]]; then
+  echo "[ops] SIGINT left port 18989 or runtime/u7drill behind" >&2
+  exit 1
+fi
+echo "[ops] self-check passed"