diff --git a/.github/scripts/oswal-s3-bench-run.sh b/.github/scripts/oswal-s3-bench-run.sh
index b2482332f85..84af2c27032 100644
--- a/.github/scripts/oswal-s3-bench-run.sh
+++ b/.github/scripts/oswal-s3-bench-run.sh
@@ -1,14 +1,40 @@
 #!/usr/bin/env bash
 # Host side of the object store WAL S3 benchmark (oswal-s3-bench.yml).
-#   wal-disk DIR  mount the instance store at DIR when the host has one
-#   generate      generate the dataset once with the o11ybench runtime
-#   cells         run every cell of $PLAN, each on a fresh GreptimeDB container
+#   check-runtime IMAGE  resolve the o11ybench runtime and check it carries the loader and generator
+#   pull                 pull the runtime and GreptimeDB images and record their metadata
+#   wal-disk DIR         mount the instance store at DIR when the host has one
+#   generate             generate the dataset once with the o11ybench runtime
+#   cells                run every cell of $PLAN, each on a fresh GreptimeDB container
+#   scrub                refuse results that contain a credential value
 set -euo pipefail
 
-: "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}"
+here=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+
 # Same isolation as the o11ybench runtime's own docker run.
-runtime=(docker run --rm --user "$(id -u):$(id -g)" --label "o11ybench.owner=$OWNER"
-    --security-opt no-new-privileges --cap-drop ALL)
+runtime() {
+    docker run --rm --user "$(id -u):$(id -g)" --label "o11ybench.owner=$OWNER" \
+        --security-opt no-new-privileges --cap-drop ALL "$@"
+}
+
+check_runtime() {
+    local image=$1 resolved
+    [[ "$image" == public.ecr.aws/* ]] || { echo 'The runtime image must come from AWS ECR Public' >&2; return 1; }
+    bash "$here/pull-benchmark-image.sh" "$image"
+    resolved=$(docker image inspect "$image" | jq -er '.[0].RepoDigests[0]')
+    docker run --rm --entrypoint sh "$resolved" -c \
+        'test -f scripts/agent-observability/greptimedb.sh && test -f scripts/generate_agent_observations.py' \
+        || { echo 'The runtime image lacks the agent observability loader or generator' >&2; return 1; }
+    printf 'runtime_image=%s\n' "$resolved" >> "${GITHUB_OUTPUT:-/dev/stdout}"
+}
+
+pull_images() {
+    : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_TAG:?}"
+    mkdir -p "$RESULTS/images"
+    bash "$here/pull-benchmark-image.sh" "$RUNTIME_IMAGE"
+    bash "$here/pull-benchmark-image.sh" "greptime/greptimedb:$GREPTIMEDB_TAG"
+    docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json"
+    docker image inspect "greptime/greptimedb:$GREPTIMEDB_TAG" > "$RESULTS/images/greptimedb.json"
+}
 
 wal_disk() {
     local root=$1 device
@@ -24,11 +50,11 @@ wal_disk() {
 }
 
 generate() {
-    local args
-    mapfile -t args < <(jq -r '.generator_args[]' "$PLAN")
+    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}"
+    local args=() arg
+    while IFS= read -r arg; do args+=("$arg"); done < <(jq -r '.generator_args[]' "$PLAN")
     mkdir -p "$CORPUS_DIR"
-    "${runtime[@]}" --name "$OWNER-generator" \
-        --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus" \
+    runtime --name "$OWNER-generator" --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus" \
         "$RUNTIME_IMAGE" scripts/generate_agent_observations.py "${args[@]}" \
         --output /work/corpus/agent_observations.jsonl \
         --summary-output /work/corpus/summary.json --evidence-output /work/corpus/evidence.json \
@@ -36,22 +62,50 @@ generate() {
     jq -e --argjson rows "$(jq .rows "$PLAN")" '.line_count == $rows' "$CORPUS_DIR/summary.json" >/dev/null
 }
 
-# One cell in its own process, so errexit holds inside it.
+# Scrape once, after the load succeeded or failed, while the server still runs.
+scrape_metrics() {
+    if curl --noproxy '*' -fsS --max-time 30 http://127.0.0.1:4000/metrics > "$dir/metrics.all"; then
+        grep -E '^(# (HELP|TYPE) )?greptime_logstore_object_store_wal_' "$dir/metrics.all" > "$dir/metrics.prom" || true
+    fi
+    rm -f "$dir/metrics.all"
+}
+
+finish() {
+    local rc=$?
+    trap - EXIT
+    set +e
+    if [[ -n "$cid" ]]; then
+        scrape_metrics
+        docker stop -t 120 "$cid" >/dev/null
+        docker logs "$cid" > "$dir/server.log" 2>&1
+        docker rm -fv "$cid" >/dev/null
+    fi
+    [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir"
+    jq -n --argjson spec "$spec" --arg device "$wal_device" --arg version "$version" \
+        --arg load_seconds "$load_seconds" --argjson exit_code "$rc" \
+        '$spec + {wal_device: $device, version: $version, exit_code: $exit_code,
+          load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json"
+    exit "$rc"
+}
+
+# One cell in its own process, so errexit holds inside it. The cell's state is
+# global because the EXIT trap runs after this function returns.
 cell() {
-    : "${GREPTIMEDB_IMAGE:?}" "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}"
+    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_IMAGE:?}"
+    : "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}"
     : "${AWS_ACCESS_KEY_ID:?}" "${AWS_SECRET_ACCESS_KEY:?}"
-    # Globals, not locals: the EXIT trap runs after this function returns.
     spec=$1 cid= wal_dir= version= load_seconds=
-    local concurrency config name
-    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec"); config=$(jq -r .config <<< "$spec")
+    local id concurrency name setting started ended
+    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec")
     dir="$RESULTS/cells/$id"; name="$OWNER-$id"
     mkdir -p "$dir"
-    # The prefix holds data/ and the WAL's default prefix wal/ side by side.
+    # The root holds data/ and the WAL's default prefix wal/ side by side.
     local settings=(STORAGE__TYPE=S3 "STORAGE__BUCKET=$S3_BUCKET" "STORAGE__REGION=$S3_REGION"
         "STORAGE__ROOT=$RUN_PREFIX/$id")
     [[ -z "${S3_CONTAINER_ENDPOINT:-}" ]] || settings+=("STORAGE__ENDPOINT=$S3_CONTAINER_ENDPOINT")
-    mapfile -t -O "${#settings[@]}" settings < <(jq -r '.wal | to_entries[] | "WAL__\(.key)=\(.value)"' <<< "$spec")
-    local docker_env=() mounts=() setting
+    while IFS= read -r setting; do settings+=("$setting"); done \
+        < <(jq -r '.wal | to_entries[] | "WAL__\(.key)=\(.value)"' <<< "$spec")
+    local docker_env=() mounts=()
     for setting in "${settings[@]}"; do
         docker_env+=(-e "GREPTIMEDB_STANDALONE__$setting")
         printf 'GREPTIMEDB_STANDALONE__%s\n' "$setting" >> "$dir/config.env"
@@ -68,26 +122,10 @@ cell() {
     else
         wal_device="s3://$S3_BUCKET/$RUN_PREFIX/$id/wal"
     fi
-    finish() {
-        local rc=$?
-        trap - EXIT
-        set +e
-        if [[ -n "$cid" ]]; then
-            docker stop -t 120 "$cid" >/dev/null
-            docker logs "$cid" > "$dir/server.log" 2>&1
-            docker rm -fv "$cid" >/dev/null
-        fi
-        [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir"
-        jq -n --argjson spec "$spec" --arg device "$wal_device" --arg version "$version" \
-            --arg load_seconds "$load_seconds" --argjson exit_code "$rc" \
-            '$spec + {wal_device: $device, version: $version, exit_code: $exit_code,
-              load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json"
-        exit "$rc"
-    }
     trap finish EXIT
     cid=$(docker run -d --init --name "$name" --label "o11ybench.owner=$OWNER" \
         --add-host host.docker.internal:host-gateway \
-        -p 127.0.0.1:4000:4000 -p 127.0.0.1:4001:4001 "${docker_env[@]}" "${mounts[@]}" \
+        -p 127.0.0.1:4000:4000 -p 127.0.0.1:4001:4001 "${docker_env[@]}" ${mounts[@]+"${mounts[@]}"} \
         "$GREPTIMEDB_IMAGE" standalone start --http-addr=0.0.0.0:4000 --grpc-bind-addr=0.0.0.0:4001)
     local deadline=$((SECONDS + 300))
     until curl --noproxy '*' -fsS --max-time 5 http://127.0.0.1:4000/health >/dev/null 2>&1; do
@@ -97,10 +135,9 @@ cell() {
     done
     version=$(docker exec "$cid" greptime --version | sed -n 's/^version: //p')
     [[ -n "$version" ]]
-    local started ended
     started=$(date +%s.%N)
     # Only the load phase, run exactly as run-agent-observability.sh runs it.
-    "${runtime[@]}" --name "$name-load" --network host \
+    runtime --name "$name-load" --network host \
         --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus,readonly" --mount "type=bind,src=$dir,dst=/results" \
         -e GRPC_ENDPOINT=127.0.0.1:4001 -e "DATA_PROFILE=$(jq -r .data_profile "$PLAN")" -e RESULT_NAME=greptimedb-1 \
         -e "ENDPOINT=http://127.0.0.1:4000/v1/sql?db=public&format=greptimedb_v1" \
@@ -112,23 +149,34 @@ cell() {
     load_seconds=$(awk -v a="$started" -v b="$ended" 'BEGIN {printf "%.3f", b - a}')
     jq -e --argjson rows "$(jq .rows "$PLAN")" '.count_before == 0 and .count_after == $rows' \
         "$dir/greptimedb-1/load-result.json" >/dev/null
-    curl --noproxy '*' -fsS --max-time 30 http://127.0.0.1:4000/metrics > "$dir/metrics.all"
-    grep -E '^(# (HELP|TYPE) )?greptime_logstore_object_store_wal_' "$dir/metrics.all" > "$dir/metrics.prom" || true
-    rm "$dir/metrics.all"
+}
+
+scrub() {
+    : "${RESULTS:?}"
+    [[ -n "${AWS_ACCESS_KEY_ID:-}" && -n "${AWS_SECRET_ACCESS_KEY:-}" && -d "$RESULTS" ]] || return 0
+    if grep -rqF -e "$AWS_ACCESS_KEY_ID" -e "$AWS_SECRET_ACCESS_KEY" "$RESULTS"; then
+        echo '::error::A credential value appears in the results; removing them instead of uploading' >&2
+        rm -rf "$RESULTS"
+        return 1
+    fi
 }
 
 case "${1:-}" in
+    check-runtime) check_runtime "${2:?}" ;;
+    pull) pull_images ;;
     wal-disk) wal_disk "${2:?}" ;;
     generate) generate ;;
     cell) cell "${2:?}" ;;
     cells)
-        mapfile -t specs < <(jq -c '.cells[]' "$PLAN")
+        : "${PLAN:?}"
         failed=0
-        for spec in "${specs[@]}"; do
-            echo "::group::cell $(jq -r .id <<< "$spec")"
-            bash "${BASH_SOURCE[0]}" cell "$spec" || { echo "::error::cell $(jq -r .id <<< "$spec") failed"; failed=1; }
+        while IFS= read -r spec; do
+            id=$(jq -r .id <<< "$spec")
+            echo "::group::cell $id"
+            bash "${BASH_SOURCE[0]}" cell "$spec" < /dev/null || { echo "::error::cell $id failed"; failed=1; }
             echo '::endgroup::'
-        done
+        done < <(jq -c '.cells[]' "$PLAN")
         exit "$failed" ;;
-    *) echo "usage: $0 wal-disk DIR | generate | cells" >&2; exit 2 ;;
+    scrub) scrub ;;
+    *) echo "usage: $0 check-runtime IMAGE | pull | wal-disk DIR | generate | cells | scrub" >&2; exit 2 ;;
 esac
diff --git a/.github/scripts/oswal-s3-bench-summary.py b/.github/scripts/oswal-s3-bench-summary.py
index 127b9288c02..ed1c4197017 100644
--- a/.github/scripts/oswal-s3-bench-summary.py
+++ b/.github/scripts/oswal-s3-bench-summary.py
@@ -87,6 +87,14 @@ def load_cell(path: Path) -> dict:
     return row
 
 
+def missing_cell(spec: dict) -> dict:
+    """A planned cell that never recorded a result, so no ratio can use its round."""
+    row = {key: None for key in ("exit_code", "wal_device", "version", "load_wall_seconds", "rows",
+                                 "load_seconds", "rows_per_second", "append_ack_p50_seconds",
+                                 "append_ack_p99_seconds", "creates_per_minute", *COUNTERS)}
+    return row | {key: spec[key] for key in ("id", "round", "concurrency", "config")} | {"missing": True}
+
+
 def ratios(rows: list) -> list:
     """D/R and E/R per round and their geometric mean over the rounds, per concurrency."""
     speed = {(r["round"], r["concurrency"], r["config"]): r["rows_per_second"] for r in rows}
@@ -110,18 +118,21 @@ def fmt(value, digits=0, scale=1.0) -> str:
 
 def render(run: dict, rows: list, pairs: list) -> str:
     lines = [f"## Object store WAL against S3 ({'smoke, MinIO' if run.get('smoke') else run.get('instance_type')})", ""]
-    lines += [f"GreptimeDB `{run.get('greptimedb_image')}` ({rows[0]['version'] if rows else '—'}); "
+    lines += [f"GreptimeDB `{run.get('greptimedb_image')}` ({next((r['version'] for r in rows if r['version']), '—')}); "
               f"{fmt(run.get('rows'))} rows in {run.get('batch_rows')}-row batches; bucket region {run.get('region')}.", ""]
     lines += ["| Cell | Rows/s | Load s | Wall s | Rows | Ack p50 ms | Ack p99 ms | Creates | Creates/min "
               "| Stalled | Poisoned | Create failures | Conflicts | WAL on |",
               "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |"]
     for r in rows:
-        cell = r["id"] if r["exit_code"] == 0 else f"{r['id']} (failed, exit {r['exit_code']})"
+        if r.get("missing"):
+            cell = f"{r['id']} (not run)"
+        else:
+            cell = r["id"] if r["exit_code"] == 0 else f"{r['id']} (failed, exit {r['exit_code']})"
         lines.append(" | ".join([
             f"| {cell}", fmt(r["rows_per_second"]), fmt(r["load_seconds"], 1), fmt(r["load_wall_seconds"], 1),
             fmt(r["rows"]), fmt(r["append_ack_p50_seconds"], 1, 1000), fmt(r["append_ack_p99_seconds"], 1, 1000),
             *(fmt(r[c]) for c in COUNTERS[:1]), fmt(r["creates_per_minute"], 1),
-            *(fmt(r[c]) for c in COUNTERS[1:]), f"`{r['wal_device']}` |"]))
+            *(fmt(r[c]) for c in COUNTERS[1:]), f"`{r['wal_device'] or '—'}` |"]))
     lines += ["", "| Concurrency | Ratio | Per round | Geometric mean |", "| ---: | --- | --- | ---: |"]
     for p in pairs:
         lines.append(f"| {p['concurrency']} | {p['ratio']} | {', '.join(fmt(v, 3) for v in p['per_round'])} "
@@ -137,11 +148,16 @@ def main() -> None:
     run_path = args.root / "run.json"
     run = json.loads(run_path.read_text()) if run_path.exists() else {}
     plan_path = args.root / "plan.json"
-    order = [cell["id"] for cell in json.loads(plan_path.read_text())["cells"]] if plan_path.exists() else []
+    planned = json.loads(plan_path.read_text())["cells"] if plan_path.exists() else []
     cells = {path.name: load_cell(path) for path in (args.root / "cells").glob("*") if (path / "cell.json").exists()}
-    rows = [cells[i] for i in order if i in cells] + [cells[i] for i in sorted(cells) if i not in order]
+    # The plan, not what happened to run, defines the expected cells and rounds.
+    rows = [cells.get(spec["id"]) or missing_cell(spec) for spec in planned]
+    rows += [cells[i] for i in sorted(cells) if i not in {spec["id"] for spec in planned}]
     pairs = ratios(rows)
-    (args.root / "summary.json").write_text(json.dumps({"run": run, "cells": rows, "ratios": pairs}, indent=2) + "\n")
+    missing = [r["id"] for r in rows if r.get("missing")]
+    (args.root / "summary.json").write_text(json.dumps(
+        {"run": run, "planned_cells": len(planned), "missing_cells": missing, "cells": rows, "ratios": pairs},
+        indent=2) + "\n")
     print(render(run, rows, pairs) if rows else "No benchmark cell completed.\n")
 
 
diff --git a/.github/scripts/oswal-s3-bench.py b/.github/scripts/oswal-s3-bench.py
index d1201ad792c..e648eb6b8dd 100644
--- a/.github/scripts/oswal-s3-bench.py
+++ b/.github/scripts/oswal-s3-bench.py
@@ -18,10 +18,10 @@
 # dependencies = ["boto3>=1.35,<2"]
 # ///
 
-"""Plan, image tag and bucket helpers of the object store WAL S3 benchmark.
+"""Plan, image tag, run manifest and bucket helpers of the object store WAL S3 benchmark.
 
-`plan` and `tag` need only the standard library; `probe` and `cleanup` need
-boto3 and run under `uv run`.
+`plan`, `tag` and `manifest` need only the standard library; `probe` and
+`cleanup` need boto3 and run under `uv run`.
 """
 
 from __future__ import annotations
@@ -33,6 +33,7 @@ import re
 import sys
 import time
 import urllib.request
+from pathlib import Path
 
 SMOKE_ROWS = 50_000
 # The o11ybench data profiles; a matching row count generates the canonical dataset.
@@ -46,10 +47,11 @@ WAL_CONFIGS = {
 ORDERS = (("R", "D", "E"), ("E", "D", "R"))
 NIGHTLY = re.compile(r"v\d+\.\d+\.\d+(?:-[0-9A-Za-z.]+)?-nightly-(\d{8})")
 TAG = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}")
+DIGEST = re.compile(r".+@sha256:[0-9a-f]{64}")
 
 
-def append_output(name: str, value: str) -> None:
-    if path := os.environ.get("GITHUB_OUTPUT"):
+def append_github(variable: str, name: str, value: str) -> None:
+    if path := os.environ.get(variable):
         with open(path, "a") as output:
             output.write(f"{name}={value}\n")
 
@@ -60,6 +62,30 @@ def positive(name: str, value: str, upper: int) -> int:
     return int(value)
 
 
+def validate_budgets(timeout_minutes: str, ttl_hours: str) -> None:
+    timeout, ttl = positive("timeout", timeout_minutes, 360), positive("ttl", ttl_hours, 168)
+    # Provisioning, teardown and queueing need two hours beyond the benchmark job.
+    if ttl * 60 < timeout + 120:
+        raise ValueError(f"janitor TTL {ttl}h must cover the {timeout}-minute timeout plus 2 hours")
+
+
+def check_regions(bucket: str, runner: str) -> None:
+    if not bucket or bucket != runner:
+        raise ValueError(f"bucket region {bucket!r} differs from runner region {runner!r}; "
+                         "a cross-region bucket measures the WAN, not S3")
+
+
+def build_manifest(plan: dict, greptimedb_inspect: list, runtime_image: str,
+                   instance_type: str, region: str, prefix: str) -> dict:
+    image = next((d for d in greptimedb_inspect[0].get("RepoDigests") or [] if DIGEST.fullmatch(d)), None)
+    if image is None:
+        raise ValueError("the GreptimeDB image has no sha256 repository digest")
+    return {"smoke": plan["smoke"], "rows": plan["rows"], "rounds": plan["rounds"],
+            "concurrency": plan["concurrency"], "batch_rows": plan["batch_rows"],
+            "greptimedb_image": image, "runtime_image": runtime_image,
+            "instance_type": instance_type, "region": region, "prefix": prefix}
+
+
 def build_plan(concurrency: str, rows: str, batch_rows: str, rounds: str, smoke: bool) -> dict:
     levels = [positive("concurrency", level.strip(), 256) for level in concurrency.split(",")]
     if len(set(levels)) != len(levels):
@@ -149,8 +175,16 @@ def main() -> int:
     for name in ("concurrency", "rows", "batch-rows", "rounds"):
         plan.add_argument(f"--{name}", required=True)
     plan.add_argument("--smoke", choices=("true", "false"), required=True)
+    plan.add_argument("--timeout-minutes")
+    plan.add_argument("--ttl-hours")
     plan.add_argument("--output")
     commands.add_parser("tag").add_argument("requested")
+    regions = commands.add_parser("regions")
+    regions.add_argument("--bucket", required=True)
+    regions.add_argument("--runner", required=True)
+    manifest = commands.add_parser("manifest")
+    for name in ("results", "runtime-image", "instance-type", "region", "prefix"):
+        manifest.add_argument(f"--{name}", required=True)
     for name in ("probe", "cleanup"):
         bucket = commands.add_parser(name)
         bucket.add_argument("--bucket", required=True)
@@ -159,6 +193,8 @@ def main() -> int:
         bucket.add_argument("--prefix", required=True)
     args = parser.parse_args()
     if args.command == "plan":
+        if args.timeout_minutes or args.ttl_hours:
+            validate_budgets(args.timeout_minutes or "", args.ttl_hours or "")
         result = build_plan(args.concurrency, args.rows, args.batch_rows, args.rounds, args.smoke == "true")
         if args.output:
             with open(args.output, "w") as output:
@@ -166,8 +202,19 @@ def main() -> int:
         print(f"{len(result['cells'])} cells, {result['rows']} rows, profile {result['data_profile']}")
     elif args.command == "tag":
         tag = resolve_tag(args.requested)
-        append_output("greptimedb_tag", tag)
+        append_github("GITHUB_OUTPUT", "greptimedb_tag", tag)
         print(f"GreptimeDB image tag: {tag}")
+    elif args.command == "regions":
+        check_regions(args.bucket, args.runner)
+        print(f"bucket and runner both in {args.runner}")
+    elif args.command == "manifest":
+        results = Path(args.results)
+        manifest = build_manifest(json.loads((results / "plan.json").read_text()),
+                                  json.loads((results / "images" / "greptimedb.json").read_text()),
+                                  args.runtime_image, args.instance_type, args.region, args.prefix)
+        (results / "run.json").write_text(json.dumps(manifest, indent=2) + "\n")
+        append_github("GITHUB_ENV", "GREPTIMEDB_IMAGE", manifest["greptimedb_image"])
+        print(f"GreptimeDB image: {manifest['greptimedb_image']}")
     else:
         if not re.fullmatch(r"oswal-bench/[0-9]+-[0-9]+", args.prefix):
             raise ValueError(f"refusing a prefix outside oswal-bench/<run>: {args.prefix!r}")
@@ -176,4 +223,8 @@ def main() -> int:
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    try:
+        sys.exit(main())
+    except ValueError as error:
+        print(f"::error::{error}", file=sys.stderr)
+        sys.exit(1)
diff --git a/.github/workflows/oswal-s3-bench.yml b/.github/workflows/oswal-s3-bench.yml
index 7c29432861f..e5395859d8a 100644
--- a/.github/workflows/oswal-s3-bench.yml
+++ b/.github/workflows/oswal-s3-bench.yml
@@ -82,11 +82,12 @@ jobs:
       run: |
         python3 greptimedb/tests/perf/test_oswal_s3_bench_scripts.py
         python3 greptimedb/tests/perf/test_ci_runner_scripts.py
-    - name: Validate plan and image tag
+    - name: Validate plan, budgets and image tag
       id: plan
       run: |
         python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" \
-          --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS"
+          --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS" \
+          --timeout-minutes "$BENCHMARK_TIMEOUT_MINUTES" --ttl-hours "$JANITOR_TTL_HOURS"
         python3 greptimedb/.github/scripts/oswal-s3-bench.py tag "$GREPTIMEDB_TAG"
     - name: Resolve CI runner
       id: runner
@@ -95,27 +96,10 @@ jobs:
       env:
         BUCKET_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
         RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }}
-      run: |
-        [[ -n "$BUCKET_REGION" && "$BUCKET_REGION" == "$RUNNER_REGION" ]] || {
-          echo "::error::Bucket region '$BUCKET_REGION' differs from runner region '$RUNNER_REGION'; a cross-region bucket measures the WAN, not S3"
-          exit 1
-        }
-    - name: Validate budgets and runtime image
+      run: python3 greptimedb/.github/scripts/oswal-s3-bench.py regions --bucket "$BUCKET_REGION" --runner "$RUNNER_REGION"
+    - name: Check the runtime image
       id: runtime
-      run: |
-        set -euo pipefail
-        for value in "$BENCHMARK_TIMEOUT_MINUTES" "$JANITOR_TTL_HOURS"; do
-          [[ "$value" =~ ^[1-9][0-9]{0,3}$ ]] || { echo 'Budgets must be positive integers'; exit 1; }
-        done
-        ((BENCHMARK_TIMEOUT_MINUTES <= 360 && JANITOR_TTL_HOURS <= 168))
-        ((JANITOR_TTL_HOURS * 60 >= BENCHMARK_TIMEOUT_MINUTES + 120)) || { echo 'Janitor TTL must cover the timeout plus 2 hours'; exit 1; }
-        [[ "$RUNTIME_IMAGE" == public.ecr.aws/* ]]
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "$RUNTIME_IMAGE"
-        resolved=$(docker image inspect "$RUNTIME_IMAGE" | jq -er '.[0].RepoDigests[0]')
-        docker run --rm --entrypoint sh "$resolved" -c \
-          'test -f scripts/agent-observability/greptimedb.sh && test -f scripts/generate_agent_observations.py' \
-          || { echo 'Runtime image lacks the agent observability loader or generator'; exit 1; }
-        printf 'runtime_image=%s\n' "$resolved" >> "$GITHUB_OUTPUT"
+      run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh check-runtime "$RUNTIME_IMAGE"
   provision:
     needs: validate
     if: ${{ !inputs.smoke }}
@@ -185,20 +169,11 @@ jobs:
         docker compose -f greptimedb/tests-integration/fixtures/docker-compose.yml up -d --wait minio
     - name: Pull images and write the plan
       run: |
-        set -euo pipefail
-        mkdir -p "$RESULTS/images"
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "$RUNTIME_IMAGE"
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "greptime/greptimedb:$GREPTIMEDB_TAG"
-        docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json"
-        docker image inspect "greptime/greptimedb:$GREPTIMEDB_TAG" > "$RESULTS/images/greptimedb.json"
-        image=$(jq -er '.[0].RepoDigests[0] | select(test("@sha256:[a-f0-9]{64}$"))' "$RESULTS/images/greptimedb.json")
-        printf 'GREPTIMEDB_IMAGE=%s\n' "$image" >> "$GITHUB_ENV"
+        bash greptimedb/.github/scripts/oswal-s3-bench-run.sh pull
         python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" --output "$PLAN" \
           --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS"
-        jq -n --slurpfile plan "$PLAN" --arg image "$image" --arg runtime "$RUNTIME_IMAGE" \
-          --arg instance_type "$INSTANCE_TYPE" --arg region "$S3_REGION" --arg prefix "$RUN_PREFIX" \
-          '{smoke: $plan[0].smoke, rows: $plan[0].rows, batch_rows: $plan[0].batch_rows, greptimedb_image: $image,
-            runtime_image: $runtime, instance_type: $instance_type, region: $region, prefix: $prefix}' > "$RESULTS/run.json"
+        python3 greptimedb/.github/scripts/oswal-s3-bench.py manifest --results "$RESULTS" \
+          --runtime-image "$RUNTIME_IMAGE" --instance-type "$INSTANCE_TYPE" --region "$S3_REGION" --prefix "$RUN_PREFIX"
     - name: Probe the bucket
       env:
         AWS_ACCESS_KEY_ID: ${{ inputs.smoke && 'superpower_ci_user' || secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
@@ -231,14 +206,7 @@ jobs:
       env:
         AWS_ACCESS_KEY_ID: ${{ inputs.smoke && 'superpower_ci_user' || secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
         AWS_SECRET_ACCESS_KEY: ${{ inputs.smoke && 'superpower_password' || secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
-      run: |
-        mkdir -p "$RESULTS"
-        [[ -n "$AWS_ACCESS_KEY_ID" && -n "$AWS_SECRET_ACCESS_KEY" ]] || exit 0
-        if grep -rqF -e "$AWS_ACCESS_KEY_ID" -e "$AWS_SECRET_ACCESS_KEY" "$RESULTS"; then
-          echo '::error::A credential value appears in the results; not uploading them'
-          rm -rf "$RESULTS"
-          exit 1
-        fi
+      run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh scrub
     - name: Upload results (no corpus)
       if: always()
       uses: actions/upload-artifact@v4
diff --git a/tests/perf/test_oswal_s3_bench_scripts.py b/tests/perf/test_oswal_s3_bench_scripts.py
index 604f281aa39..d9503ae24a8 100644
--- a/tests/perf/test_oswal_s3_bench_scripts.py
+++ b/tests/perf/test_oswal_s3_bench_scripts.py
@@ -17,6 +17,7 @@
 
 import importlib.util
 import json
+import os
 import subprocess
 import sys
 import tempfile
@@ -86,8 +87,25 @@ class PlanTest(unittest.TestCase):
         with self.assertRaises(ValueError):
             bench.resolve_tag("bad tag")
 
-    def test_run_script_parses(self):
-        subprocess.run(["bash", "-n", str(SCRIPTS / "oswal-s3-bench-run.sh")], check=True)
+    def test_budgets_and_regions(self):
+        bench.validate_budgets("240", "8")
+        for timeout, ttl in (("360", "7"), ("0", "8"), ("361", "168"), ("240", "169")):
+            with self.subTest(timeout=timeout, ttl=ttl), self.assertRaises(ValueError):
+                bench.validate_budgets(timeout, ttl)
+        bench.check_regions("ap-southeast-1", "ap-southeast-1")
+        for bucket, runner in (("us-east-1", "ap-southeast-1"), ("", "")):
+            with self.subTest(bucket=bucket), self.assertRaisesRegex(ValueError, "cross-region"):
+                bench.check_regions(bucket, runner)
+
+    def test_manifest_takes_the_sha256_digest(self):
+        plan = bench.build_plan("1,8", "2000000", "1000", "2", smoke=False)
+        inspect = [{"RepoDigests": ["greptime/greptimedb@sha256:" + "a" * 64]}]
+        manifest = bench.build_manifest(plan, inspect, "runtime@sha256:" + "b" * 64, "c7id.2xlarge",
+                                        "ap-southeast-1", "oswal-bench/1-1")
+        self.assertEqual(manifest["greptimedb_image"], "greptime/greptimedb@sha256:" + "a" * 64)
+        self.assertEqual((manifest["rows"], manifest["rounds"], manifest["concurrency"]), (2_000_000, 2, [1, 8]))
+        with self.assertRaises(ValueError):
+            bench.build_manifest(plan, [{"RepoDigests": []}], "", "", "", "")
 
 
 class SummaryTest(unittest.TestCase):
@@ -138,6 +156,23 @@ class SummaryTest(unittest.TestCase):
             self.assertEqual(d_cell["stalled_appends_total"], 3)
             self.assertIn("| 1-8-D | 80 | 60.0 | 61.0 | 6,000 | 128.0 | 253.4 | 120 | 120.0 | 3 | 0 | — | — |", output)
 
+    def test_a_planned_round_that_never_ran_leaves_the_ratio_undefined(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plan = bench.build_plan("8", "6000", "1000", "2", smoke=False)
+            (root / "plan.json").write_text(json.dumps(plan))
+            for cell_id, speed in {"1-8-R": 100.0, "1-8-D": 80.0, "1-8-E": 90.0}.items():
+                self.write_cell(root, cell_id, speed)
+            output = subprocess.run([sys.executable, str(SCRIPTS / "oswal-s3-bench-summary.py"), "--root", str(root)],
+                                    check=True, capture_output=True, text=True).stdout
+            result = json.loads((root / "summary.json").read_text())
+            self.assertEqual(result["missing_cells"], ["2-8-E", "2-8-D", "2-8-R"])
+            self.assertEqual(result["planned_cells"], 6)
+            for ratio in result["ratios"]:
+                self.assertIsNone(ratio["geomean"])
+                self.assertIsNone(ratio["per_round"][1])
+            self.assertIn("2-8-D (not run)", output)
+
     def test_a_failed_cell_leaves_its_ratio_undefined(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
@@ -149,5 +184,98 @@ class SummaryTest(unittest.TestCase):
             self.assertIn("1-1-D (failed, exit 1)", summary.render({}, rows, summary.ratios(rows)))
 
 
+FAKE_DOCKER = r"""#!/usr/bin/env bash
+printf '%s\n' "docker $*" >> "$FAKE_LOG"
+case "$1" in
+    run)
+        for arg in "$@"; do [[ "$arg" != -d ]] || { echo "keys=${GREPTIMEDB_STANDALONE__STORAGE__ACCESS_KEY_ID:+set}${GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY:+set}" >> "$FAKE_LOG"; echo fakecid; exit 0; }; done
+        for arg in "$@"; do
+            case "$arg" in type=bind,src=*,dst=/results) results=${arg#type=bind,src=}; results=${results%,dst=/results} ;; esac
+        done
+        [[ "$FAKE_LOAD_RC" == 0 ]] || exit "$FAKE_LOAD_RC"
+        mkdir -p "$results/greptimedb-1"
+        echo "{\"count_before\": 0, \"count_after\": $FAKE_ROWS, \"load_seconds\": 2, \"rows_per_second\": 3000}" \
+            > "$results/greptimedb-1/load-result.json" ;;
+    inspect) echo true ;;
+    exec) echo 'version: 1.4.0-test' ;;
+    logs) echo 'server log' ;;
+esac
+"""
+FAKE_CURL = r"""#!/usr/bin/env bash
+printf '%s\n' "curl $*" >> "$FAKE_LOG"
+for arg in "$@"; do
+    [[ "$arg" != */metrics ]] || { printf '%s\n' 'greptime_logstore_object_store_wal_poisoned_total 1' 'other_total 2'; exit 0; }
+done
+"""
+
+
+class CellLifecycleTest(unittest.TestCase):
+    """Drives the run script against fake docker and curl binaries."""
+
+    def run_cells(self, root, load_rc):
+        bin_dir = root / "bin"
+        bin_dir.mkdir()
+        for name, body in (("docker", FAKE_DOCKER), ("curl", FAKE_CURL)):
+            (bin_dir / name).write_text(body)
+            (bin_dir / name).chmod(0o755)
+        plan = bench.build_plan("8", "6000", "1000", "1", smoke=False)
+        plan["cells"] = [cell for cell in plan["cells"] if cell["config"] != "R"]
+        (root / "results").mkdir()
+        (root / "results" / "plan.json").write_text(json.dumps(plan))
+        env = os.environ | {
+            "PATH": f"{bin_dir}:{os.environ['PATH']}", "FAKE_LOG": str(root / "fake.log"),
+            "FAKE_LOAD_RC": str(load_rc), "FAKE_ROWS": "6000", "PLAN": str(root / "results" / "plan.json"),
+            "RESULTS": str(root / "results"), "CORPUS_DIR": str(root / "corpus"), "OWNER": "o11ybench-oswal-1-1",
+            "RUNTIME_IMAGE": "runtime@sha256:" + "b" * 64, "GREPTIMEDB_IMAGE": "greptimedb@sha256:" + "a" * 64,
+            "RUN_PREFIX": "oswal-bench/1-1", "WAL_ROOT": str(root / "wal"), "S3_BUCKET": "bucket",
+            "S3_REGION": "ap-southeast-1", "AWS_ACCESS_KEY_ID": "AKIAFAKEKEY", "AWS_SECRET_ACCESS_KEY": "fake-secret-value",
+        }
+        script = str(SCRIPTS / "oswal-s3-bench-run.sh")
+        cells = subprocess.run(["bash", script, "cells"], env=env, capture_output=True, text=True, timeout=60)
+        scrub = subprocess.run(["bash", script, "scrub"], env=env, capture_output=True, text=True, timeout=60)
+        return cells, scrub, (root / "fake.log").read_text(), env
+
+    def test_a_failed_load_still_scrapes_metrics_and_removes_the_server(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            cells, scrub, log, _ = self.run_cells(root, load_rc=3)
+            self.assertNotEqual(cells.returncode, 0)
+            self.assertEqual(scrub.returncode, 0, scrub.stderr)
+            for cell_id in ("1-8-D", "1-8-E"):
+                cell = root / "results" / "cells" / cell_id
+                self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 3)
+                self.assertEqual((cell / "metrics.prom").read_text(),
+                                 "greptime_logstore_object_store_wal_poisoned_total 1\n")
+                self.assertFalse((cell / "metrics.all").exists())
+                self.assertEqual((cell / "server.log").read_text(), "server log\n")
+            self.assertEqual(log.count("docker stop -t 120 fakecid"), 2)
+            self.assertEqual(log.count("docker rm -fv fakecid"), 2)
+
+    def test_a_cell_passes_the_keys_by_name_and_records_a_redacted_config(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            cells, scrub, log, env = self.run_cells(root, load_rc=0)
+            self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr)
+            self.assertEqual(scrub.returncode, 0, scrub.stderr)
+            self.assertNotIn("fake-secret-value", log)
+            self.assertNotIn("AKIAFAKEKEY", log)
+            self.assertEqual(log.count("keys=setset"), 2)
+            self.assertIn("-e GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY ", log)
+            cell = root / "results" / "cells" / "1-8-D"
+            config = (cell / "config.env").read_text().splitlines()
+            self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__ROOT=oswal-bench/1-1/1-8-D", config)
+            self.assertIn("GREPTIMEDB_STANDALONE__WAL__ACK_MODE=durable", config)
+            self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY=<redacted>", config)
+            recorded = json.loads((cell / "cell.json").read_text())
+            self.assertEqual((recorded["exit_code"], recorded["version"]), (0, "1.4.0-test"))
+            self.assertEqual(recorded["wal_device"], "s3://bucket/oswal-bench/1-1/1-8-D/wal")
+            # A leaked value makes scrub refuse the results.
+            (cell / "leak.txt").write_text("fake-secret-value")
+            leaked = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "scrub"], env=env,
+                                    capture_output=True, text=True, timeout=60)
+            self.assertNotEqual(leaked.returncode, 0)
+            self.assertFalse((root / "results").exists())
+
+
 if __name__ == "__main__":
     unittest.main()