diff --git a/.github/scripts/oswal-s3-bench-run.sh b/.github/scripts/oswal-s3-bench-run.sh
index f822a4a9b6f..d158ab14d60 100644
--- a/.github/scripts/oswal-s3-bench-run.sh
+++ b/.github/scripts/oswal-s3-bench-run.sh
@@ -10,24 +10,64 @@ runtime() {
         --security-opt no-new-privileges --cap-drop ALL "$@"
 }
 
+# Pin the runtime image to a digest without pulling it: ECR Public's anonymous data limit can refuse a
+# hosted runner even a manifest read. A reference that names its digest is pinned as given; only a tag
+# is resolved from the registry's manifest. The benchmark host pulls the image and checks the contents.
 check_runtime() {
-    local image=$1 resolved
+    local image=$1 repository digest attempt component='[a-z0-9]+(([._]|__|-+)[a-z0-9]+)*'
     [[ "$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}"
+    # Docker's reference grammar, checked here because a named digest never reaches the registry.
+    local grammar="^public\.ecr\.aws(/$component)+(:[A-Za-z0-9_][A-Za-z0-9_.-]{0,127})?(@sha256:[0-9a-f]{64})?\$"
+    [[ "$image" =~ $grammar ]] || { echo "$image is not a full ECR Public image reference" >&2; return 1; }
+    if [[ "$image" == *@* ]]; then
+        digest=${image##*@}
+    else
+        for attempt in 1 2 3; do
+            digest=$(docker buildx imagetools inspect --format '{{json .Manifest}}' "$image" | jq -er .digest) && break
+            ((attempt < 3)) || { echo "Cannot read the manifest of $image" >&2; return 1; }
+            sleep $((attempt * 10))
+        done
+    fi
+    [[ "$digest" =~ ^sha256:[0-9a-f]{64}$ ]] || { echo "$image has no sha256 digest" >&2; return 1; }
+    repository=${image%@*}
+    [[ "${repository##*/}" != *:* ]] || repository=${repository%:*}
+    printf 'runtime_image=%s@%s\n' "$repository" "$digest" >> "${GITHUB_OUTPUT:-/dev/stdout}"
+}
+
+# Pull every GreptimeDB image of the run, pin it to its digest and record the version and
+# commit it reports, before any instance is allocated.
+check_images() {
+    local entry image resolved info version commit checked=()
+    while IFS= read -r entry; do
+        image=$(jq -r .image <<< "$entry")
+        bash "$here/pull-benchmark-image.sh" "$image"
+        resolved=$(docker image inspect "$image" | jq -er '[.[0].RepoDigests[]? | select(test("@sha256:[0-9a-f]{64}$"))][0]') \
+            || { echo "$image has no sha256 repository digest" >&2; return 1; }
+        info=$(docker run --rm --entrypoint greptime "$resolved" --version)
+        version=$(sed -n 's/^version: //p' <<< "$info") commit=$(sed -n 's/^commit: //p' <<< "$info")
+        [[ -n "$version" && -n "$commit" ]] || { echo "$image reports no version or commit" >&2; return 1; }
+        echo "${image}: $resolved, version $version, commit $commit"
+        checked+=("$(jq -c --arg image "$resolved" --arg version "$version" --arg commit "$commit" \
+            '{label, image: $image, requested: .image, version: $version, commit: $commit}' <<< "$entry")")
+    done < <(jq -c '.[]' <<< "$1")
+    ((${#checked[@]})) || { echo 'No GreptimeDB image to check' >&2; return 1; }
+    printf 'greptimedb_images=%s\n' "$(jq -sc . <<< "${checked[*]}")" >> "${GITHUB_OUTPUT:-/dev/stdout}"
 }
 
 pull_images() {
-    : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_REF:?}"
+    : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_IMAGES:?}"
+    local image images=()
     mkdir -p "$RESULTS/images"
     bash "$here/pull-benchmark-image.sh" "$RUNTIME_IMAGE"
-    bash "$here/pull-benchmark-image.sh" "$GREPTIMEDB_REF"
+    docker run --rm --entrypoint sh "$RUNTIME_IMAGE" -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; }
+    while IFS= read -r image; do
+        bash "$here/pull-benchmark-image.sh" "$image"
+        images+=("$image")
+    done < <(jq -r '.[].image' <<< "$GREPTIMEDB_IMAGES")
     docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json"
-    docker image inspect "$GREPTIMEDB_REF" > "$RESULTS/images/greptimedb.json"
+    docker image inspect "${images[@]}" > "$RESULTS/images/greptimedb.json"
 }
 
 wal_disk() {
@@ -92,6 +132,8 @@ elapsed() { awk -v a="$1" -v b="$(now)" 'BEGIN {printf "%.3f", b - a}'; }
 # CPU seconds the server's cgroup used, and busy and total jiffies of the host.
 server_cpu() { docker exec "$cid" cat /sys/fs/cgroup/cpu.stat 2>/dev/null | awk '/^usage_usec / {printf "%.3f", $2 / 1e6}' || true; }
 host_cpu() { awk '/^cpu / {print $2 + $3 + $4 + $7 + $8 + $9, $2 + $3 + $4 + $5 + $6 + $7 + $8 + $9}' /proc/stat 2>/dev/null || true; }
+# The host's 1, 5 and 15 minute load averages as a JSON list.
+host_load() { awk '{printf "[%s, %s, %s]", $1, $2, $3}' "${OSWAL_LOADAVG:-/proc/loadavg}" 2>/dev/null || true; }
 
 # A server's log is named after the phase it started in, whatever phase stops it.
 start_server() {
@@ -294,6 +336,7 @@ finish() {
     [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir"
     [[ -z "$home_dir" ]] || sudo -n rm -rf "$home_dir"
     jq -n --argjson spec "$spec" --arg device "$wal_device" --arg version "$version" \
+        --arg host_load_start "$host_load_start" --arg host_load_end "$(host_load)" \
         --arg load_seconds "$load_seconds" --argjson exit_code "$rc" --argjson scraped "$metrics_scraped" \
         --argjson restart_scraped "$restart_scraped" --arg restart_seconds "$restart_seconds" \
         --arg restart_query_seconds "$restart_query_seconds" --arg restart_count "$restart_count" \
@@ -306,21 +349,26 @@ finish() {
           restart_scraped: $restart_scraped, restart_seconds: ($restart_seconds | tonumber? // null),
           restart_query_seconds: ($restart_query_seconds | tonumber? // null),
           restart_count: ($restart_count | tonumber? // null),
-          indexed_before_kill: ($indexed_before_kill | tonumber? // null)}' > "$dir/cell.json"
+          indexed_before_kill: ($indexed_before_kill | tonumber? // null),
+          host_load_start: ($host_load_start | fromjson? // null), host_load_end: ($host_load_end | fromjson? // 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() {
-    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_IMAGE:?}"
+    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_IMAGES:?}"
     : "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}" "${AWS_ACCESS_KEY_ID:?}" "${AWS_SECRET_ACCESS_KEY:?}"
     spec=$1 cid= wal_dir= home_dir= version= load_seconds= metrics_scraped=false id= phase= server_phase=
     restart_scraped=false restart_seconds= restart_query_seconds= restart_count= server_cpu_seconds= host_cpu_busy=
-    indexed_before_kill=
-    local concurrency setting started cpu_before host_before host_after corpus partitions load script
-    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec")
+    indexed_before_kill= host_load_start=$(host_load)
+    local concurrency setting started cpu_before host_before host_after corpus partitions load script label
+    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec") label=$(jq -r .image <<< "$spec")
     dir="$RESULTS/cells/$id" name="$OWNER-$id"
+    # The digest the image of the cell's label was pinned to before the run.
+    GREPTIMEDB_IMAGE=$(jq -er --arg label "$label" '.[] | select(.label == $label) | .image' <<< "$GREPTIMEDB_IMAGES") \
+        || { echo "$id: no image labelled '$label'" >&2; exit 1; }
     mkdir -p "$dir"
     # 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"
@@ -419,6 +467,7 @@ scrub() {
 
 case "${1:-}" in
     check-runtime) check_runtime "${2:?}" ;;
+    check-images) check_images "${2:?}" ;;
     pull) pull_images ;;
     wal-disk) wal_disk "${2:?}" ;;
     generate) generate ;;
@@ -434,5 +483,5 @@ case "${1:-}" in
         done < <(jq -c '.cells[]' "$PLAN")
         exit "$failed" ;;
     scrub) scrub ;;
-    *) echo "usage: $0 check-runtime IMAGE | pull | wal-disk DIR | generate | cells | scrub" >&2; exit 2 ;;
+    *) echo "usage: $0 check-runtime IMAGE | check-images JSON | 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 305a647a9a4..e693b5f63db 100644
--- a/.github/scripts/oswal-s3-bench-summary.py
+++ b/.github/scripts/oswal-s3-bench-summary.py
@@ -192,6 +192,7 @@ def load_cell(path: Path, spec: dict | None = None) -> dict:
     wall, cpu = cell.get("load_wall_seconds"), cell.get("server_cpu_seconds")
     row = {
         "id": cell["id"], "group": cell.get("group", ""), "mode": cell.get("mode", "load"),
+        "image": cell.get("image", ""), "warmup": cell.get("warmup", False),
         "round": cell["round"], "concurrency": cell["concurrency"],
         "config": cell["config"], "exit_code": cell["exit_code"], "wal_device": cell.get("wal_device"),
         "missing": cell.get("missing", False), "env": cell.get("env", {}),
@@ -200,6 +201,7 @@ def load_cell(path: Path, spec: dict | None = None) -> dict:
         "rows_per_second": result.get("rows_per_second") if cell["exit_code"] == 0 else None,
         "server_cpu_cores": cpu / wall if cpu is not None and wall else None,
         "host_cpu_busy": cell.get("host_cpu_busy"), "host_cpus": cell.get("host_cpus"),
+        "host_load_start": cell.get("host_load_start"), "host_load_end": cell.get("host_load_end"),
     }
     for name, key in (("append_ack_seconds", "append_ack"), ("seal_to_durable_seconds", "seal_to_durable"),
                       ("object_bytes", "object_bytes")):
@@ -264,13 +266,14 @@ def load_cell(path: Path, spec: dict | None = None) -> dict:
 
 
 def ratios(rows: list) -> list:
-    """Each object store WAL cell over each Raft Engine control per round, and their geometric mean over the rounds, per concurrency."""
-    rows = [r for r in rows if r["mode"] == "load"]
-    speed = {(r["group"], r["round"], r["concurrency"], r["config"]): r["rows_per_second"]
+    """Each object store WAL cell over each Raft Engine control of the same image per round, and their
+    geometric mean over the rounds, per concurrency; warm-up cells are left out."""
+    rows = [r for r in rows if r["mode"] == "load" and not r["warmup"]]
+    speed = {(r["group"], r["image"], r["round"], r["concurrency"], r["config"]): r["rows_per_second"]
              for r in rows}
     result = []
-    for group in dict.fromkeys(r["group"] for r in rows):
-        members = [r for r in rows if r["group"] == group]
+    for group, image in dict.fromkeys((r["group"], r["image"]) for r in rows):
+        members = [r for r in rows if (r["group"], r["image"]) == (group, image)]
         series = sorted({r["config"] for r in members if r["config"] not in BASELINES})
         baselines = [b for b in BASELINES if any(r["config"] == b for r in members)]
         for concurrency in sorted({r["concurrency"] for r in members}):
@@ -278,16 +281,42 @@ def ratios(rows: list) -> list:
             for config, baseline in ((s, b) for s in series for b in baselines):
                 per_round = []
                 for round_number in rounds:
-                    base = speed.get((group, round_number, concurrency, baseline))
-                    measured = speed.get((group, round_number, concurrency, config))
+                    base = speed.get((group, image, round_number, concurrency, baseline))
+                    measured = speed.get((group, image, round_number, concurrency, config))
                     per_round.append(measured / base if base and measured else None)
                 complete = [ratio for ratio in per_round if ratio is not None]
-                result.append({"group": group, "concurrency": concurrency, "ratio": f"{config}/{baseline}",
-                               "per_round": per_round,
+                result.append({"group": group, "image": image, "concurrency": concurrency,
+                               "ratio": f"{config}/{baseline}", "per_round": per_round,
                                "geomean": geomean(complete) if len(complete) == len(per_round) else None})
     return result
 
 
+def image_ratios(rows: list, images: list) -> list:
+    """Every load cell of every image per round, and each image's rows/s over the first image's in the
+    same round, with their geometric mean, minimum and maximum over the rounds; warm-up cells are left out."""
+    rows = [r for r in rows if r["mode"] == "load" and not r["warmup"]]
+    cells = {(r["group"], r["image"], r["round"], r["concurrency"], r["config"]): r for r in rows}
+    result = []
+    for group, concurrency, config in dict.fromkeys((r["group"], r["concurrency"], r["config"]) for r in rows):
+        rounds = sorted({r["round"] for r in rows if (r["group"], r["concurrency"], r["config"]) == (group, concurrency, config)})
+        for image in images:
+            measured = [cells.get((group, image, n, concurrency, config)) or {} for n in rounds]
+            reference = [cells.get((group, images[0], n, concurrency, config)) or {} for n in rounds]
+            per_round = [ratio(m.get("rows_per_second"), r.get("rows_per_second")) for m, r in zip(measured, reference)]
+            complete = image != images[0] and None not in per_round
+            result.append({
+                "group": group, "concurrency": concurrency, "config": config, "image": image,
+                "rounds": [{"round": n, "id": m.get("id"), "rows_per_second": m.get("rows_per_second"),
+                            "client_p50": m.get("client_p50"), "client_p99": m.get("client_p99"),
+                            "host_load_start": m.get("host_load_start"), "host_load_end": m.get("host_load_end")}
+                           for n, m in zip(rounds, measured)],
+                "ratio_to": None if image == images[0] else images[0],
+                "per_round": None if image == images[0] else per_round,
+                "geomean": geomean(per_round) if complete else None,
+                "min": min(per_round) if complete else None, "max": max(per_round) if complete else None})
+    return result
+
+
 def operations(counts: dict) -> str:
     return ", ".join(f"{op} {n:,.0f}" for op, n in sorted(counts.items())) or "—"
 
@@ -304,18 +333,31 @@ def fmt(value, digits=0, scale=1.0) -> str:
     return "—" if value is None else f"{value * scale:,.{digits}f}"
 
 
-def render(run: dict, rows: list, pairs: list) -> str:
+def load_average(value) -> str:
+    """The 1-minute load average of a recorded [1, 5, 15] minute triple."""
+    return "—" if not value else f"{value[0]:.2f}"
+
+
+def render(run: dict, rows: list, pairs: list, compared: 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')}` ({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')}.", ""]
+    images = run.get("greptimedb_images") or []
+    if len(images) > 1:
+        lines += [f"{fmt(run.get('rows'))} rows in {run.get('batch_rows')}-row batches; bucket region {run.get('region')}; "
+                  f"each image is compared with the first, `{images[0]['label']}`.", "",
+                  "| Image | Reference | Version | Commit |", "| --- | --- | --- | --- |"]
+        lines += [f"| {i['label']} | `{i['image']}` | {i['version']} | `{i['commit']}` |" for i in images] + [""]
+    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 | Overrides | Rows/s | Load s | Wall s | Ack p50 / p99 ms | Seal to durable p50 / p99 ms "
               "| Object p50 / p99 MiB | Creates | Creates/min | Stalled | Poisoned | Create failures | Conflicts "
-              "| Server CPUs | Host busy | WAL on |",
+              "| Server CPUs | Host busy | WAL on | Load average at start / end |",
               "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: "
-              "| ---: | ---: | --- |"]
+              "| ---: | ---: | --- | ---: |"]
     def label(r):
-        return f"{r['id']} (not run)" if r["missing"] else r["id"] if r["exit_code"] == 0 \
-            else f"{r['id']} (failed, exit {r['exit_code']})"
+        cell = f"{r['id']} (warm-up)" if r["warmup"] else r["id"]
+        return f"{cell} (not run)" if r["missing"] else cell if r["exit_code"] == 0 \
+            else f"{cell} (failed, exit {r['exit_code']})"
     loads = [r for r in rows if r["mode"] == "load"]
     for r in loads:
         cell = label(r)
@@ -327,7 +369,8 @@ def render(run: dict, rows: list, pairs: list) -> str:
             *(fmt(r[c]) for c in COUNTERS[:1]), fmt(r["creates_per_minute"], 1),
             *(fmt(r[c]) for c in COUNTERS[1:]), fmt(r["server_cpu_cores"], 2),
             "—" if r["host_cpu_busy"] is None else f"{r['host_cpu_busy']:.0%} of {r['host_cpus'] or '?'}",
-            f"`{r['wal_device'] or '—'}` |"]))
+            f"`{r['wal_device'] or '—'}`",
+            f"{load_average(r['host_load_start'])} / {load_average(r['host_load_end'])} |"]))
     lines += ["", "| Cell | WAL MB/s | Objects/s | Mean object MiB | Sealed on size | OpenDAL S3 operations during the load "
               "| OpenDAL S3 operations of the restart | Idle s | Idle creates | Idle OpenDAL S3 operations |",
               "| --- | ---: | ---: | ---: | --- | --- | --- | ---: | ---: | --- |"]
@@ -431,11 +474,22 @@ def render(run: dict, rows: list, pairs: list) -> str:
                 fmt(x["wall_seconds"], 2),
                 fmt(x["query_seconds"], 2), fmt(x["recovery_seconds"], 3), fmt(x["recovered_objects"]),
                 f"{fmt(x['rows'])} |"]))
-    lines += ["", "| Group | Concurrency | Ratio | Per round | Geometric mean |",
-              "| --- | ---: | --- | --- | ---: |"]
+    lines += ["", "| Group | Image | Concurrency | Ratio | Per round | Geometric mean |",
+              "| --- | --- | ---: | --- | --- | ---: |"]
     for p in pairs:
-        lines.append(f"| {p['group'] or '—'} | {p['concurrency']} | {p['ratio']} "
+        lines.append(f"| {p['group'] or '—'} | {p['image'] or '—'} | {p['concurrency']} | {p['ratio']} "
                      f"| {', '.join(fmt(v, 3) for v in p['per_round'])} | {fmt(p['geomean'], 3)} |")
+    if compared:
+        lines += ["", f"| Group | Concurrency | Config | Image | Rows/s per round | Client p50 / p99 ms per round "
+                  f"| Ratio to {compared[0]['image']} per round | Geometric mean | Min | Max |",
+                  "| --- | ---: | --- | --- | --- | --- | --- | ---: | ---: | ---: |"]
+        for c in compared:
+            lines.append(" | ".join([
+                f"| {c['group'] or '—'}", str(c["concurrency"]), c["config"], c["image"],
+                "; ".join(fmt(r["rows_per_second"]) for r in c["rounds"]),
+                "; ".join(pair(r["client_p50"], r["client_p99"], 1, 1000) for r in c["rounds"]),
+                "—" if c["per_round"] is None else ", ".join(fmt(v, 3) for v in c["per_round"]),
+                fmt(c["geomean"], 3), fmt(c["min"], 3), fmt(c["max"], 3)]) + " |")
     return "\n".join(lines) + "\n"
 
 
@@ -447,17 +501,20 @@ 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"
-    planned = json.loads(plan_path.read_text())["cells"] if plan_path.exists() else []
+    plan = json.loads(plan_path.read_text()) if plan_path.exists() else {}
+    planned = plan.get("cells", [])
     # The plan, not what happened to run, defines the expected cells and rounds.
     rows = [load_cell(args.root / "cells" / spec["id"], spec) for spec in planned]
     rows += [load_cell(path) for path in sorted((args.root / "cells").glob("*"))
              if path.name not in {spec["id"] for spec in planned} and (path / "cell.json").exists()]
     pairs = ratios(rows)
+    images = plan.get("images", [""])
+    compared = image_ratios(rows, images) if len(images) > 1 else []
     missing = [r["id"] for r in rows if r["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")
+        {"run": run, "planned_cells": len(planned), "missing_cells": missing, "cells": rows, "ratios": pairs,
+         "image_ratios": compared}, indent=2) + "\n")
+    print(render(run, rows, pairs, compared) if rows else "No benchmark cell completed.\n")
 
 
 if __name__ == "__main__":
diff --git a/.github/scripts/oswal-s3-bench.py b/.github/scripts/oswal-s3-bench.py
index 9d674400fe2..eedd906ffef 100644
--- a/.github/scripts/oswal-s3-bench.py
+++ b/.github/scripts/oswal-s3-bench.py
@@ -24,6 +24,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import math
 import os
 import re
 import sys
@@ -80,6 +81,7 @@ WITH(
   'compaction.twcs.time_window' = '2h'
 )"""
 GROUP_NAME = re.compile(r"[a-z][a-z0-9]{0,11}")
+IMAGE_LABEL = re.compile(r"[a-z][a-z0-9]{0,7}")
 RESERVED_ENV = ("GREPTIMEDB_STANDALONE__STORAGE__", "GREPTIMEDB_STANDALONE__WAL__")
 ENV_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
 # Docker's reference grammar: path components separated by ".", "_", "__" or runs of "-".
@@ -88,6 +90,18 @@ IMAGE = re.compile(r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*(?::[0-9]+)?(?:/[a-z0-
 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}")
+# Rows/s of a load cell in 1000-row batches on c6id.2xlarge at the concurrency levels earlier runs
+# measured, interpolated on a log scale between them; durable writes slow down with several regions.
+# Smaller batches are taken to keep the request rate, so their rows/s shrink with the batch.
+MEASURED_BATCH_ROWS = 1000
+LOAD_RATES = {"durable": [(1, 9_000), (8, 34_000), (32, 55_000)],
+              "durable, partitioned": [(1, 7_000), (8, 26_000), (32, 55_000)],
+              "other": [(1, 28_000), (8, 60_000)]}
+# Starting, scraping and stopping a server, and the restart of a restart cell or a crash cycle.
+CELL_SECONDS, RESTART_SECONDS, CYCLE_SECONDS = 5, 60, 15
+# Checkout, dependencies, image pulls and the bucket probe; corpus generation per million rows; and the
+# deletion of the run's objects one at a time, which took 12-17 % of the cells' time in earlier runs.
+JOB_SECONDS, GENERATE_SECONDS_PER_MILLION, CLEANUP_SHARE = 180, 30, 0.2
 
 
 def append_github(variable: str, name: str, value: str) -> None:
@@ -115,16 +129,18 @@ def check_regions(bucket: str, runner: str) -> None:
                          "a cross-region bucket measures the WAN, not S3")
 
 
-def build_manifest(plan: dict, greptimedb_inspect: list, runtime_image: str,
+def build_manifest(plan: dict, images: 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")
+    """The run's record; `images` are the checked images, each pinned to its digest, the first the reference."""
+    if [image["label"] for image in images] != plan["images"]:
+        raise ValueError(f"the checked images {[i['label'] for i in images]} are not the plan's {plan['images']}")
+    if not all(DIGEST.fullmatch(image["image"]) for image in images):
+        raise ValueError("every GreptimeDB image must be pinned to a sha256 digest")
     return {"smoke": plan["smoke"], "rows": plan["rows"], "rounds": plan["rounds"],
             "concurrency": plan["concurrency"], "batch_rows": plan["batch_rows"],
             "configs": plan["configs"], "extra_env": plan["extra_env"],
-            "restart": plan["restart"], "groups": plan["groups"],
-            "greptimedb_image": image, "runtime_image": runtime_image,
+            "restart": plan["restart"], "groups": plan["groups"], "warmup": plan["warmup"],
+            "greptimedb_image": images[0]["image"], "greptimedb_images": images, "runtime_image": runtime_image,
             "instance_type": instance_type, "region": region, "prefix": prefix}
 
 
@@ -166,8 +182,32 @@ def partitioned_table(summary: dict, partitions: int) -> str:
     return AGENT_TABLE.format(partitions=clause)
 
 
-def group_cells(group: dict, smoke: bool) -> list:
-    """The cells of one group, in run order; a group without a name is the top-level inputs."""
+def parse_images(text: str) -> list:
+    """The images to compare: a JSON list of 2..4 {label, image}, the first the reference. Empty
+    means the one image of greptimedb_tag or greptimedb_image, with an empty label."""
+    if not text.strip():
+        return [{"label": "", "image": ""}]
+    images = json.loads(text)
+    if not isinstance(images, list) or not 2 <= len(images) <= 4 \
+            or not all(isinstance(i, dict) and set(i) == {"label", "image"} for i in images):
+        raise ValueError("images must be a JSON list of 2..4 objects, each with only a label and an image")
+    labels = [image["label"] for image in images]
+    for image in images:
+        if not isinstance(image["label"], str) or not IMAGE_LABEL.fullmatch(image["label"]):
+            raise ValueError(f"an image label is up to 8 lowercase letters and digits, got {image['label']!r}")
+        full_reference("an image", image["image"])
+    if len(set(labels)) != len(labels):
+        raise ValueError(f"image labels repeat: {labels}")
+    return images
+
+
+def cell_id(group: str, round_number: int, level: int, config: str, label: str) -> str:
+    return (f"{group}-" if group else "") + f"{round_number}-{level}-{config}" + (f"-{label}" if label else "")
+
+
+def group_cells(group: dict, smoke: bool, labels: list) -> list:
+    """The cells of one group, in run order; a group without a name is the top-level inputs.
+    Each cell runs once per image label, back to back."""
     name, mode = group.get("name", ""), group.get("mode", "load")
     if mode not in ("load", "crash", "drop"):
         raise ValueError(f"group {name!r}: mode must be load, crash or drop, got {mode!r}")
@@ -219,14 +259,16 @@ def group_cells(group: dict, smoke: bool) -> list:
     series = [config for config in WAL_CONFIGS if config in chosen]
     cells = []
     for round_number in range(1, rounds + 1):
-        # Reverse the order every other round so no configuration always runs first.
+        # Reverse the order every other round so no configuration or image always runs first.
+        forward = round_number % 2
         for level in levels:
-            for config in series if round_number % 2 else reversed(series):
+            for config in series if forward else reversed(series):
                 wal = WAL_CONFIGS[config] | ({"FLUSH_INTERVAL": RESTART_FLUSH_INTERVAL} if restart and config not in BASELINES else {})
-                cells.append(common | {
-                    "id": (f"{name}-" if name else "") + f"{round_number}-{level}-{config}",
-                    "round": round_number, "concurrency": level, "config": config, "wal": wal, "env": env,
-                })
+                for label in labels if forward else reversed(labels):
+                    cells.append(common | {
+                        "id": cell_id(name, round_number, level, config, label), "round": round_number,
+                        "concurrency": level, "config": config, "image": label, "wal": wal, "env": env,
+                    })
     return cells
 
 
@@ -251,11 +293,20 @@ def parse_groups(text: str, defaults: dict) -> list:
 
 
 def build_plan(concurrency: str, rows: str, batch_rows: str, rounds: str, smoke: bool,
-               configs: str = "R,D,E", extra_env: str = "", restart: bool = False, groups: str = "") -> dict:
+               configs: str = "R,D,E", extra_env: str = "", restart: bool = False, groups: str = "",
+               images: str = "", warmup: bool = False) -> dict:
     defaults = {"concurrency": concurrency, "rows": rows, "rounds": rounds, "configs": configs,
                 "extra_env": extra_env, "restart": restart}
     resolved = parse_groups(groups, defaults)
-    cells = [cell for group in resolved for cell in group_cells(group, smoke)]
+    labels = [image["label"] for image in parse_images(images)]
+    cells = [cell for group in resolved for cell in group_cells(group, smoke, labels)]
+    if warmup:
+        # Round 0 runs the first load cell once per image before anything else; no statistic uses it.
+        first = next((cell for cell in cells if cell["mode"] == "load"), None)
+        if first is None:
+            raise ValueError("warmup needs a load group")
+        cells = [first | {"id": cell_id(first["group"], 0, first["concurrency"], first["config"], label),
+                          "round": 0, "image": label, "warmup": True} for label in labels] + cells
     if len({cell["id"] for cell in cells}) != len(cells):
         raise ValueError("two cells share an id")
     corpora = []
@@ -270,10 +321,36 @@ def build_plan(concurrency: str, rows: str, batch_rows: str, rounds: str, smoke:
         "concurrency": sorted({cell["concurrency"] for cell in cells}),
         "configs": list(dict.fromkeys(cell["config"] for cell in cells)),
         "extra_env": parse_extra_env(extra_env), "groups": [g.get("name", "") for g in resolved],
-        "restart": restart, "corpora": corpora, "cells": cells,
+        "restart": restart, "images": labels, "warmup": warmup, "corpora": corpora, "cells": cells,
     }
 
 
+def load_rate(cell: dict) -> float:
+    durable = cell["wal"].get("ACK_MODE") == "durable"
+    points = LOAD_RATES["other" if not durable else "durable, partitioned" if cell["partitions"] > 1 else "durable"]
+    level = cell["concurrency"]
+    for (low, low_rate), (high, high_rate) in zip(points, points[1:]):
+        if level < high:
+            return low_rate + (high_rate - low_rate) * math.log(max(level, low) / low) / math.log(high / low)
+    return points[-1][1]
+
+
+def cell_seconds(cell: dict, batch_rows: int) -> float:
+    if cell["mode"] == "crash":
+        return CELL_SECONDS + cell["cycles"] * (sum(cell["kill_after"]) / 2 + CYCLE_SECONDS)
+    if cell["mode"] == "drop":
+        return CELL_SECONDS + cell["write_seconds"] + RESTART_SECONDS / 2
+    rate = load_rate(cell) * min(1.0, batch_rows / MEASURED_BATCH_ROWS)
+    return CELL_SECONDS + cell["rows"] / rate + cell["idle_seconds"] + (RESTART_SECONDS if cell["restart"] else 0)
+
+
+def estimate_minutes(plan: dict) -> int:
+    """The benchmark job's minutes on c6id.2xlarge, from the times of earlier runs."""
+    generate = GENERATE_SECONDS_PER_MILLION * sum(corpus["rows"] for corpus in plan["corpora"]) / 1e6
+    cells = sum(cell_seconds(cell, plan["batch_rows"]) for cell in plan["cells"])
+    return math.ceil((JOB_SECONDS + generate + cells * (1 + CLEANUP_SHARE)) / 60)
+
+
 def newest_nightly(tags: list[str]) -> str:
     dated = [(match.group(1), tag) for tag in tags if (match := NIGHTLY.fullmatch(tag))]
     if not dated:
@@ -281,15 +358,30 @@ def newest_nightly(tags: list[str]) -> str:
     return max(dated)[1]
 
 
+def full_reference(name: str, image: str) -> str:
+    if not isinstance(image, str) or not IMAGE.fullmatch(image) \
+            or not re.search(r"(:[^/]+|@sha256:[0-9a-f]{64})$", image):
+        raise ValueError(f"{name} must be a full reference with a tag or digest: {image!r}")
+    return image
+
+
 def resolve_image(image: str, tag: str) -> str:
     """The full image reference when given, else greptime/greptimedb at the resolved tag."""
     if image:
-        if not IMAGE.fullmatch(image) or not re.search(r"(:[^/]+|@sha256:[0-9a-f]{64})$", image):
-            raise ValueError(f"greptimedb_image must be a full reference with a tag or digest: {image!r}")
-        return image
+        return full_reference("greptimedb_image", image)
     return f"greptime/greptimedb:{resolve_tag(tag)}"
 
 
+def resolve_images(images: str, image: str, tag: str) -> list:
+    """The images to pull and check: the list to compare, or the one image of the other two inputs."""
+    entries = parse_images(images)
+    if entries[0]["label"]:
+        if image:
+            raise ValueError("set either images or greptimedb_image, not both")
+        return entries
+    return [{"label": "", "image": resolve_image(image, tag)}]
+
+
 def resolve_tag(requested: str) -> str:
     if requested != "nightly":
         if not TAG.fullmatch(requested):
@@ -347,13 +439,15 @@ def main() -> int:
     plan.add_argument("--smoke", choices=("true", "false"), required=True)
     plan.add_argument("--restart", choices=("true", "false"), default="false")
     plan.add_argument("--configs", default="R,D,E")
-    for name in ("extra-env", "groups"):
+    plan.add_argument("--warmup", choices=("true", "false"), default="false")
+    for name in ("extra-env", "groups", "images"):
         plan.add_argument(f"--{name}", default="")
     for name in ("timeout-minutes", "ttl-hours", "output"):
         plan.add_argument(f"--{name}")
     image = commands.add_parser("image")
     image.add_argument("--tag", required=True)
     image.add_argument("--image", default="")
+    image.add_argument("--images", default="")
     regions = commands.add_parser("regions")
     for name in ("bucket", "runner"):
         regions.add_argument(f"--{name}", required=True)
@@ -361,7 +455,7 @@ def main() -> int:
     table.add_argument("--summary", type=Path, required=True)
     table.add_argument("--partitions", type=int, required=True)
     manifest = commands.add_parser("manifest")
-    for name in ("results", "runtime-image", "instance-type", "region", "prefix"):
+    for name in ("results", "images", "runtime-image", "instance-type", "region", "prefix"):
         manifest.add_argument(f"--{name}", required=True)
     for name in ("probe", "cleanup"):
         bucket = commands.add_parser(name)
@@ -373,15 +467,21 @@ def main() -> int:
         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",
-                            args.configs, args.extra_env, args.restart == "true", args.groups)
+                            args.configs, args.extra_env, args.restart == "true", args.groups,
+                            args.images, args.warmup == "true")
+        minutes = estimate_minutes(result)
+        if args.timeout_minutes and minutes > int(args.timeout_minutes):
+            raise ValueError(f"the plan needs about {minutes} minutes on c6id.2xlarge, "
+                             f"more than the {args.timeout_minutes}-minute benchmark timeout")
         if args.output:
             with open(args.output, "w") as output:
                 json.dump(result, output, indent=2)
-        print(f"{len(result['cells'])} cells; corpora {[c['rows'] for c in result['corpora']]}")
+        print(f"{len(result['cells'])} cells; corpora {[c['rows'] for c in result['corpora']]}; "
+              f"about {minutes} minutes on c6id.2xlarge")
     elif args.command == "image":
-        reference = resolve_image(args.image, args.tag)
-        append_github("GITHUB_OUTPUT", "greptimedb_ref", reference)
-        print(f"GreptimeDB image: {reference}")
+        images = resolve_images(args.images, args.image, args.tag)
+        append_github("GITHUB_OUTPUT", "requested_images", json.dumps(images))
+        print("GreptimeDB images: " + ", ".join(f"{i['label']} {i['image']}".strip() for i in images))
     elif args.command == "regions":
         check_regions(args.bucket, args.runner)
         print(f"bucket and runner both in {args.runner}")
@@ -389,12 +489,10 @@ def main() -> int:
         print(partitioned_table(json.loads(args.summary.read_text()), args.partitions))
     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()),
+        manifest = build_manifest(json.loads((results / "plan.json").read_text()), json.loads(args.images),
                                   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']}")
+        print("GreptimeDB images: " + ", ".join(i["image"] for i in manifest["greptimedb_images"]))
     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}")
diff --git a/.github/workflows/oswal-s3-bench.yml b/.github/workflows/oswal-s3-bench.yml
index f5e92d2cad5..e61754cb1f1 100644
--- a/.github/workflows/oswal-s3-bench.yml
+++ b/.github/workflows/oswal-s3-bench.yml
@@ -14,6 +14,14 @@ on:
         description: 'Full image reference with a tag or digest; used instead of greptimedb_tag when set'
         type: string
         default: ''
+      images:
+        description: 'JSON list of 2..4 {"label", "image"} to compare on one instance instead of greptimedb_tag and greptimedb_image: every cell runs once per image, in an order that reverses every other round; a label is up to 8 lowercase letters and digits, an image a full reference with a tag or digest, the first the reference of the ratios'
+        type: string
+        default: ''
+      warmup:
+        description: Before round 1, run the first load cell once per image and leave it out of every statistic
+        type: boolean
+        default: false
       configs:
         description: 'WAL configurations: R (Raft Engine, sync_write), R0 (Raft Engine, standalone default), D (durable), E (enqueued)'
         type: string
@@ -47,7 +55,7 @@ on:
         type: number
         default: 1000
       rounds:
-        description: Rounds; the configuration order alternates between rounds (R, D, E then E, D, R)
+        description: Rounds; the configuration and image order alternates between rounds (R, D, E then E, D, R)
         type: number
         default: 2
       runtime_image:
@@ -68,6 +76,8 @@ env:
   SMOKE: ${{ inputs.smoke }}
   GREPTIMEDB_TAG: ${{ inputs.greptimedb_tag }}
   GREPTIMEDB_IMAGE_INPUT: ${{ inputs.greptimedb_image }}
+  OSWAL_IMAGES: ${{ inputs.images }}
+  WARMUP: ${{ inputs.warmup }}
   CONFIGS: ${{ inputs.configs }}
   EXTRA_ENV: ${{ inputs.extra_env }}
   RESTART: ${{ inputs.restart }}
@@ -86,7 +96,7 @@ jobs:
     timeout-minutes: 15
     outputs:
       instance_type: ${{ steps.runner.outputs.instance_type }}
-      greptimedb_ref: ${{ steps.plan.outputs.greptimedb_ref }}
+      greptimedb_images: ${{ steps.images.outputs.greptimedb_images }}
       runtime_image: ${{ steps.runtime.outputs.runtime_image }}
     steps:
     - uses: actions/checkout@v4
@@ -103,8 +113,15 @@ jobs:
         python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" \
           --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS" \
           --configs "$CONFIGS" --extra-env "$EXTRA_ENV" --restart "$RESTART" --groups "$OSWAL_GROUPS" \
+          --images "$OSWAL_IMAGES" --warmup "$WARMUP" \
           --timeout-minutes "$BENCHMARK_TIMEOUT_MINUTES" --ttl-hours "$JANITOR_TTL_HOURS"
-        python3 greptimedb/.github/scripts/oswal-s3-bench.py image --tag "$GREPTIMEDB_TAG" --image "$GREPTIMEDB_IMAGE_INPUT"
+        python3 greptimedb/.github/scripts/oswal-s3-bench.py image --tag "$GREPTIMEDB_TAG" --image "$GREPTIMEDB_IMAGE_INPUT" \
+          --images "$OSWAL_IMAGES"
+    - name: Pull every GreptimeDB image and record its version
+      id: images
+      env:
+        REQUESTED_IMAGES: ${{ steps.plan.outputs.requested_images }}
+      run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh check-images "$REQUESTED_IMAGES"
     - name: Resolve CI runner
       id: runner
       run: python3 greptimedb/.github/scripts/ci-runner-config.py --provider AWS --instance-type "$REQUESTED_INSTANCE_TYPE"
@@ -113,7 +130,7 @@ jobs:
         BUCKET_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
         RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }}
       run: python3 greptimedb/.github/scripts/oswal-s3-bench.py regions --bucket "$BUCKET_REGION" --runner "$RUNNER_REGION"
-    - name: Check the runtime image
+    - name: Pin the runtime image digest
       id: runtime
       run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh check-runtime "$RUNTIME_IMAGE"
   provision:
@@ -158,7 +175,7 @@ jobs:
     timeout-minutes: ${{ fromJSON(format('{0}', inputs.benchmark_timeout_minutes)) }}
     env:
       INSTANCE_TYPE: ${{ inputs.smoke && 'ubuntu-latest' || needs.validate.outputs.instance_type }}
-      GREPTIMEDB_REF: ${{ needs.validate.outputs.greptimedb_ref }}
+      GREPTIMEDB_IMAGES: ${{ needs.validate.outputs.greptimedb_images }}
       RUNTIME_IMAGE: ${{ needs.validate.outputs.runtime_image }}
       RUN_PREFIX: oswal-bench/${{ github.run_id }}-${{ github.run_attempt }}
       OWNER: o11ybench-oswal-${{ github.run_id }}-${{ github.run_attempt }}
@@ -183,13 +200,14 @@ jobs:
       run: |
         bash greptimedb/.github/scripts/pull-benchmark-image.sh greptime/minio:2024
         docker compose -f greptimedb/tests-integration/fixtures/docker-compose.yml up -d --wait minio
-    - name: Pull images and write the plan
+    - name: Pull and check images, write the plan
       run: |
         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" \
-          --configs "$CONFIGS" --extra-env "$EXTRA_ENV" --restart "$RESTART" --groups "$OSWAL_GROUPS"
-        python3 greptimedb/.github/scripts/oswal-s3-bench.py manifest --results "$RESULTS" \
+          --configs "$CONFIGS" --extra-env "$EXTRA_ENV" --restart "$RESTART" --groups "$OSWAL_GROUPS" \
+          --images "$OSWAL_IMAGES" --warmup "$WARMUP"
+        python3 greptimedb/.github/scripts/oswal-s3-bench.py manifest --results "$RESULTS" --images "$GREPTIMEDB_IMAGES" \
           --runtime-image "$RUNTIME_IMAGE" --instance-type "$INSTANCE_TYPE" --region "$S3_REGION" --prefix "$RUN_PREFIX"
     - name: Probe the bucket
       env:
diff --git a/tests/perf/test_oswal_s3_bench_scripts.py b/tests/perf/test_oswal_s3_bench_scripts.py
index 6934ff0db13..d95b29f7a52 100644
--- a/tests/perf/test_oswal_s3_bench_scripts.py
+++ b/tests/perf/test_oswal_s3_bench_scripts.py
@@ -224,15 +224,106 @@ class PlanTest(unittest.TestCase):
             with self.subTest(bucket=bucket), self.assertRaisesRegex(ValueError, "cross-region"):
                 bench.check_regions(bucket, runner)
 
-    def test_manifest_takes_the_sha256_digest(self):
+    def test_manifest_records_the_checked_images_pinned_to_their_digests(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, "c6id.2xlarge",
+        checked = [{"label": "", "image": "greptime/greptimedb@sha256:" + "a" * 64,
+                    "requested": "greptime/greptimedb:v1.4.0-nightly-20261012", "version": "1.4.0", "commit": "abc"}]
+        manifest = bench.build_manifest(plan, checked, "runtime@sha256:" + "b" * 64, "c6id.2xlarge",
                                         "ap-southeast-1", "oswal-bench/1-1")
         self.assertEqual(manifest["greptimedb_image"], "greptime/greptimedb@sha256:" + "a" * 64)
+        self.assertEqual(manifest["greptimedb_images"], checked)
         self.assertEqual((manifest["rows"], manifest["rounds"], manifest["concurrency"]), (2_000_000, 2, [1, 8]))
-        with self.assertRaises(ValueError):
-            bench.build_manifest(plan, [{"RepoDigests": []}], "", "", "", "")
+        for images in ([checked[0] | {"image": "greptime/greptimedb:v1.4.0"}], [checked[0] | {"label": "base"}]):
+            with self.subTest(images=images), self.assertRaises(ValueError):
+                bench.build_manifest(plan, images, "", "", "", "")
+
+    def test_every_cell_runs_once_per_image_in_an_order_that_reverses_every_other_round(self):
+        images = json.dumps([{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"}
+                             for label in ("base", "pr1", "pr2")])
+        plan = bench.build_plan("8", "2000000", "1000", "3", smoke=False, configs="D", images=images)
+        self.assertEqual([c["id"] for c in plan["cells"]], [
+            "1-8-D-base", "1-8-D-pr1", "1-8-D-pr2", "2-8-D-pr2", "2-8-D-pr1", "2-8-D-base",
+            "3-8-D-base", "3-8-D-pr1", "3-8-D-pr2"])
+        self.assertEqual(plan["images"], ["base", "pr1", "pr2"])
+        # The configurations reverse with the images, and a warm-up runs the first load cell per image.
+        groups = json.dumps([{"name": "crash", "mode": "crash", "configs": "E", "rounds": 1, "cycles": 1},
+                             {"name": "single", "configs": "D,E"}])
+        plan = bench.build_plan("8", "2000000", "1000", "2", smoke=False, groups=groups,
+                                images=json.dumps(json.loads(images)[:2]), warmup=True)
+        self.assertEqual([c["id"] for c in plan["cells"]], [
+            "single-0-8-D-base", "single-0-8-D-pr1", "crash-1-8-E-base", "crash-1-8-E-pr1",
+            "single-1-8-D-base", "single-1-8-D-pr1", "single-1-8-E-base", "single-1-8-E-pr1",
+            "single-2-8-E-pr1", "single-2-8-E-base", "single-2-8-D-pr1", "single-2-8-D-base"])
+        warmups = [c for c in plan["cells"] if c.get("warmup")]
+        self.assertEqual([(c["round"], c["image"], c["mode"]) for c in warmups], [(0, "base", "load"), (0, "pr1", "load")])
+        self.assertEqual((plan["rounds"], plan["warmup"]), (2, True))
+        # Without images every cell keeps its id and runs the one image.
+        single = bench.build_plan("8", "2000000", "1000", "1", smoke=False, configs="D")
+        self.assertEqual([(c["id"], c["image"]) for c in single["cells"]], [("1-8-D", "")])
+        self.assertEqual((single["images"], single["warmup"]), ([""], False))
+
+    def test_invalid_images_and_warmups_are_rejected(self):
+        ref = "docker.io/greptime/greptimedb-dev:dev"
+        for images in ([{"label": "base", "image": ref}], [{"label": f"i{n}", "image": ref} for n in range(5)],
+                       [{"label": "base", "image": ref}, {"label": "base", "image": ref}],
+                       [{"label": "Base", "image": ref}, {"label": "pr1", "image": ref}],
+                       [{"label": "ninechars", "image": ref}, {"label": "pr1", "image": ref}],
+                       [{"label": 1, "image": ref}, {"label": "pr1", "image": ref}],
+                       [{"label": "base", "image": "greptimedb"}, {"label": "pr1", "image": ref}],
+                       [{"label": "base", "image": ref, "env": "A=1"}, {"label": "pr1", "image": ref}],
+                       {"base": ref}):
+            with self.subTest(images=images), self.assertRaises(ValueError):
+                bench.build_plan("8", "6000", "1000", "1", smoke=False, images=json.dumps(images))
+        crash_only = json.dumps([{"name": "c", "mode": "crash", "configs": "D"}])
+        with self.assertRaisesRegex(ValueError, "warmup needs a load group"):
+            bench.build_plan("8", "6000", "1000", "1", smoke=False, groups=crash_only, warmup=True)
+        # One image twice compares it with itself.
+        pair = [{"label": "base", "image": ref}, {"label": "again", "image": ref}]
+        self.assertEqual(bench.resolve_images(json.dumps(pair), "", "nightly"), pair)
+        with self.assertRaisesRegex(ValueError, "not both"):
+            bench.resolve_images(json.dumps(pair), ref, "nightly")
+        self.assertEqual(bench.resolve_images("", ref, "nightly"), [{"label": "", "image": ref}])
+
+    def test_a_plan_longer_than_the_benchmark_timeout_is_refused(self):
+        groups = json.dumps([{"name": "single", "configs": "D,E"}, {"name": "multi", "configs": "D,E", "partitions": 8}])
+        images = json.dumps([{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"}
+                             for label in ("base", "pr1", "pr2")])
+        plan = bench.build_plan("8", "2000000", "1000", "5", smoke=False, groups=groups, images=images, warmup=True)
+        self.assertEqual(len(plan["cells"]), 63)
+        self.assertEqual(bench.estimate_minutes(plan), 75)
+        cells = {c["id"]: c for c in plan["cells"]}
+        # Durable at concurrency 8 writes 34,000 rows/s to one region and 26,000 to eight.
+        self.assertAlmostEqual(bench.cell_seconds(cells["single-1-8-D-base"], 1000), 5 + 2_000_000 / 34_000)
+        self.assertAlmostEqual(bench.cell_seconds(cells["multi-1-8-D-pr1"], 1000), 5 + 2_000_000 / 26_000)
+        self.assertAlmostEqual(bench.cell_seconds(cells["multi-1-8-E-pr2"], 4000), 5 + 2_000_000 / 60_000)
+        # A smaller batch keeps the request rate: one-row batches write a thousandth of the rows.
+        self.assertAlmostEqual(bench.cell_seconds(cells["single-1-8-D-base"], 1), 5 + 2_000_000 / 34)
+        durable = cells["single-1-8-D-base"]
+        self.assertAlmostEqual(bench.load_rate(durable | {"concurrency": 4}), 9_000 + 25_000 * 2 / 3)
+        self.assertEqual(bench.load_rate(durable | {"concurrency": 64}), 55_000)
+        crash = bench.build_plan("8", "6000", "1000", "1", smoke=False, groups=json.dumps(
+            [{"name": "c", "mode": "crash", "configs": "D", "cycles": 5, "kill_after": [5, 20]}]))["cells"][0]
+        self.assertEqual(bench.cell_seconds(crash, 1), 5 + 5 * (12.5 + 15))
+        command = [sys.executable, str(SCRIPTS / "oswal-s3-bench.py"), "plan", "--smoke", "false", "--concurrency",
+                   "8", "--rows", "2000000", "--rounds", "5", "--groups", groups,
+                   "--images", images, "--warmup", "true", "--ttl-hours", "8"]
+        for batch_rows, timeout, expected in (("1000", "75", "about 75 minutes"), ("1000", "74", "more than the 74-minute"),
+                                               ("1", "360", "more than the 360-minute")):
+            done = subprocess.run(command + ["--batch-rows", batch_rows, "--timeout-minutes", timeout],
+                                  capture_output=True, text=True, timeout=60)
+            with self.subTest(batch_rows=batch_rows, timeout=timeout):
+                self.assertEqual(done.returncode, 0 if expected.startswith("about") else 1, done.stderr)
+                self.assertIn(expected, done.stdout + done.stderr)
+
+    def test_the_workflow_checks_every_image_before_an_instance_is_allocated(self):
+        workflow = (SCRIPTS.parent / "workflows" / "oswal-s3-bench.yml").read_text()
+        self.assertIn("OSWAL_IMAGES: ${{ inputs.images }}", workflow)
+        self.assertEqual(workflow.count('--images "$OSWAL_IMAGES"'), 3)
+        self.assertEqual(workflow.count('--warmup "$WARMUP"'), 2)
+        validate = workflow[workflow.index("  validate:"):workflow.index("  provision:")]
+        self.assertIn("greptimedb_images: ${{ steps.images.outputs.greptimedb_images }}", validate)
+        self.assertIn("check-images \"$REQUESTED_IMAGES\"", validate)
+        self.assertIn("GREPTIMEDB_IMAGES: ${{ needs.validate.outputs.greptimedb_images }}", workflow)
 
 
 def write_lines(path, lines):
@@ -425,11 +516,12 @@ class SummaryTest(unittest.TestCase):
         self.assertNotIn("kafka_client_produce_elapsed_count", summary.parse_metrics(ACK_BUCKETS)["samples"])
 
     def write_cell(self, root, cell_id, rows_per_second, exit_code=0, metrics="", **extra):
-        round_number, concurrency, config = cell_id.split("-")
+        round_number, concurrency, config, *image = cell_id.split("-")
         path = root / "cells" / cell_id
         (path / "greptimedb-1").mkdir(parents=True)
         (path / "cell.json").write_text(json.dumps({
             "id": cell_id, "round": int(round_number), "concurrency": int(concurrency), "config": config,
+            "image": image[0] if image else "",
             "wal": bench.WAL_CONFIGS[config], "metrics_scraped": True, "env": {},
             "exit_code": exit_code, "wal_device": "/dev/nvme1n1", "version": "1.4.0", "load_wall_seconds": 61.0} | extra))
         if exit_code == 0:
@@ -664,6 +756,51 @@ class SummaryTest(unittest.TestCase):
             self.assertIn("1-1-D (failed, exit 1)", summary.render({}, rows, summary.ratios(rows)))
 
 
+    def test_images_are_compared_with_the_first_in_every_round(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            images = [{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"} for label in ("base", "pr1")]
+            plan = bench.build_plan("8", "6000", "1000", "2", smoke=False, configs="R,D", images=json.dumps(images),
+                                    warmup=True)
+            (root / "plan.json").write_text(json.dumps(plan))
+            (root / "run.json").write_text(json.dumps({"rows": 6000, "batch_rows": 1000, "greptimedb_images": [
+                i | {"image": f"greptime/greptimedb-dev@sha256:{n * 64}", "version": "1.4.0", "commit": f"{n * 7}"}
+                for i, n in zip(images, "ab")]}))
+            # The warm-up cells are far off and must not move any ratio.
+            speeds = {"0-8-R-base": 10.0, "0-8-R-pr1": 999.0,
+                      "1-8-R-base": 100.0, "1-8-R-pr1": 100.0, "1-8-D-base": 50.0, "1-8-D-pr1": 40.0,
+                      "2-8-D-pr1": 66.0, "2-8-D-base": 55.0, "2-8-R-pr1": 100.0, "2-8-R-base": 110.0}
+            for cell_id, speed in speeds.items():
+                self.write_cell(root, cell_id, speed, warmup=cell_id.startswith("0-"),
+                                host_load_start=[0.5, 0.4, 0.3], host_load_end=[1.25, 0.5, 0.3])
+            with open(root / "cells" / "2-8-D-pr1" / "greptimedb-1" / "load-transcript.jsonl", "w") as out:
+                for seconds in (0.2, 0.3):
+                    out.write(json.dumps({"step": "grpc_write_batch", "status": "success", "duration_seconds": seconds}) + "\n")
+            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())
+            compared = {(c["config"], c["image"]): c for c in result["image_ratios"]}
+            self.assertEqual(list(compared), [("R", "base"), ("R", "pr1"), ("D", "base"), ("D", "pr1")])
+            durable = compared[("D", "pr1")]
+            self.assertEqual(durable["ratio_to"], "base")
+            self.assertEqual(durable["per_round"], [0.8, 1.2])
+            self.assertAlmostEqual(durable["geomean"], 0.96 ** 0.5)
+            self.assertEqual((durable["min"], durable["max"]), (0.8, 1.2))
+            self.assertEqual([r["rows_per_second"] for r in durable["rounds"]], [40.0, 66.0])
+            self.assertEqual((durable["rounds"][1]["client_p50"], durable["rounds"][1]["client_p99"]), (0.2, 0.3))
+            self.assertEqual(durable["rounds"][0]["host_load_end"], [1.25, 0.5, 0.3])
+            self.assertEqual((compared[("D", "base")]["per_round"], compared[("D", "base")]["geomean"]), (None, None))
+            # The ratios to the Raft Engine control stay within each image.
+            pairs = {(r["image"], r["ratio"]): r["per_round"] for r in result["ratios"]}
+            self.assertEqual(pairs, {("base", "D/R"): [0.5, 0.5], ("pr1", "D/R"): [0.4, 0.66]})
+            self.assertIn("| base | `greptime/greptimedb-dev@sha256:" + "a" * 64 + "` | 1.4.0 | `aaaaaaa` |", output)
+            self.assertIn("| — | 8 | D | pr1 | 40; 66 | — / —; 200.0 / 300.0 | 0.800, 1.200 | 0.980 | 0.800 | 1.200 |", output)
+            self.assertIn("| — | 8 | D | base | 50; 55 | — / —; — / — | — | — | — | — |", output)
+            self.assertIn("| 0-8-R-pr1 (warm-up) |", output)
+            self.assertIn("| `/dev/nvme1n1` | 0.50 / 1.25 |", output)
+            self.assertIn("| — | pr1 | 8 | D/R | 0.400, 0.660 | 0.514 |", output)
+
+
 FAKE_DOCKER = r"""#!/usr/bin/env bash
 printf '%s\n' "docker $*" >> "$FAKE_LOG"
 case "$1" in
@@ -677,6 +814,7 @@ case "$1" in
         echo "{\"count_before\": 0, \"count_after\": $FAKE_ROWS, \"load_seconds\": 2, \"rows_per_second\": 3000, \"table\": \"agent_observations\"}" \
             > "$results/greptimedb-1/load-result.json" ;;
     kill) touch "${FAKE_KILL_FLAG:-/dev/null}" ;;
+    rm) [[ -z "${FAKE_LOAD_AFTER:-}" ]] || echo "$FAKE_LOAD_AFTER" > "$OSWAL_LOADAVG" ;;
     inspect) echo true ;;
     exec) [[ "$*" == *cpu.stat* ]] && echo 'usage_usec 5000000' || echo 'version: 1.4.0-test' ;;
     logs) echo 'server log'
@@ -723,14 +861,16 @@ done
 
 class CellLifecycleTest(unittest.TestCase):  # the run script against fake docker and curl
 
-    def run_cells(self, root, load_rc, metrics_rc=0, restart=False, count=6000, groups=""):
+    def run_cells(self, root, load_rc, metrics_rc=0, restart=False, count=6000, groups="", images="", checked=None):
         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, configs="D,E", restart=restart,
-                                extra_env="RUST_LOG=info", groups=groups)
+                                extra_env="RUST_LOG=info", groups=groups, images=images)
+        if checked is None:
+            checked = [{"label": label, "image": f"greptimedb{label}@sha256:" + "a" * 64} for label in plan["images"]]
         (root / "results").mkdir()
         (root / "results" / "plan.json").write_text(json.dumps(plan))
         env = os.environ | {
@@ -738,7 +878,7 @@ class CellLifecycleTest(unittest.TestCase):  # the run script against fake docke
             "FAKE_LOAD_RC": str(load_rc), "FAKE_METRICS_RC": str(metrics_rc), "FAKE_ROWS": "6000", "FAKE_COUNT": str(count),
             "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,
+            "RUNTIME_IMAGE": "runtime@sha256:" + "b" * 64, "GREPTIMEDB_IMAGES": json.dumps(checked),
             "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",
         }
@@ -1070,5 +1210,155 @@ class CellLifecycleTest(unittest.TestCase):  # the run script against fake docke
             self.assertEqual((recorded["exit_code"], recorded["restart_count"]), (5, 5999))
 
 
+    def test_every_cell_starts_its_own_image_and_records_the_host_load(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "loadavg").write_text("0.50 0.40 0.30 1/200 99\n")
+            images = json.dumps([{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"}
+                                 for label in ("base", "pr1")])
+            old = os.environ.copy()
+            os.environ.update({"OSWAL_LOADAVG": str(root / "loadavg"), "FAKE_LOAD_AFTER": "1.25 1.00 0.75 1/200 99"})
+            try:
+                cells, _, log, _ = self.run_cells(root, load_rc=0, images=images)
+            finally:
+                os.environ.clear()
+                os.environ.update(old)
+            self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr)
+            starts = [line for line in log.splitlines() if line.startswith("docker run -d")]
+            self.assertEqual([re.search(r"--name o11ybench-oswal-1-1-(\S+) ", line).group(1) for line in starts],
+                             ["1-8-D-base", "1-8-D-pr1", "1-8-E-base", "1-8-E-pr1"])
+            for line in starts:
+                label = line.split("--name ")[1].split()[0].rsplit("-", 1)[1]
+                self.assertIn(f" greptimedb{label}@sha256:{'a' * 64} standalone start", line)
+            self.assertIn(f"TARGET_IMAGE=greptimedbpr1@sha256:{'a' * 64}", log)
+            first, second = (json.loads((root / "results" / "cells" / i / "cell.json").read_text())
+                             for i in ("1-8-D-base", "1-8-D-pr1"))
+            self.assertEqual((first["image"], first["host_load_start"], first["host_load_end"]),
+                             ("base", [0.5, 0.4, 0.3], [1.25, 1.0, 0.75]))
+            self.assertEqual(second["host_load_start"], [1.25, 1.0, 0.75])
+
+    def test_a_cell_without_a_checked_image_fails(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            images = json.dumps([{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"}
+                                 for label in ("base", "pr1")])
+            cells, _, log, _ = self.run_cells(root, load_rc=0, images=images,
+                                              checked=[{"label": "base", "image": "greptimedb@sha256:" + "a" * 64}])
+            self.assertNotEqual(cells.returncode, 0)
+            self.assertIn("1-8-D-pr1: no image labelled 'pr1'", cells.stderr)
+            self.assertEqual(log.count("docker run -d"), 2)
+
+    def test_check_images_pins_every_image_to_its_digest_and_records_its_version(self):
+        fake_docker = r"""#!/usr/bin/env bash
+printf '%s\n' "docker $*" >> "$FAKE_LOG"
+case "$1 $2" in
+    "image inspect") printf '[{"RepoDigests": ["%s@sha256:%064d"]}]\n' "${3%:*}" "$(printf '%s' "$3" | cksum | cut -d' ' -f1)" ;;
+    "run --rm") [[ "$*" == *nocommit* ]] || commit="commit: 0123456789abcdef$(printf '%s' "$4" | cksum | cut -c1-4)"
+        printf 'greptime\nbranch: main\n%s\nclean: true\nversion: 1.4.0\n' "${commit:-}" ;;
+esac
+"""
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for name, body in (("docker", fake_docker), ("timeout", '#!/usr/bin/env bash\nshift 2\nexec "$@"\n')):
+                (root / name).write_text(body)
+                (root / name).chmod(0o755)
+            env = os.environ | {"PATH": f"{root}:{os.environ['PATH']}", "FAKE_LOG": str(root / "fake.log"),
+                                "GITHUB_OUTPUT": str(root / "output")}
+            requested = [{"label": label, "image": f"docker.io/greptime/greptimedb-dev:{label}"} for label in ("base", "pr1")]
+            done = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "check-images", json.dumps(requested)],
+                                  env=env, capture_output=True, text=True, timeout=60)
+            self.assertEqual(done.returncode, 0, done.stderr)
+            name, _, value = (root / "output").read_text().strip().partition("=")
+            checked = json.loads(value)
+            self.assertEqual(name, "greptimedb_images")
+            self.assertEqual([(c["label"], c["requested"], c["version"]) for c in checked],
+                             [(r["label"], r["image"], "1.4.0") for r in requested])
+            for c in checked:
+                self.assertRegex(c["image"], r"^docker\.io/greptime/greptimedb-dev@sha256:[0-9]{64}$")
+                self.assertTrue(c["commit"].startswith("0123456789abcdef"), c)
+                self.assertIn(f"docker run --rm --entrypoint greptime {c['image']} --version", (root / "fake.log").read_text())
+            self.assertNotEqual(checked[0]["image"], checked[1]["image"])
+            requested[1]["image"] = "docker.io/greptime/nocommit:pr1"
+            done = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "check-images", json.dumps(requested)],
+                                  env=env, capture_output=True, text=True, timeout=60)
+            self.assertNotEqual(done.returncode, 0)
+            self.assertIn("reports no version or commit", done.stderr)
+
+
+    def test_check_runtime_pins_a_named_digest_offline_and_resolves_only_a_tag(self):
+        digest = "sha256:" + "c" * 64
+        fake_docker = f"""#!/usr/bin/env bash
+printf '%s\\n' "docker $*" >> "$FAKE_LOG"
+[[ "$1 $2 $3" == "buildx imagetools inspect" ]] || exit 9
+printf '{{"mediaType": "application/vnd.oci.image.index.v1+json", "digest": "{digest}", "size": 856}}\\n'
+"""
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "docker").write_text(fake_docker)
+            (root / "docker").chmod(0o755)
+            env = os.environ | {"PATH": f"{root}:{os.environ['PATH']}", "FAKE_LOG": str(root / "fake.log"),
+                                "GITHUB_OUTPUT": str(root / "output")}
+            repository, named = "public.ecr.aws/i8k6a5e1/greptime/o11ybenchruntime", "sha256:" + "d" * 64
+            # Expected pin, or error, and whether the registry is asked.
+            for image, expected, asks in ((f"{repository}:77a84e1c8136", f"{repository}@{digest}", True),
+                                          (f"{repository}:77a84e1c8136@{named}", f"{repository}@{named}", False),
+                                          (f"{repository}@{named}", f"{repository}@{named}", False),
+                                          (f"{repository}:77a84e1c8136@sha256:abc", "not a full ECR Public", False),
+                                          (f"public.ecr.aws/@{named}", "not a full ECR Public", False),
+                                          (f"{repository}@@{named}", "not a full ECR Public", False),
+                                          (f"{repository}:bad tag", "not a full ECR Public", False),
+                                          ("docker.io/greptime/o11ybenchruntime:1", "must come from AWS ECR Public", False)):
+                with self.subTest(image=image):
+                    (root / "output").write_text("")
+                    (root / "fake.log").write_text("")
+                    done = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "check-runtime", image],
+                                          env=env, capture_output=True, text=True, timeout=60)
+                    if expected.startswith(repository):
+                        self.assertEqual(done.returncode, 0, done.stderr)
+                        self.assertEqual((root / "output").read_text(), f"runtime_image={expected}\n")
+                    else:
+                        self.assertNotEqual(done.returncode, 0)
+                        self.assertIn(expected, done.stderr)
+                        self.assertEqual((root / "output").read_text(), "")
+                    calls = (root / "fake.log").read_text().splitlines()
+                    self.assertEqual(calls, [f"docker buildx imagetools inspect --format {{{{json .Manifest}}}} {image}"]
+                                     if asks else [])
+
+    def test_the_benchmark_host_checks_the_runtime_contents_before_pulling_greptimedb(self):
+        # The fake runs the step's own check in a directory standing in for the image's working directory.
+        fake_docker = r"""#!/usr/bin/env bash
+printf '%s\n' "docker $*" >> "$FAKE_LOG"
+case "$1" in
+    run) cd "$FAKE_RUNTIME_ROOT" && exec sh -c "$7" ;;
+    image) echo '[{}]' ;;
+esac
+"""
+        loader, generator = "scripts/agent-observability/greptimedb.sh", "scripts/generate_agent_observations.py"
+        for present, passes in (((loader, generator), True), ((generator,), False), ((loader,), False)):
+            with self.subTest(present=present), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                for name, body in (("docker", fake_docker), ("timeout", '#!/usr/bin/env bash\nshift 2\nexec "$@"\n')):
+                    (root / name).write_text(body)
+                    (root / name).chmod(0o755)
+                for path in present:
+                    (root / "image" / path).parent.mkdir(parents=True, exist_ok=True)
+                    (root / "image" / path).write_text("")
+                runtime = "public.ecr.aws/i8k6a5e1/greptime/o11ybenchruntime@sha256:" + "c" * 64
+                env = os.environ | {"PATH": f"{root}:{os.environ['PATH']}", "FAKE_LOG": str(root / "fake.log"),
+                                    "FAKE_RUNTIME_ROOT": str(root / "image"), "RESULTS": str(root / "results"),
+                                    "RUNTIME_IMAGE": runtime,
+                                    "GREPTIMEDB_IMAGES": json.dumps([{"label": "", "image": "greptimedb@sha256:" + "a" * 64}])}
+                done = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "pull"], env=env,
+                                      capture_output=True, text=True, timeout=60)
+                log = (root / "fake.log").read_text().splitlines()
+                self.assertTrue(log[1].startswith(f"docker run --rm --entrypoint sh {runtime} -c "), log[1])
+                if passes:
+                    self.assertEqual(done.returncode, 0, done.stderr)
+                    self.assertEqual([line.split()[1] for line in log], ["pull", "run", "pull", "image", "image"])
+                else:
+                    self.assertNotEqual(done.returncode, 0)
+                    self.assertIn("lacks the agent observability loader or generator", done.stderr)
+                    self.assertEqual([line.split()[1] for line in log], ["pull", "run"])
+
 if __name__ == "__main__":
     unittest.main()