Add the C4 disconnect drill for PostgreSQL and Redis

Why: The release checklist cannot mark disconnect recovery until an isolated drill records bounded failures, no SQLite fallback, and idempotent writes. Evidence: bash -n passed, drill_driver go test passed, unknown arguments exit 2. The timed drill is recorded next. Impact: New drill entry only. No product change and no release decision. Refs: docs/release-multi-instance.md C4 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-12 13:33:45 +00:00
parent d7f8e711ee
commit 4c2e326c03
2 changed files with 449 additions and 1 deletions
@@ -45,6 +45,8 @@ func main() {
 		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
@@ -59,7 +61,7 @@ func main() {
 }

 func usage() {
-	fmt.Fprintln(os.Stderr, "drill_driver prepare|timeline|idem-verify|sse-check|markers|artifacts-put|artifacts-verify|burst|perm-check")
+	fmt.Fprintln(os.Stderr, "drill_driver prepare|timeline|idem-verify|sse-check|markers|artifacts-put|artifacts-verify|burst|perm-check|writes")
 }

 type session struct {
@@ -508,6 +510,33 @@ func cmdBurst(args []string) error {
 	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")
+	n := fs.Int("n", 5, "consecutive successful writes required")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	s, err := newSession(*base, *owner, *password)
+	if err != nil {
+		return err
+	}
+	for i := 0; i < *n; i++ {
+		reqID := fmt.Sprintf("wr-%d-%d", time.Now().UnixNano(), i)
+		res, err := s.do("POST", "/api/conversations/"+*conv+"/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")
@@ -0,0 +1,419 @@
+#!/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")"
+  "$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
+  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 ' ')"
+  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))
+    if "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -n 5 \
+        >"$phase/writes-a.txt" 2>&1 \
+      && "$DRIVER" writes -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -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" -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" <<'PY'
+import json, os, subprocess, sys
+from datetime import datetime
+phase, outage_start, outage_end, rto, recovered, note, probe_alive, sqlite_n = 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
+apps_alive = True
+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),
+  "request_dups": dup.strip(), "missing_confirmed": missing,
+  "sse_holes": sse.get("holes"), "apps_alive": apps_alive,
+}
+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 断连时新实例应拒绝启动。
+  local 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
+  )
+  local 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
+  local 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
+  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
+  [[ $? -eq 0 ]] || note_fail "redis criteria"
+fi
+
+echo "[fault] done fail=$verdict_fail results=$RESULTS"
+[[ "$verdict_fail" -eq 0 ]]