Add the C5 backup and PITR drill

Why: The checklist had no measured restore path for a physical backup, WAL archive, or pg_dump onto a new target. Evidence: bash -n, go test in tests/ops/drill_driver, and run_backup_drill.sh --self-check passed locally (SELF). The timed run is recorded separately. Impact: Isolated u7bk only. Upload responses keep artifactId so the restore check can download the same objects. Refs: feature/u7-ops-drills Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-12 14:55:48 +00:00
parent 3240697b49
commit 19bc78d6c0
3 changed files with 463 additions and 4 deletions
@@ -452,14 +452,22 @@ func upload(s *session, project, name string, payload []byte) (id string, status
 	}
 	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"`
+		ID         string `json:"id"`
+		ArtifactID string `json:"artifactId"`
 	}
 	_ = json.Unmarshal(raw, &body)
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return body.ID, resp.StatusCode, string(raw), acked
+	if body.ID != "" {
+		return body.ID
 	}
-	return body.ID, resp.StatusCode, "", acked
+	return body.ArtifactID
 }

 func cmdArtifactsVerify(args []string) error {
@@ -2,6 +2,15 @@ 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
@@ -0,0 +1,442 @@
+#!/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
+  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" "$RESULTS/restore/sock"
+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 ${RESULTS}/restore/sock" 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]))["artifacts"]; print(next(a["id"] for a in arts if a.get("id")))' "$RESULTS/artifacts.json")"
+"$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
+
+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 ]]