[Accuracy Benchmark] feat: add accuracy benchmark integrations for t2i and i2i (#1917)

Signed-off-by: David Chen <530634352@qq.com>
Signed-off-by: WeiQing Chen <40507679+david6666666@users.noreply.github.com>

.buildkite/test-nightly.yml

@@ -225,6 +225,93 @@ steps:
                     mountPath: /dev/shm
                   - name: hf-cache
                     mountPath: /root/.cache/huggingface
+                env:
+                  - name: HF_HOME
+                    value: /root/.cache/huggingface
+                  - name: HF_TOKEN
+                    valueFrom:
+                      secretKeyRef:
+                        name: hf-token-secret
+                        key: token
+            nodeSelector:
+              node.kubernetes.io/instance-type: gpu-h100-sxm
+            volumes:
+              - name: devshm
+                emptyDir:
+                  medium: Memory
+              - name: hf-cache
+                hostPath:
+                  path: /mnt/hf-cache
+                  type: DirectoryOrCreate
+
+  - label: ":full_moon: GEBench Accuracy Test with H100"
+    key: nightly-gebench-accuracy
+    timeout_in_minutes: 60
+    depends_on: upload-nightly-pipeline
+    if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test"
+    commands:
+      - export VLLM_WORKER_MULTIPROC_METHOD=spawn
+      - pytest -s -v tests/e2e/accuracy/test_gebench_h100_smoke.py --run-level advanced_model --gebench-model Qwen/Qwen-Image-2512 --accuracy-judge-model QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ --accuracy-gpu 0 --gebench-port 8093 --accuracy-workers 1
+      - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gebench_qwen-image-2512/summary*.json"
+    agents:
+      queue: "mithril-h100-pool"
+    plugins:
+      - kubernetes:
+          podSpec:
+            containers:
+              - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+                resources:
+                  limits:
+                    nvidia.com/gpu: 1
+                volumeMounts:
+                  - name: devshm
+                    mountPath: /dev/shm
+                  - name: hf-cache
+                    mountPath: /root/.cache/huggingface
+                env:
+                  - name: HF_HOME
+                    value: /root/.cache/huggingface
+                  - name: HF_TOKEN
+                    valueFrom:
+                      secretKeyRef:
+                        name: hf-token-secret
+                        key: token
+            nodeSelector:
+              node.kubernetes.io/instance-type: gpu-h100-sxm
+            volumes:
+              - name: devshm
+                emptyDir:
+                  medium: Memory
+              - name: hf-cache
+                hostPath:
+                  path: /mnt/hf-cache
+                  type: DirectoryOrCreate
+
+  - label: ":full_moon: GEdit-Bench Accuracy Test with H100"
+    key: nightly-gedit-bench-accuracy
+    timeout_in_minutes: 60
+    depends_on: upload-nightly-pipeline
+    if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test"
+    commands:
+      - export VLLM_WORKER_MULTIPROC_METHOD=spawn
+      - pytest -s -v tests/e2e/accuracy/test_gedit_bench_h100_smoke.py --run-level advanced_model --gedit-model Qwen/Qwen-Image-Edit --accuracy-judge-model QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ --accuracy-gpu 0 --gedit-port 8093 --gedit-samples-per-group 20 --accuracy-workers 1
+      - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gedit_scores_qwen-image-edit/qwen-image-edit_all_all_vie_score_*.csv"
+      - buildkite-agent artifact upload "tests/e2e/accuracy/artifacts/gedit_scores_qwen-image-edit/qwen-image-edit_all_all_summary_*.json"
+    agents:
+      queue: "mithril-h100-pool"
+    plugins:
+      - kubernetes:
+          podSpec:
+            containers:
+              - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+                resources:
+                  limits:
+                    nvidia.com/gpu: 1
+                volumeMounts:
+                  - name: devshm
+                    mountPath: /dev/shm
+                  - name: hf-cache
+                    mountPath: /root/.cache/huggingface
                 env:
                   - name: HF_HOME
                     value: /root/.cache/huggingface

benchmarks/README.md

@@ -15,6 +15,8 @@ This document explains the benchmark architecture across all benchmark assets in
 ## Directory layout
 - `benchmarks/build_dataset/` — dataset prep utilities (e.g., SeedTTS top100).
 - `benchmarks/<model>/vllm_omni/` — vLLM-Omni pipeline benchmarks, logs, outputs.
+- `benchmarks/accuracy/` — accuracy benchmark integrations that adapt external
+  benchmark suites to vLLM-Omni serving and evaluation flows.
 - Add new tasks under `benchmarks/<model>/...` with the same pattern: `transformers/`, `vllm_omni/`, task-specific README, and (optionally) dataset prep notes.
 
 ## Reference workflows

benchmarks/__init__.py

@@ -0,0 +1 @@
+"""Benchmark helpers and runnable benchmark entrypoints."""

benchmarks/accuracy/README.md

@@ -0,0 +1,27 @@
+# Accuracy Benchmarks
+
+This directory hosts accuracy benchmark integrations that run entirely through a
+local `vllm-omni serve` deployment.
+
+Current integrations:
+
+- `text_to_image/`: GEBench generation + local judge scoring flow.
+- `image_to_image/`: GEdit-Bench generation + local VIEScore-style scoring flow.
+
+Design notes:
+
+- Generation is executed through the OpenAI-compatible endpoints exposed by
+  `vllm-omni serve`.
+- Evaluation is also executed through a local OpenAI-compatible judge model
+  served by `vllm-omni`.
+- Both generation and judge requests accept either `http://host:port` or
+  `http://host:port/v1`.
+- Output directory layout intentionally stays close to the upstream repos.
+
+Test guidance:
+
+- Local static/self-checks live in `tests/benchmarks/test_accuracy_bench_utils.py`.
+- End-to-end generation/evaluation should be validated in a remote GPU
+  environment. In the current repo marker system there is `L4` but no `L5`
+  marker, so benchmark smoke tests should be wired as `advanced_model +
+  benchmark + L4` when GPU capacity is available.

benchmarks/accuracy/__init__.py

@@ -0,0 +1 @@
+"""Accuracy benchmark integrations for vLLM-Omni."""

benchmarks/accuracy/common.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+import base64
+import io
+import json
+from pathlib import Path
+from typing import Any
+
+import requests
+from PIL import Image
+
+
+def ensure_dir(path: Path) -> Path:
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def write_json(path: Path, payload: dict[str, Any]) -> None:
+    ensure_dir(path.parent)
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(payload, handle, indent=2, ensure_ascii=False)
+
+
+def save_image(path: Path, image: Image.Image) -> None:
+    ensure_dir(path.parent)
+    image.save(path)
+
+
+def find_first_image(folder: Path, stem: str | None = None) -> Path | None:
+    patterns = [f"{stem}.*"] if stem else ["*.png", "*.jpg", "*.jpeg", "*.webp"]
+    for pattern in patterns:
+        for candidate in sorted(folder.glob(pattern)):
+            if candidate.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}:
+                return candidate
+    return None
+
+
+def extract_json_object(raw_text: str) -> dict[str, Any]:
+    raw_text = raw_text.strip()
+    delimiter = "||V^=^V||"
+    if raw_text.count(delimiter) >= 2:
+        start = raw_text.find(delimiter) + len(delimiter)
+        end = raw_text.rfind(delimiter)
+        raw_text = raw_text[start:end].strip()
+
+    start = raw_text.find("{")
+    end = raw_text.rfind("}")
+    if start == -1 or end == -1 or end < start:
+        raise ValueError(f"Could not find JSON object in: {raw_text[:200]}")
+    return json.loads(raw_text[start : end + 1])
+
+
+def build_openai_url(base_url: str, api_path: str) -> str:
+    base = base_url.rstrip("/")
+    normalized_path = api_path if api_path.startswith("/") else f"/{api_path}"
+    if base.endswith(normalized_path):
+        return base
+    if base.endswith("/v1"):
+        return f"{base}{normalized_path}"
+    return f"{base}/v1{normalized_path}"
+
+
+def pil_to_base64(image: Image.Image, image_format: str = "PNG") -> str:
+    buffer = io.BytesIO()
+    image.save(buffer, format=image_format)
+    return base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+
+def pil_to_data_url(image: Image.Image, image_format: str = "PNG") -> str:
+    return f"data:image/{image_format.lower()};base64,{pil_to_base64(image, image_format=image_format)}"
+
+
+def decode_base64_image(encoded: str) -> Image.Image:
+    image = Image.open(io.BytesIO(base64.b64decode(encoded)))
+    image.load()
+    return image.convert("RGB")
+
+
+def pil_to_png_bytes(image: Image.Image) -> bytes:
+    buffer = io.BytesIO()
+    image.save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+class VllmOmniImageClient:
+    """Thin OpenAI-compatible image client for vLLM-Omni serving."""
+
+    def __init__(self, base_url: str, api_key: str = "EMPTY", timeout: int = 600):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.timeout = timeout
+
+    @property
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+
+    def generate_text_to_image(
+        self,
+        *,
+        model: str,
+        prompt: str,
+        width: int,
+        height: int,
+        num_inference_steps: int = 20,
+        guidance_scale: float | None = None,
+        seed: int | None = None,
+        output_compression: int | None = None,
+    ) -> Image.Image:
+        payload: dict[str, Any] = {
+            "model": model,
+            "prompt": prompt,
+            "n": 1,
+            "size": f"{width}x{height}",
+            "response_format": "b64_json",
+            "num_inference_steps": num_inference_steps,
+        }
+        if guidance_scale is not None:
+            payload["guidance_scale"] = guidance_scale
+        if seed is not None:
+            payload["seed"] = seed
+        if output_compression is not None:
+            payload["output_compression"] = output_compression
+
+        response = requests.post(
+            build_openai_url(self.base_url, "/images/generations"),
+            json=payload,
+            headers=self._headers,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return decode_base64_image(response.json()["data"][0]["b64_json"])
+
+    def generate_image_edit(
+        self,
+        *,
+        model: str,
+        prompt: str,
+        images: Image.Image | list[Image.Image],
+        width: int,
+        height: int,
+        num_inference_steps: int = 20,
+        guidance_scale: float | None = None,
+        seed: int | None = None,
+        negative_prompt: str | None = None,
+        output_compression: int | None = None,
+    ) -> Image.Image:
+        if not isinstance(images, list):
+            images = [images]
+        data: dict[str, Any] = {
+            "model": model,
+            "prompt": prompt,
+            "n": 1,
+            "size": f"{width}x{height}",
+            "response_format": "b64_json",
+            "num_inference_steps": str(num_inference_steps),
+        }
+        if guidance_scale is not None:
+            data["guidance_scale"] = str(guidance_scale)
+        if seed is not None:
+            data["seed"] = str(seed)
+        if negative_prompt:
+            data["negative_prompt"] = negative_prompt
+        if output_compression is not None:
+            data["output_compression"] = str(output_compression)
+
+        files = [
+            (
+                "image[]" if len(images) > 1 else "image",
+                (f"image_{index}.png", pil_to_png_bytes(image), "image/png"),
+            )
+            for index, image in enumerate(images)
+        ]
+
+        edit_paths = ["/images/edits", "/images/edit"]
+        last_response: requests.Response | None = None
+        for api_path in edit_paths:
+            response = requests.post(
+                build_openai_url(self.base_url, api_path),
+                data=data,
+                files=files,
+                headers={"Authorization": f"Bearer {self.api_key}"},
+                timeout=self.timeout,
+            )
+            last_response = response
+            if response.status_code != 404:
+                response.raise_for_status()
+                return decode_base64_image(response.json()["data"][0]["b64_json"])
+
+        assert last_response is not None
+        last_response.raise_for_status()
+        raise ValueError("No image payload returned from image edit endpoint")

benchmarks/accuracy/image_to_image/README.md

@@ -0,0 +1,103 @@
+# GEdit-Bench on vLLM-Omni
+
+This integration adapts the upstream `stepfun-ai/Step1X-Edit/GEdit-Bench`
+evaluation flow into `vllm-omni/benchmarks/accuracy/image_to_image`.
+
+Upstream mapping:
+
+- `run_gedit_score.py` -> `run_gedit_bench.py evaluate`
+- `calculate_statistics.py` -> `run_gedit_bench.py summarize`
+- upstream output layout under `results/<method>/fullset/...` is preserved
+
+What changed:
+
+- The upstream repo mainly ships evaluation scripts. This integration adds a
+  generation runner that uses the local `vllm-omni` OpenAI-compatible
+  `/v1/images/edits` endpoint to produce benchmark outputs in the expected
+  directory structure.
+- The evaluator keeps the same VIEScore-style decomposition:
+  - `semantics_score`
+  - `quality_score`
+  - `overall_score = sqrt(semantics_score * quality_score)`
+- Judge calls are routed to a local OpenAI-compatible model served by
+  `vllm-omni`, not a remote provider.
+
+Dataset:
+
+- Default `--dataset-ref` is `stepfun-ai/GEdit-Bench`
+- You can also pass a local dataset directory previously saved with
+  Hugging Face `datasets`
+
+Example usage:
+
+```bash
+python benchmarks/accuracy/image_to_image/run_gedit_bench.py generate \
+  --output-root benchmarks/accuracy/image_to_image/results \
+  --base-url http://127.0.0.1:8000 \
+  --model Qwen/Qwen-Image-Edit \
+  --model-name qwen_image_edit \
+  --dataset-ref stepfun-ai/GEdit-Bench \
+  --task-type all \
+  --instruction-language en
+```
+
+```bash
+python benchmarks/accuracy/image_to_image/run_gedit_bench.py evaluate \
+  --output-root benchmarks/accuracy/image_to_image/results \
+  --model-name qwen_image_edit \
+  --save-dir benchmarks/accuracy/image_to_image/scores \
+  --dataset-ref stepfun-ai/GEdit-Bench \
+  --judge-base-url http://127.0.0.1:8000 \
+  --judge-model Qwen/Qwen2.5-VL-7B-Instruct \
+  --judge-api-key EMPTY
+```
+
+```bash
+python benchmarks/accuracy/image_to_image/run_gedit_bench.py summarize \
+  --csv-path benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_vie_score.csv \
+  --language en
+```
+
+Example summary output:
+
+```json
+{
+  "language": "all",
+  "languages": {
+    "en": {
+      "overall": {"count": 110, "Q_SC": 6.58, "Q_PQ": 5.89, "Q_O": 5.86},
+      "intersection": {"count": 78, "Q_SC": 6.50, "Q_PQ": 5.66, "Q_O": 5.65}
+    },
+    "cn": {
+      "overall": {"count": 110, "Q_SC": 6.90, "Q_PQ": 5.78, "Q_O": 6.11},
+      "intersection": {"count": 63, "Q_SC": 7.22, "Q_PQ": 5.59, "Q_O": 6.28}
+    }
+  }
+}
+```
+
+Example generated images to inspect:
+
+- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/background_change/en/<sample>.png`
+- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/text_change/en/<sample>.png`
+- `benchmarks/accuracy/image_to_image/results/qwen_image_edit/fullset/subject-replace/cn/<sample>.png`
+
+Example score artifacts to inspect together with the images:
+
+- `benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_vie_score.csv`
+- `benchmarks/accuracy/image_to_image/scores/qwen_image_edit_all_all_summary.json`
+
+What to expect:
+
+- `Q_SC` measures instruction following and content preservation.
+- `Q_PQ` measures image naturalness and artifact quality.
+- `Q_O` is the combined overall score; higher is better.
+- `overall.count` is the number of evaluated samples for that language, while `intersection.count` is the subset with `intersection_exist == True`.
+
+Notes:
+
+- This flow requires the optional Hugging Face `datasets` package.
+- `generate` writes `generation_manifest.json` with local output coverage.
+- The current repo marker set exposes `L4` but not `L5`, so if you promote an
+  end-to-end smoke test into CI, use the existing `advanced_model`, `benchmark`,
+  and `L4` markers or introduce a new repo-wide marker explicitly first.

benchmarks/accuracy/image_to_image/__init__.py

@@ -0,0 +1 @@
+"""GEdit-Bench integration for vLLM-Omni."""

benchmarks/accuracy/image_to_image/gedit_bench.py

@@ -0,0 +1,718 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import logging
+import math
+import statistics
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timezone
+from io import BytesIO
+from pathlib import Path
+from typing import Any
+
+import requests
+from PIL import Image
+from tqdm.auto import tqdm
+
+from benchmarks.accuracy.common import (
+    VllmOmniImageClient,
+    build_openai_url,
+    extract_json_object,
+    write_json,
+)
+
+GROUPS = [
+    "background_change",
+    "color_alter",
+    "material_alter",
+    "motion_change",
+    "ps_human",
+    "style_change",
+    "subject-add",
+    "subject-remove",
+    "subject-replace",
+    "text_change",
+    "tone_transfer",
+]
+DEFAULT_SAMPLES_PER_GROUP = 10
+logger = logging.getLogger(__name__)
+
+
+def infer_model_name(model: str) -> str:
+    normalized = model.rstrip("/\\")
+    name = Path(normalized).name
+    return name or normalized
+
+
+def resolve_model_name(*, model_name: str | None, model: str | None = None, output_root: Path | None = None) -> str:
+    if model_name:
+        return model_name
+    if model:
+        return infer_model_name(model)
+    if output_root is not None:
+        candidates = sorted(path.name for path in output_root.iterdir() if path.is_dir())
+        if len(candidates) == 1:
+            return candidates[0]
+        if not candidates:
+            raise ValueError(f"Could not infer model-name from empty output root: {output_root}")
+        raise ValueError(
+            f"Could not infer model-name from output root {output_root}; multiple candidates found: {candidates}"
+        )
+    raise ValueError("model-name is required when it cannot be inferred from model or output-root")
+
+
+def parse_score_payload(raw_text: str) -> dict[str, Any]:
+    try:
+        parsed = extract_json_object(raw_text)
+    except Exception:
+        stripped = raw_text.strip()
+        if stripped.startswith("[") and stripped.endswith("]"):
+            parsed = {"score": json.loads(stripped), "reasoning": ""}
+        else:
+            raise
+    score = parsed.get("score", [])
+    if not isinstance(score, list):
+        score = [score]
+    parsed["score"] = [int(value) for value in score]
+    return parsed
+
+
+def _utc_timestamp() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+def summarize_generated_records(records: list[dict[str, Any]]) -> dict[str, Any]:
+    by_task: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    by_language: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for record in records:
+        by_task[record["task_type"]].append(record)
+        by_language[record["instruction_language"]].append(record)
+
+    return {
+        "count": len(records),
+        "by_task": {
+            task_type: {
+                "count": len(rows),
+                "samples": sorted(row["key"] for row in rows),
+            }
+            for task_type, rows in sorted(by_task.items())
+        },
+        "by_language": {
+            language: {
+                "count": len(rows),
+                "samples": sorted(row["key"] for row in rows),
+            }
+            for language, rows in sorted(by_language.items())
+        },
+    }
+
+
+def select_balanced_gedit_rows(
+    rows: list[dict[str, Any]],
+    *,
+    task_type: str = "all",
+    instruction_language: str = "all",
+    samples_per_group: int | None,
+) -> list[dict[str, Any]]:
+    filtered_rows = []
+    for row in rows:
+        if task_type != "all" and row["task_type"] != task_type:
+            continue
+        if instruction_language != "all" and row["instruction_language"] != instruction_language:
+            continue
+        filtered_rows.append(row)
+
+    if samples_per_group is None:
+        return filtered_rows
+
+    def _select_rows_for_group(group_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        if instruction_language != "all":
+            return group_rows[:samples_per_group]
+
+        rows_by_language: dict[str, list[dict[str, Any]]] = defaultdict(list)
+        for row in group_rows:
+            rows_by_language[str(row.get("instruction_language", "")).strip()].append(row)
+
+        ordered_languages = ["en", "cn"]
+        per_language_quota = samples_per_group // max(len(ordered_languages), 1)
+        remainder = samples_per_group % max(len(ordered_languages), 1)
+
+        selected_rows: list[dict[str, Any]] = []
+        leftovers: list[dict[str, Any]] = []
+        for index, language in enumerate(ordered_languages):
+            language_rows = rows_by_language.get(language, [])
+            quota = per_language_quota + (1 if index < remainder else 0)
+            selected_rows.extend(language_rows[:quota])
+            leftovers.extend(language_rows[quota:])
+
+        if len(selected_rows) < samples_per_group:
+            selected_rows.extend(leftovers[: samples_per_group - len(selected_rows)])
+
+        return selected_rows
+
+    if task_type != "all":
+        return _select_rows_for_group(filtered_rows)
+
+    grouped_rows: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in filtered_rows:
+        grouped_rows[row["task_type"]].append(row)
+
+    selected: list[dict[str, Any]] = []
+    for group in GROUPS:
+        selected.extend(_select_rows_for_group(grouped_rows.get(group, [])))
+    return selected
+
+
+def summarize_gedit_rows(rows: list[dict[str, Any]], language: str = "all") -> dict[str, Any]:
+    return summarize_gedit_rows_with_backbone(rows, language=language)
+
+
+def _mean_or_none(values: list[float]) -> float | None:
+    return statistics.fmean(values) if values else None
+
+
+def _to_q_metrics(section: dict[str, float | None]) -> dict[str, float | None]:
+    return {
+        "count": section.get("count"),
+        "Q_SC": section.get("avg_semantics"),
+        "Q_PQ": section.get("avg_quality"),
+        "Q_O": section.get("avg_overall"),
+    }
+
+
+def _summarize_gedit_rows_single_language(
+    rows: list[dict[str, Any]],
+    *,
+    language: str,
+) -> dict[str, Any]:
+    filtered_rows = [row for row in rows if str(row.get("instruction_language", "")).strip() == language]
+
+    def _to_bool(value: Any) -> bool:
+        if isinstance(value, bool):
+            return value
+        return str(value).strip().lower() in {"1", "true", "yes"}
+
+    per_group: dict[str, dict[str, float | None]] = {}
+    intersection_group: dict[str, dict[str, float | None]] = {}
+
+    for group in GROUPS:
+        group_rows = [row for row in filtered_rows if row.get("task_type") == group]
+        semantics = [float(row["semantics_score"]) for row in group_rows]
+        quality = [float(row["quality_score"]) for row in group_rows]
+        overall = [math.sqrt(float(row["semantics_score"]) * float(row["quality_score"])) for row in group_rows]
+
+        per_group[group] = {
+            "count": len(group_rows),
+            "avg_semantics": _mean_or_none(semantics),
+            "avg_quality": _mean_or_none(quality),
+            "avg_overall": _mean_or_none(overall),
+        }
+
+        intersection_rows = [row for row in group_rows if _to_bool(row.get("intersection_exist", False))]
+        intersection_semantics = [float(row["semantics_score"]) for row in intersection_rows]
+        intersection_quality = [float(row["quality_score"]) for row in intersection_rows]
+        intersection_overall = [
+            math.sqrt(float(row["semantics_score"]) * float(row["quality_score"])) for row in intersection_rows
+        ]
+        intersection_group[group] = {
+            "count": len(intersection_rows),
+            "avg_semantics": _mean_or_none(intersection_semantics),
+            "avg_quality": _mean_or_none(intersection_quality),
+            "avg_overall": _mean_or_none(intersection_overall),
+        }
+
+    overall_section = {
+        "count": len(filtered_rows),
+        "avg_semantics": _mean_or_none(
+            [score["avg_semantics"] for score in per_group.values() if score["avg_semantics"] is not None]
+        ),
+        "avg_quality": _mean_or_none(
+            [score["avg_quality"] for score in per_group.values() if score["avg_quality"] is not None]
+        ),
+        "avg_overall": _mean_or_none(
+            [score["avg_overall"] for score in per_group.values() if score["avg_overall"] is not None]
+        ),
+    }
+    intersection_section = {
+        "count": sum(score["count"] for score in intersection_group.values()),
+        "avg_semantics": _mean_or_none(
+            [score["avg_semantics"] for score in intersection_group.values() if score["avg_semantics"] is not None]
+        ),
+        "avg_quality": _mean_or_none(
+            [score["avg_quality"] for score in intersection_group.values() if score["avg_quality"] is not None]
+        ),
+        "avg_overall": _mean_or_none(
+            [score["avg_overall"] for score in intersection_group.values() if score["avg_overall"] is not None]
+        ),
+    }
+
+    return {
+        "language": language,
+        "by_group": {group: _to_q_metrics(section) for group, section in per_group.items()},
+        "overall": _to_q_metrics(overall_section),
+        "intersection": _to_q_metrics(intersection_section),
+    }
+
+
+def summarize_gedit_rows_with_backbone(
+    rows: list[dict[str, Any]],
+    *,
+    language: str = "all",
+) -> dict[str, Any]:
+    if language == "all":
+        return {
+            "language": "all",
+            "languages": {
+                single_language: _summarize_gedit_rows_single_language(
+                    rows,
+                    language=single_language,
+                )
+                for single_language in ["en", "cn"]
+            },
+        }
+
+    return _summarize_gedit_rows_single_language(rows, language=language)
+
+
+def _require_datasets():
+    try:
+        from datasets import load_dataset, load_from_disk
+    except ImportError as exc:
+        raise ImportError("GEdit-Bench requires the optional `datasets` package.") from exc
+    return load_dataset, load_from_disk
+
+
+def _load_gedit_dataset(dataset_ref: str):
+    load_dataset, load_from_disk = _require_datasets()
+    dataset_path = Path(dataset_ref)
+    if dataset_path.exists():
+        if (dataset_path / "state.json").exists() and (dataset_path / "dataset_info.json").exists():
+            return load_from_disk(str(dataset_path))
+        return load_dataset(str(dataset_path))
+    return load_dataset(dataset_ref)
+
+
+def _resolve_gedit_split(dataset_obj: Any) -> Any:
+    if isinstance(dataset_obj, dict):
+        if "train" in dataset_obj:
+            return dataset_obj["train"]
+        return dataset_obj
+    try:
+        return dataset_obj["train"]
+    except Exception:
+        return dataset_obj
+
+
+def _to_pil_image(value: Any) -> Image.Image:
+    if isinstance(value, Image.Image):
+        return value.convert("RGB")
+    if isinstance(value, dict) and "bytes" in value:
+        image = Image.open(BytesIO(value["bytes"]))
+        image.load()
+        return image.convert("RGB")
+    raise TypeError(f"Unsupported image payload type: {type(value)!r}")
+
+
+class LocalVIEScorer:
+    def __init__(self, *, base_url: str, api_key: str, model: str, timeout: int = 600):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self.timeout = timeout
+        self.sc_prompt = (
+            "You are evaluating image editing quality.\n"
+            "Two images are provided: the source image and the edited image.\n"
+            'Return JSON only in the format {"score": [edit_success, content_preservation], '
+            '"reasoning": "..."}.\n'
+            "Each score must be an integer from 0 to 10.\n"
+            "Editing instruction: <instruction>"
+        )
+        self.pq_prompt = (
+            "You are evaluating image quality.\n"
+            'Return JSON only in the format {"score": [naturalness, artifact_free], "reasoning": "..."}.\n'
+            "Each score must be an integer from 0 to 10."
+        )
+
+    def _request(self, prompt: str, images: list[Image.Image]) -> dict[str, Any]:
+        from benchmarks.accuracy.common import pil_to_data_url
+
+        content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
+        for image in images:
+            content.append({"type": "image_url", "image_url": {"url": pil_to_data_url(image)}})
+
+        response = requests.post(
+            build_openai_url(self.base_url, "/chat/completions"),
+            json={
+                "model": self.model,
+                "messages": [{"role": "user", "content": content}],
+                "temperature": 0,
+            },
+            headers={
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+            },
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        message_content = response.json()["choices"][0]["message"]["content"]
+        if isinstance(message_content, list):
+            text = "\n".join(part.get("text", "") for part in message_content if part.get("type") == "text")
+        else:
+            text = str(message_content)
+        return parse_score_payload(text)
+
+    def evaluate(self, source_image: Image.Image, edited_image: Image.Image, instruction: str) -> dict[str, float]:
+        sc_payload = self._request(self.sc_prompt.replace("<instruction>", instruction), [source_image, edited_image])
+        pq_payload = self._request(self.pq_prompt, [edited_image])
+        semantics = float(min(sc_payload["score"])) if sc_payload["score"] else 0.0
+        quality = float(min(pq_payload["score"])) if pq_payload["score"] else 0.0
+        overall = math.sqrt(semantics * quality)
+        return {
+            "semantics_score": semantics,
+            "quality_score": quality,
+            "overall_score": overall,
+        }
+
+
+class GEditBenchRunner:
+    def __init__(
+        self,
+        *,
+        dataset_ref: str,
+        output_root: Path,
+        base_url: str,
+        model: str,
+        api_key: str = "EMPTY",
+        width: int = 512,
+        height: int = 512,
+        num_inference_steps: int = 20,
+        guidance_scale: float | None = None,
+        seed: int | None = 42,
+    ):
+        self.dataset_ref = dataset_ref
+        self.output_root = output_root
+        self.model = model
+        self.width = width
+        self.height = height
+        self.num_inference_steps = num_inference_steps
+        self.guidance_scale = guidance_scale
+        self.seed = seed
+        self.client = VllmOmniImageClient(base_url=base_url, api_key=api_key)
+
+    def generate(
+        self,
+        *,
+        model_name: str,
+        task_type: str = "all",
+        instruction_language: str = "all",
+        workers: int = 1,
+        max_samples: int | None = None,
+        samples_per_group: int | None = None,
+    ) -> list[dict[str, Any]]:
+        dataset = _resolve_gedit_split(_load_gedit_dataset(self.dataset_ref))
+        rows = select_balanced_gedit_rows(
+            list(dataset),
+            task_type=task_type,
+            instruction_language=instruction_language,
+            samples_per_group=samples_per_group,
+        )
+        if max_samples is not None:
+            rows = rows[:max_samples]
+
+        outputs: list[dict[str, Any]] = []
+        total = len(rows)
+        if workers <= 1:
+            with tqdm(total=total, desc="GEdit generate", unit="sample") as progress:
+                for item in rows:
+                    result = self._safe_generate_one(model_name, item)
+                    if result:
+                        outputs.append(result)
+                    progress.update(1)
+            return outputs
+
+        with tqdm(total=total, desc="GEdit generate", unit="sample") as progress:
+            with ThreadPoolExecutor(max_workers=workers) as executor:
+                futures = [executor.submit(self._safe_generate_one, model_name, item) for item in rows]
+                for future in as_completed(futures):
+                    result = future.result()
+                    if result:
+                        outputs.append(result)
+                    progress.update(1)
+        return outputs
+
+    def _safe_generate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None:
+        try:
+            return self._generate_one(model_name, item)
+        except Exception:
+            logger.exception("Failed to generate GEdit-Bench sample %s", item.get("key", "<unknown>"))
+            return None
+
+    def _generate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None:
+        output_path = (
+            self.output_root
+            / model_name
+            / "fullset"
+            / item["task_type"]
+            / item["instruction_language"]
+            / f"{item['key']}.png"
+        )
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        if output_path.exists():
+            return {
+                "task_type": item["task_type"],
+                "instruction_language": item["instruction_language"],
+                "key": item["key"],
+                "output_path": str(output_path),
+            }
+
+        source_image = _to_pil_image(item["input_image_raw"])
+        edited_image = self.client.generate_image_edit(
+            model=self.model,
+            prompt=item["instruction"],
+            images=source_image,
+            width=self.width,
+            height=self.height,
+            num_inference_steps=self.num_inference_steps,
+            guidance_scale=self.guidance_scale,
+            seed=self.seed,
+        )
+        edited_image.save(output_path)
+        return {
+            "task_type": item["task_type"],
+            "instruction_language": item["instruction_language"],
+            "key": item["key"],
+            "output_path": str(output_path),
+        }
+
+
+class GEditBenchEvaluator:
+    def __init__(self, *, dataset_ref: str, output_root: Path, scorer: LocalVIEScorer):
+        self.dataset_ref = dataset_ref
+        self.output_root = output_root
+        self.scorer = scorer
+
+    def evaluate(
+        self,
+        *,
+        model_name: str,
+        save_dir: Path,
+        task_type: str = "all",
+        instruction_language: str = "all",
+        workers: int = 1,
+        max_samples: int | None = None,
+        samples_per_group: int | None = None,
+    ) -> dict[str, Any]:
+        dataset = _resolve_gedit_split(_load_gedit_dataset(self.dataset_ref))
+        rows = select_balanced_gedit_rows(
+            list(dataset),
+            task_type=task_type,
+            instruction_language=instruction_language,
+            samples_per_group=samples_per_group,
+        )
+        if max_samples is not None:
+            rows = rows[:max_samples]
+
+        results: list[dict[str, Any]] = []
+        total = len(rows)
+        if workers <= 1:
+            with tqdm(total=total, desc="GEdit evaluate", unit="sample") as progress:
+                for item in rows:
+                    result = self._safe_evaluate_one(model_name, item)
+                    if result:
+                        results.append(result)
+                    progress.update(1)
+        else:
+            with tqdm(total=total, desc="GEdit evaluate", unit="sample") as progress:
+                with ThreadPoolExecutor(max_workers=workers) as executor:
+                    futures = [executor.submit(self._safe_evaluate_one, model_name, item) for item in rows]
+                    for future in as_completed(futures):
+                        result = future.result()
+                        if result:
+                            results.append(result)
+                        progress.update(1)
+
+        save_dir.mkdir(parents=True, exist_ok=True)
+        base_name = f"{model_name}_{task_type}_{instruction_language}"
+        timestamp = _utc_timestamp()
+        csv_path = save_dir / f"{base_name}_vie_score.csv"
+        timestamped_csv_path = save_dir / f"{base_name}_vie_score_{timestamp}.csv"
+
+        def _write_csv(path: Path) -> None:
+            with path.open("w", encoding="utf-8", newline="") as handle:
+                writer = csv.DictWriter(
+                    handle,
+                    fieldnames=[
+                        "key",
+                        "task_type",
+                        "edited_image",
+                        "instruction",
+                        "semantics_score",
+                        "quality_score",
+                        "overall_score",
+                        "intersection_exist",
+                        "instruction_language",
+                    ],
+                )
+                writer.writeheader()
+                for row in results:
+                    writer.writerow(row)
+
+        _write_csv(csv_path)
+        _write_csv(timestamped_csv_path)
+
+        summary = summarize_gedit_rows_with_backbone(
+            results,
+            language=instruction_language,
+        )
+        summary_path = save_dir / f"{base_name}_summary.json"
+        timestamped_summary_path = save_dir / f"{base_name}_summary_{timestamp}.json"
+        write_json(summary_path, summary)
+        write_json(timestamped_summary_path, summary)
+        return {
+            "results": results,
+            "summary": summary,
+            "csv_path": str(csv_path),
+            "summary_path": str(summary_path),
+            "timestamped_csv_path": str(timestamped_csv_path),
+            "timestamped_summary_path": str(timestamped_summary_path),
+        }
+
+    def _safe_evaluate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None:
+        try:
+            return self._evaluate_one(model_name, item)
+        except Exception:
+            logger.exception("Failed to evaluate GEdit-Bench sample %s", item.get("key", "<unknown>"))
+            return None
+
+    def _evaluate_one(self, model_name: str, item: dict[str, Any]) -> dict[str, Any] | None:
+        edited_image_path = (
+            self.output_root
+            / model_name
+            / "fullset"
+            / item["task_type"]
+            / item["instruction_language"]
+            / f"{item['key']}.png"
+        )
+        if not edited_image_path.exists():
+            return None
+
+        source_image = _to_pil_image(item["input_image_raw"])
+        edited_image = Image.open(edited_image_path).convert("RGB")
+        scores = self.scorer.evaluate(source_image, edited_image, item["instruction"])
+        return {
+            "key": item["key"],
+            "task_type": item["task_type"],
+            "edited_image": str(edited_image_path),
+            "instruction": item["instruction"],
+            "semantics_score": scores["semantics_score"],
+            "quality_score": scores["quality_score"],
+            "overall_score": scores["overall_score"],
+            "intersection_exist": item.get("Intersection_exist", False),
+            "instruction_language": item["instruction_language"],
+        }
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Run the GEdit-Bench integration against a local vLLM-Omni server.")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    generate = subparsers.add_parser("generate")
+    generate.add_argument("--dataset-ref", type=str, default="stepfun-ai/GEdit-Bench")
+    generate.add_argument("--output-root", type=Path, required=True)
+    generate.add_argument("--base-url", type=str, required=True)
+    generate.add_argument("--model", type=str, required=True)
+    generate.add_argument("--model-name", type=str, default=None)
+    generate.add_argument("--api-key", type=str, default="EMPTY")
+    generate.add_argument("--task-type", choices=["all", *GROUPS], default="all")
+    generate.add_argument("--instruction-language", choices=["all", "en", "cn"], default="all")
+    generate.add_argument("--width", type=int, default=512)
+    generate.add_argument("--height", type=int, default=512)
+    generate.add_argument("--num-inference-steps", type=int, default=20)
+    generate.add_argument("--guidance-scale", type=float, default=None)
+    generate.add_argument("--seed", type=int, default=42)
+    generate.add_argument("--workers", type=int, default=1)
+    generate.add_argument("--max-samples", type=int, default=None)
+    generate.add_argument("--samples-per-group", type=int, default=None)
+
+    evaluate = subparsers.add_parser("evaluate")
+    evaluate.add_argument("--dataset-ref", type=str, default="stepfun-ai/GEdit-Bench")
+    evaluate.add_argument("--output-root", type=Path, required=True)
+    evaluate.add_argument("--model-name", type=str, default=None)
+    evaluate.add_argument("--save-dir", type=Path, required=True)
+    evaluate.add_argument("--task-type", choices=["all", *GROUPS], default="all")
+    evaluate.add_argument("--instruction-language", choices=["all", "en", "cn"], default="all")
+    evaluate.add_argument("--judge-base-url", type=str, required=True)
+    evaluate.add_argument("--judge-model", type=str, required=True)
+    evaluate.add_argument("--judge-api-key", type=str, default="EMPTY")
+    evaluate.add_argument("--workers", type=int, default=1)
+    evaluate.add_argument("--max-samples", type=int, default=None)
+    evaluate.add_argument("--samples-per-group", type=int, default=None)
+
+    summarize = subparsers.add_parser("summarize")
+    summarize.add_argument("--csv-path", type=Path, required=True)
+    summarize.add_argument("--language", choices=["all", "en", "cn"], default="all")
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+
+    if args.command == "generate":
+        model_name = resolve_model_name(model_name=args.model_name, model=args.model)
+        runner = GEditBenchRunner(
+            dataset_ref=args.dataset_ref,
+            output_root=args.output_root,
+            base_url=args.base_url,
+            model=args.model,
+            api_key=args.api_key,
+            width=args.width,
+            height=args.height,
+            num_inference_steps=args.num_inference_steps,
+            guidance_scale=args.guidance_scale,
+            seed=args.seed,
+        )
+        records = runner.generate(
+            model_name=model_name,
+            task_type=args.task_type,
+            instruction_language=args.instruction_language,
+            workers=args.workers,
+            max_samples=args.max_samples,
+            samples_per_group=args.samples_per_group,
+        )
+        payload = {"records": records, "summary": summarize_generated_records(records)}
+        write_json(args.output_root / model_name / "generation_manifest.json", payload)
+        return 0
+
+    if args.command == "evaluate":
+        model_name = resolve_model_name(model_name=args.model_name, output_root=args.output_root)
+        scorer = LocalVIEScorer(
+            base_url=args.judge_base_url,
+            api_key=args.judge_api_key,
+            model=args.judge_model,
+        )
+        evaluator = GEditBenchEvaluator(dataset_ref=args.dataset_ref, output_root=args.output_root, scorer=scorer)
+        evaluator.evaluate(
+            model_name=model_name,
+            save_dir=args.save_dir,
+            task_type=args.task_type,
+            instruction_language=args.instruction_language,
+            workers=args.workers,
+            max_samples=args.max_samples,
+            samples_per_group=args.samples_per_group,
+        )
+        return 0
+
+    if args.command == "summarize":
+        with args.csv_path.open("r", encoding="utf-8", newline="") as handle:
+            rows = list(csv.DictReader(handle))
+        summary = summarize_gedit_rows_with_backbone(rows, language=args.language)
+        print(json.dumps(summary, indent=2, ensure_ascii=False))
+        return 0
+
+    parser.error(f"Unknown command: {args.command}")
+    return 1

benchmarks/accuracy/image_to_image/run_gedit_bench.py

@@ -0,0 +1,13 @@
+# ruff: noqa: E402, I001
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from benchmarks.accuracy.image_to_image.gedit_bench import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/accuracy/text_to_image/README.md

@@ -0,0 +1,103 @@
+# GEBench on vLLM-Omni
+
+This integration adapts the upstream `stepfun-ai/GEBench` scripts into
+`vllm-omni/benchmarks/accuracy/text_to_image`.
+
+Upstream mapping:
+
+- `scripts/generate.py` -> `run_gebench.py generate`
+- `scripts/evaluate.py` -> `run_gebench.py evaluate`
+- upstream prompt / judge logic -> `gbench.py`
+
+What changed:
+
+- Generation calls the local OpenAI-compatible `vllm-omni` endpoints:
+  - `/v1/images/generations` for text-only frame generation
+  - `/v1/images/edits` for image-conditioned GUI transition generation
+- Evaluation still keeps the GEBench scoring dimensions:
+  - `goal`
+  - `logic`
+  - `cons`
+  - `ui`
+  - `qual`
+- Judge calls are also routed to a local OpenAI-compatible model served by
+  `vllm-omni` instead of a remote service.
+
+Expected dataset layout:
+
+- Clone the benchmark dataset from Hugging Face into a local directory:
+
+```bash
+git clone https://huggingface.co/datasets/stepfun-ai/GEBench /path/to/GEBench
+```
+
+Example usage:
+
+```bash
+python benchmarks/accuracy/text_to_image/run_gebench.py generate \
+  --dataset-root /path/to/GEBench \
+  --output-root benchmarks/accuracy/text_to_image/outputs \
+  --base-url http://127.0.0.1:8000 \
+  --model Tongyi-MAI/Z-Image-Turbo \
+  --data-type type3
+```
+
+```bash
+python benchmarks/accuracy/text_to_image/run_gebench.py evaluate \
+  --dataset-root /path/to/GEBench \
+  --output-root benchmarks/accuracy/text_to_image/outputs \
+  --data-type type3 \
+  --judge-base-url http://127.0.0.1:8000 \
+  --judge-model Qwen/Qwen2.5-VL-7B-Instruct \
+  --judge-api-key EMPTY
+```
+
+```bash
+python benchmarks/accuracy/text_to_image/run_gebench.py summarize \
+  --output-root benchmarks/accuracy/text_to_image/outputs
+```
+
+Example summary output:
+
+```json
+{
+  "generation": {
+    "count": 20,
+    "by_type": {
+      "type3": {"count": 10},
+      "type4": {"count": 10}
+    }
+  },
+  "evaluation": {
+    "count": 20,
+    "overall_mean": 0.52,
+    "by_type": {
+      "type3": {"count": 10, "overall_mean": 0.50, "overall_mean_100": 50.0},
+      "type4": {"count": 10, "overall_mean": 0.54, "overall_mean_100": 54.0}
+    }
+  }
+}
+```
+
+Example generated images to inspect:
+
+- `benchmarks/accuracy/text_to_image/outputs/03_trajectory_text_fictionalapp/<lang_device>/<sample>/frame0.png`
+- `benchmarks/accuracy/text_to_image/outputs/03_trajectory_text_fictionalapp/<lang_device>/<sample>/frame5.png`
+- `benchmarks/accuracy/text_to_image/outputs/04_trajectory_text_realapp/<lang_device>/<sample>/frame0.png`
+- `benchmarks/accuracy/text_to_image/outputs/04_trajectory_text_realapp/<lang_device>/<sample>/frame5.png`
+
+What to expect:
+
+- `overall_mean` is normalized to `0.0 ~ 1.0`; higher is better.
+- `frame0.png` is the initial GUI frame; `frame5.png` is the final trajectory frame most often used for quick inspection.
+- For full debugging, inspect the whole `frame0.png` ... `frame5.png` sequence for one sample directory.
+
+Notes:
+
+- GEBench upstream leaves type3/type4 generation unfinished. This integration
+  fills that gap with a trajectory runner that generates `frame0.png` followed
+  by `frame1.png` ... `frame5.png`.
+- Type1/2/5 require an image-edit capable model exposed through
+  `vllm-omni serve`.
+- `summarize` will report both generated coverage and any existing evaluation
+  summary files.

benchmarks/accuracy/text_to_image/__init__.py

@@ -0,0 +1 @@
+"""GEBench integration for vLLM-Omni."""

benchmarks/accuracy/text_to_image/gbench.py

@@ -0,0 +1,927 @@
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import requests
+from PIL import Image
+
+from benchmarks.accuracy.common import (
+    VllmOmniImageClient,
+    build_openai_url,
+    ensure_dir,
+    extract_json_object,
+    find_first_image,
+    load_json,
+    pil_to_data_url,
+    save_image,
+    write_json,
+)
+
+TYPE_TO_FOLDER = {
+    "type1": "01_single_step",
+    "type2": "02_multi_step",
+    "type3": "03_trajectory_text_fictionalapp",
+    "type4": "04_trajectory_text_realapp",
+    "type5": "05_grounding_data",
+}
+SCORE_KEYS = ("goal", "logic", "cons", "ui", "qual")
+DEFAULT_SAMPLES_PER_TYPE = 10
+
+
+def _utc_timestamp() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+def _write_json_with_timestamp(path: Path, payload: dict[str, Any]) -> Path:
+    write_json(path, payload)
+    timestamped_path = path.with_name(f"{path.stem}_{_utc_timestamp()}{path.suffix}")
+    write_json(timestamped_path, payload)
+    return timestamped_path
+
+
+@dataclass(frozen=True)
+class GEBenchSampleSpec:
+    sample_path: Path
+    metadata: dict[str, Any]
+    sample_name: str
+    lang_device: str
+
+
+def summarize_generated_records(records: list[dict[str, Any]]) -> dict[str, Any]:
+    by_type: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for record in records:
+        by_type[record["data_type"]].append(record)
+
+    return {
+        "count": len(records),
+        "by_type": {
+            data_type: {
+                "count": len(rows),
+            }
+            for data_type, rows in sorted(by_type.items())
+        },
+    }
+
+
+def summarize_gebench_results(results: list[dict[str, Any]]) -> dict[str, Any]:
+    by_type: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for result in results:
+        by_type[result["data_type"]].append(result)
+
+    summary: dict[str, Any] = {
+        "count": len(results),
+        "overall_mean": statistics.fmean(r["overall"] for r in results) if results else 0.0,
+        "by_type": {},
+    }
+    for data_type, rows in by_type.items():
+        score_means: dict[str, float] = {}
+        all_score_keys = {key for row in rows for key in row.get("scores", {}).keys()}
+        for score_key in all_score_keys:
+            values = [row["scores"][score_key] for row in rows if score_key in row.get("scores", {})]
+            score_means[score_key] = statistics.fmean(values) if values else 0.0
+        overall_mean = statistics.fmean(row["overall"] for row in rows)
+        summary["by_type"][data_type] = {
+            "count": len(rows),
+            "overall_mean": overall_mean,
+            "overall_mean_100": overall_mean * 100.0,
+            "score_means": score_means,
+        }
+    return summary
+
+
+def select_balanced_gebench_samples(
+    sample_paths_by_type: dict[str, list[Any]],
+    *,
+    samples_per_type: int | None,
+) -> dict[str, list[Any]]:
+    if samples_per_type is None:
+        return {data_type: list(paths) for data_type, paths in sample_paths_by_type.items()}
+    return {data_type: list(paths)[:samples_per_type] for data_type, paths in sample_paths_by_type.items()}
+
+
+def collect_gebench_generation_summary(output_root: Path) -> dict[str, Any]:
+    records: list[dict[str, Any]] = []
+    for data_type, folder_name in TYPE_TO_FOLDER.items():
+        type_root = output_root / folder_name
+        if not type_root.exists():
+            continue
+        for lang_dir in sorted(path for path in type_root.iterdir() if path.is_dir()):
+            for sample_dir in sorted(path for path in lang_dir.iterdir() if path.is_dir()):
+                expected = sample_dir / "frame5.png" if data_type in {"type2", "type3", "type4"} else None
+                if expected is None:
+                    expected = find_first_image(sample_dir)
+                elif not expected.exists():
+                    expected = None
+                if expected is None:
+                    continue
+                records.append(
+                    {
+                        "data_type": data_type,
+                        "sample_name": f"{lang_dir.name}/{sample_dir.name}",
+                        "output_path": str(expected),
+                    }
+                )
+    return summarize_generated_records(records)
+
+
+def _normalize_score_key(key: str) -> str:
+    mapping = {
+        "goal": "goal",
+        "logic": "logic",
+        "cons": "cons",
+        "consistency": "cons",
+        "ui": "ui",
+        "qual": "qual",
+        "quality": "qual",
+    }
+    return mapping.get(key.lower(), key.lower())
+
+
+def _normalize_scores(raw_scores: dict[str, Any]) -> dict[str, int]:
+    scores: dict[str, int] = {}
+    for key, value in raw_scores.items():
+        normalized = _normalize_score_key(key)
+        if normalized not in SCORE_KEYS:
+            continue
+        scalar = value.get("s", 0) if isinstance(value, dict) else value
+        try:
+            scores[normalized] = int(scalar)
+        except (TypeError, ValueError):
+            scores[normalized] = 0
+    for key in SCORE_KEYS:
+        scores.setdefault(key, 0)
+    return scores
+
+
+def _compute_overall(scores: dict[str, int]) -> float:
+    return sum(scores.values()) / (len(SCORE_KEYS) * 5.0)
+
+
+def _iter_sample_paths(dataset_root: Path, data_type: str) -> list[Path]:
+    data_dir = dataset_root / TYPE_TO_FOLDER[data_type]
+    if not data_dir.exists():
+        raise FileNotFoundError(f"Dataset path not found: {data_dir}")
+
+    samples: list[Path] = []
+    for lang_dir in sorted(path for path in data_dir.iterdir() if path.is_dir()):
+        for child in sorted(lang_dir.iterdir()):
+            if child.is_dir():
+                samples.append(child)
+            elif child.suffix.lower() == ".json":
+                samples.append(child)
+    return samples
+
+
+def _sample_name_from_metadata(metadata: dict[str, Any], sample_path: Path, item_index: int | None = None) -> str:
+    sample_id = metadata.get("id") or metadata.get("sample_id") or metadata.get("name")
+    if sample_id:
+        return str(sample_id)
+    if item_index is not None:
+        return f"{sample_path.stem}_{item_index:04d}"
+    return sample_path.stem if sample_path.is_file() else sample_path.name
+
+
+def _expand_sample_path(sample_path: Path) -> list[GEBenchSampleSpec]:
+    if sample_path.is_dir():
+        metadata = _load_metadata(sample_path)
+        return [
+            GEBenchSampleSpec(
+                sample_path=sample_path,
+                metadata=metadata,
+                sample_name=_sample_name(sample_path),
+                lang_device=_lang_device(sample_path, metadata),
+            )
+        ]
+
+    payload = load_json(sample_path)
+    if isinstance(payload, dict):
+        return [
+            GEBenchSampleSpec(
+                sample_path=sample_path,
+                metadata=payload,
+                sample_name=_sample_name_from_metadata(payload, sample_path),
+                lang_device=_lang_device(sample_path, payload),
+            )
+        ]
+
+    if isinstance(payload, list):
+        specs: list[GEBenchSampleSpec] = []
+        for item_index, item in enumerate(payload):
+            if not isinstance(item, dict):
+                continue
+            specs.append(
+                GEBenchSampleSpec(
+                    sample_path=sample_path,
+                    metadata=item,
+                    sample_name=_sample_name_from_metadata(item, sample_path, item_index),
+                    lang_device=_lang_device(sample_path, item),
+                )
+            )
+        return specs
+
+    raise TypeError(f"Unsupported metadata payload type for sample {sample_path}: {type(payload)!r}")
+
+
+def _iter_sample_specs(dataset_root: Path, data_type: str) -> list[GEBenchSampleSpec]:
+    sample_specs: list[GEBenchSampleSpec] = []
+    for sample_path in _iter_sample_paths(dataset_root, data_type):
+        sample_specs.extend(_expand_sample_path(sample_path))
+    return sample_specs
+
+
+def _load_metadata(sample_path: Path) -> dict[str, Any]:
+    if sample_path.is_file():
+        return load_json(sample_path)
+    for candidate in ("meta_data.json", "metadata.json"):
+        meta_path = sample_path / candidate
+        if meta_path.exists():
+            return load_json(meta_path)
+    raise FileNotFoundError(f"Metadata not found for sample: {sample_path}")
+
+
+def _sample_name(sample_path: Path) -> str:
+    return sample_path.stem if sample_path.is_file() else sample_path.name
+
+
+def _lang_device(sample_path: Path, metadata: dict[str, Any]) -> str:
+    return str(metadata.get("lang_device") or sample_path.parent.name)
+
+
+def _resolve_referenced_image(
+    *,
+    metadata: dict[str, Any],
+    sample_path: Path,
+    dataset_root: Path,
+    data_type: str,
+) -> Image.Image | None:
+    for key in ("image", "input_image", "initial_image", "reference_image"):
+        image_ref = metadata.get(key)
+        if not image_ref:
+            continue
+        candidate = dataset_root / TYPE_TO_FOLDER[data_type] / str(image_ref)
+        if candidate.exists():
+            image = Image.open(candidate)
+            image.load()
+            return image.convert("RGB")
+    if sample_path.is_dir():
+        local_image = find_first_image(sample_path)
+        if local_image:
+            image = Image.open(local_image)
+            image.load()
+            return image.convert("RGB")
+    return None
+
+
+def _trajectory_steps(metadata: dict[str, Any]) -> list[dict[str, Any]]:
+    for key in ("trajectory", "steps", "frames"):
+        value = metadata.get(key)
+        if isinstance(value, list):
+            return [step for step in value if isinstance(step, dict)]
+    extracted: list[dict[str, Any]] = []
+    for index in range(1, 6):
+        value = metadata.get(f"step{index}") or metadata.get(str(index))
+        if isinstance(value, dict):
+            extracted.append(value)
+    return extracted
+
+
+def _text_or_default(value: Any, default: str = "") -> str:
+    return str(value).strip() if value is not None else default
+
+
+def _type1_prompt(metadata: dict[str, Any]) -> str:
+    caption = _text_or_default(metadata.get("caption") or metadata.get("instruction"), "Transform the reference GUI.")
+    return (
+        "Using the reference GUI screenshot, generate the next GUI state after the requested interaction.\n\n"
+        f"Requested change:\n{caption}\n\n"
+        "Requirements:\n"
+        "- Preserve layout, visual identity, and unrelated regions.\n"
+        "- Only apply the requested state change.\n"
+        "- Keep all text and controls readable.\n"
+    )
+
+
+def _type2_prompt(goal: str, step_num: int) -> str:
+    return (
+        "Generate the next GUI state for a multi-step task.\n\n"
+        f"Overall goal: {goal}\n"
+        f"Current progress step: {step_num}/5\n\n"
+        "Requirements:\n"
+        "- The change should be incremental and plausible.\n"
+        "- Preserve layout and visual identity.\n"
+        "- Make text/buttons readable.\n"
+    )
+
+
+def _type34_initial_prompt(metadata: dict[str, Any], first_step: dict[str, Any]) -> str:
+    app_name = _text_or_default(metadata.get("app_name"), "App")
+    final_goal = _text_or_default(metadata.get("final_goal") or metadata.get("instruction"), "Complete the task.")
+    visual_description = _text_or_default(
+        metadata.get("visual_description") or first_step.get("visual_description") or first_step.get("description"),
+        "A clean product-quality app home screen.",
+    )
+    return (
+        "Generate the first GUI frame for a task trajectory.\n\n"
+        f"App name: {app_name}\n"
+        f"Final goal: {final_goal}\n"
+        f"Visual description:\n{visual_description}\n\n"
+        "Requirements:\n"
+        "- Generate a production-looking UI screenshot only.\n"
+        "- Keep the layout coherent and readable.\n"
+    )
+
+
+def _type34_next_prompt(step_num: int, step_info: dict[str, Any]) -> str:
+    action = _text_or_default(step_info.get("action") or step_info.get("instruction"), "Continue the task.")
+    visual_description = _text_or_default(
+        step_info.get("visual_description") or step_info.get("description"),
+        "Reflect the expected next GUI state.",
+    )
+    return (
+        "Using the previous frame as reference, generate the next GUI frame.\n\n"
+        f"Step {step_num} action: {action}\n"
+        f"Expected visual state:\n{visual_description}\n\n"
+        "Requirements:\n"
+        "- Only change UI regions affected by this action.\n"
+        "- Preserve persistent bars, layout, and style.\n"
+        "- Keep text and icons readable.\n"
+    )
+
+
+def _type5_prompt(metadata: dict[str, Any]) -> str:
+    grounding = metadata.get("grounding") or {}
+    explanation = _text_or_default(
+        metadata.get("grounding_explanation") or grounding.get("effect") or grounding.get("description"),
+        "Predict the immediate GUI reaction to the indicated target.",
+    )
+    return (
+        "Using the reference GUI screenshot, predict the immediate GUI state after the grounded interaction.\n\n"
+        f"Expected effect: {explanation}\n"
+        f"Grounding metadata: {json.dumps(grounding, ensure_ascii=False)}\n\n"
+        "Requirements:\n"
+        "- Apply only the interaction-triggered change.\n"
+        "- Preserve unrelated regions.\n"
+        "- Keep the UI realistic and readable.\n"
+    )
+
+
+def _make_storyboard_image(
+    frames: list[Image.Image],
+    *,
+    columns: int = 3,
+    background_color: tuple[int, int, int] = (255, 255, 255),
+) -> Image.Image:
+    if not frames:
+        raise ValueError("Expected at least one frame to build a storyboard image.")
+
+    normalized = [frame.convert("RGB") for frame in frames]
+    frame_width = max(frame.width for frame in normalized)
+    frame_height = max(frame.height for frame in normalized)
+    rows = (len(normalized) + columns - 1) // columns
+    storyboard = Image.new("RGB", (frame_width * columns, frame_height * rows), color=background_color)
+
+    for index, frame in enumerate(normalized):
+        x_offset = (index % columns) * frame_width
+        y_offset = (index // columns) * frame_height
+        if frame.size != (frame_width, frame_height):
+            frame = frame.resize((frame_width, frame_height))
+        storyboard.paste(frame, (x_offset, y_offset))
+    return storyboard
+
+
+def _trajectory_judge_payload(frames: list[Image.Image]) -> tuple[str, list[Image.Image]]:
+    storyboard = _make_storyboard_image(frames, columns=3)
+    prompt_suffix = (
+        "The attached image is a storyboard containing six frames arranged left-to-right, "
+        "top-to-bottom as frame0, frame1, frame2, frame3, frame4, frame5."
+    )
+    return prompt_suffix, [storyboard]
+
+
+class LocalJudgeClient:
+    def __init__(self, base_url: str, api_key: str, model: str, timeout: int = 600):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self.timeout = timeout
+
+    def _build_scoring_prompt(self, task_prompt: str) -> str:
+        return (
+            "You are an expert evaluator for GUI image editing and GUI trajectory generation.\n"
+            "Evaluate whether the generated image(s) satisfy the task.\n\n"
+            "Score these five dimensions from 0 to 5:\n"
+            "- goal: whether the user goal is completed correctly\n"
+            "- logic: whether the transition/state change is logically correct\n"
+            "- cons: whether unrelated regions remain consistent\n"
+            "- ui: whether the UI layout/components remain realistic and coherent\n"
+            "- qual: whether the images are visually clear and artifact-free\n\n"
+            "Return JSON only. Do not add any prose outside JSON.\n"
+            "Use exactly this schema:\n"
+            "{\n"
+            '  "goal": 0,\n'
+            '  "logic": 0,\n'
+            '  "cons": 0,\n'
+            '  "ui": 0,\n'
+            '  "qual": 0,\n'
+            '  "reasoning": "short explanation"\n'
+            "}\n\n"
+            "Scoring task:\n"
+            f"{task_prompt}"
+        )
+
+    def _request_text(self, prompt: str, images: list[Image.Image]) -> str:
+        content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
+        for image in images:
+            content.append({"type": "image_url", "image_url": {"url": pil_to_data_url(image)}})
+
+        response = requests.post(
+            build_openai_url(self.base_url, "/chat/completions"),
+            json={
+                "model": self.model,
+                "messages": [{"role": "user", "content": content}],
+                "temperature": 0,
+            },
+            headers={
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+            },
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        message_content = response.json()["choices"][0]["message"]["content"]
+        if isinstance(message_content, list):
+            return "\n".join(part.get("text", "") for part in message_content if part.get("type") == "text")
+        return str(message_content)
+
+    def evaluate(self, *, prompt: str, images: list[Image.Image]) -> dict[str, Any]:
+        primary_prompt = self._build_scoring_prompt(prompt)
+        text = self._request_text(primary_prompt, images)
+        try:
+            return extract_json_object(text)
+        except ValueError:
+            retry_prompt = (
+                self._build_scoring_prompt(prompt) + "\n\nYour previous response was not valid JSON. "
+                "Return only the JSON object with integer scores."
+            )
+            retry_text = self._request_text(retry_prompt, images)
+            try:
+                return extract_json_object(retry_text)
+            except ValueError:
+                return {
+                    "goal": 0,
+                    "logic": 0,
+                    "cons": 0,
+                    "ui": 0,
+                    "qual": 0,
+                    "reasoning": retry_text.strip() or text.strip() or "Judge response was not valid JSON.",
+                }
+
+
+class GEBenchRunner:
+    def __init__(
+        self,
+        *,
+        dataset_root: Path,
+        output_root: Path,
+        base_url: str,
+        model: str,
+        api_key: str = "EMPTY",
+        width: int = 768,
+        height: int = 576,
+        num_inference_steps: int = 8,
+        output_compression: int | None = 98,
+        guidance_scale: float | None = None,
+        seed: int | None = 42,
+    ):
+        self.dataset_root = dataset_root
+        self.output_root = output_root
+        self.model = model
+        self.width = width
+        self.height = height
+        self.num_inference_steps = num_inference_steps
+        self.output_compression = output_compression
+        self.guidance_scale = guidance_scale
+        self.seed = seed
+        self.client = VllmOmniImageClient(base_url=base_url, api_key=api_key)
+
+    def generate(
+        self,
+        *,
+        data_type: str,
+        workers: int = 1,
+        max_samples: int | None = None,
+        samples_per_type: int | None = None,
+    ) -> list[dict[str, Any]]:
+        sample_specs = select_balanced_gebench_samples(
+            {data_type: _iter_sample_specs(self.dataset_root, data_type)},
+            samples_per_type=samples_per_type,
+        )[data_type]
+        if max_samples is not None:
+            sample_specs = sample_specs[:max_samples]
+
+        results: list[dict[str, Any]] = []
+        if workers <= 1:
+            for sample_spec in sample_specs:
+                result = self._generate_one(data_type, sample_spec)
+                if result:
+                    results.append(result)
+            return results
+
+        with ThreadPoolExecutor(max_workers=workers) as executor:
+            futures = [executor.submit(self._generate_one, data_type, sample_spec) for sample_spec in sample_specs]
+            for future in as_completed(futures):
+                result = future.result()
+                if result:
+                    results.append(result)
+        return results
+
+    def _generate_one(self, data_type: str, sample_spec: GEBenchSampleSpec) -> dict[str, Any] | None:
+        sample_path = sample_spec.sample_path
+        metadata = sample_spec.metadata
+        lang_device = sample_spec.lang_device
+        sample_name = sample_spec.sample_name
+        output_dir = ensure_dir(self.output_root / TYPE_TO_FOLDER[data_type] / lang_device / sample_name)
+
+        if data_type == "type1":
+            output_path = output_dir / "generated.png"
+            if output_path.exists():
+                return {
+                    "data_type": data_type,
+                    "sample_name": f"{lang_device}/{sample_name}",
+                    "output_path": str(output_path),
+                }
+            source = _resolve_referenced_image(
+                metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type
+            )
+            if source is None:
+                return None
+            generated = self.client.generate_image_edit(
+                model=self.model,
+                prompt=_type1_prompt(metadata),
+                images=source,
+                width=self.width,
+                height=self.height,
+                num_inference_steps=self.num_inference_steps,
+                output_compression=self.output_compression,
+                guidance_scale=self.guidance_scale,
+                seed=self.seed,
+            )
+            save_image(output_path, generated)
+            return {
+                "data_type": data_type,
+                "sample_name": f"{lang_device}/{sample_name}",
+                "output_path": str(output_path),
+            }
+
+        if data_type == "type2":
+            goal = _text_or_default(metadata.get("question") or metadata.get("caption"), "Complete the task.")
+            source = _resolve_referenced_image(
+                metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type
+            )
+            if source is None:
+                return None
+            frame0_path = output_dir / "frame0.png"
+            if not frame0_path.exists():
+                save_image(frame0_path, source)
+            previous = source
+            for step_num in range(1, 6):
+                frame_path = output_dir / f"frame{step_num}.png"
+                if frame_path.exists():
+                    previous = Image.open(frame_path).convert("RGB")
+                    continue
+                generated = self.client.generate_image_edit(
+                    model=self.model,
+                    prompt=_type2_prompt(goal, step_num),
+                    images=previous,
+                    width=self.width,
+                    height=self.height,
+                    num_inference_steps=self.num_inference_steps,
+                    output_compression=self.output_compression,
+                    guidance_scale=self.guidance_scale,
+                    seed=self.seed,
+                )
+                save_image(frame_path, generated)
+                previous = generated
+            output_path = output_dir / "frame5.png"
+            return {
+                "data_type": data_type,
+                "sample_name": f"{lang_device}/{sample_name}",
+                "output_path": str(output_path),
+            }
+
+        if data_type in {"type3", "type4"}:
+            steps = _trajectory_steps(metadata)
+            frame0_path = output_dir / "frame0.png"
+            if frame0_path.exists():
+                previous = Image.open(frame0_path).convert("RGB")
+            else:
+                previous = self.client.generate_text_to_image(
+                    model=self.model,
+                    prompt=_type34_initial_prompt(metadata, steps[0] if steps else {}),
+                    width=self.width,
+                    height=self.height,
+                    num_inference_steps=self.num_inference_steps,
+                    output_compression=self.output_compression,
+                    guidance_scale=self.guidance_scale,
+                    seed=self.seed,
+                )
+                save_image(frame0_path, previous)
+
+            for step_num in range(1, 6):
+                frame_path = output_dir / f"frame{step_num}.png"
+                if frame_path.exists():
+                    previous = Image.open(frame_path).convert("RGB")
+                    continue
+                step_info = steps[step_num - 1] if step_num - 1 < len(steps) else {}
+                generated = self.client.generate_image_edit(
+                    model=self.model,
+                    prompt=_type34_next_prompt(step_num, step_info),
+                    images=previous,
+                    width=self.width,
+                    height=self.height,
+                    num_inference_steps=self.num_inference_steps,
+                    output_compression=self.output_compression,
+                    guidance_scale=self.guidance_scale,
+                    seed=self.seed,
+                )
+                save_image(frame_path, generated)
+                previous = generated
+            output_path = output_dir / "frame5.png"
+            return {
+                "data_type": data_type,
+                "sample_name": f"{lang_device}/{sample_name}",
+                "output_path": str(output_path),
+            }
+
+        if data_type == "type5":
+            output_path = output_dir / "generated.png"
+            if output_path.exists():
+                return {
+                    "data_type": data_type,
+                    "sample_name": f"{lang_device}/{sample_name}",
+                    "output_path": str(output_path),
+                }
+            source = _resolve_referenced_image(
+                metadata=metadata, sample_path=sample_path, dataset_root=self.dataset_root, data_type=data_type
+            )
+            if source is None:
+                return None
+            generated = self.client.generate_image_edit(
+                model=self.model,
+                prompt=_type5_prompt(metadata),
+                images=source,
+                width=self.width,
+                height=self.height,
+                num_inference_steps=self.num_inference_steps,
+                output_compression=self.output_compression,
+                guidance_scale=self.guidance_scale,
+                seed=self.seed,
+            )
+            save_image(output_path, generated)
+            return {
+                "data_type": data_type,
+                "sample_name": f"{lang_device}/{sample_name}",
+                "output_path": str(output_path),
+            }
+
+        raise ValueError(f"Unsupported data type: {data_type}")
+
+
+class GEBenchEvaluator:
+    def __init__(self, *, dataset_root: Path, output_root: Path, judge: LocalJudgeClient):
+        self.dataset_root = dataset_root
+        self.output_root = output_root
+        self.judge = judge
+
+    def evaluate(
+        self,
+        *,
+        data_type: str,
+        workers: int = 1,
+        max_samples: int | None = None,
+        samples_per_type: int | None = None,
+    ) -> dict[str, Any]:
+        output_type_dir = self.output_root / TYPE_TO_FOLDER[data_type]
+        sample_specs_by_name = {
+            (spec.lang_device, spec.sample_name): spec for spec in _iter_sample_specs(self.dataset_root, data_type)
+        }
+        if not output_type_dir.exists():
+            payload = {"data_type": data_type, "results": [], "summary": summarize_gebench_results([])}
+            write_json(self.output_root / "evaluations" / f"{data_type}.json", payload)
+            return payload
+        sample_dirs = [
+            sample_dir
+            for lang_dir in sorted(path for path in output_type_dir.iterdir() if path.is_dir())
+            for sample_dir in sorted(path for path in lang_dir.iterdir() if path.is_dir())
+            if (lang_dir.name, sample_dir.name) in sample_specs_by_name
+        ]
+        sample_dirs = select_balanced_gebench_samples(
+            {data_type: sample_dirs},
+            samples_per_type=samples_per_type,
+        )[data_type]
+        if max_samples is not None:
+            sample_dirs = sample_dirs[:max_samples]
+        results: list[dict[str, Any]] = []
+        if workers <= 1:
+            for sample_dir in sample_dirs:
+                result = self._evaluate_one(
+                    data_type,
+                    sample_dir,
+                    sample_specs_by_name[(sample_dir.parent.name, sample_dir.name)],
+                )
+                if result:
+                    results.append(result)
+        else:
+            with ThreadPoolExecutor(max_workers=workers) as executor:
+                futures = [
+                    executor.submit(
+                        self._evaluate_one,
+                        data_type,
+                        sample_dir,
+                        sample_specs_by_name[(sample_dir.parent.name, sample_dir.name)],
+                    )
+                    for sample_dir in sample_dirs
+                ]
+                for future in as_completed(futures):
+                    result = future.result()
+                    if result:
+                        results.append(result)
+
+        payload = {"data_type": data_type, "results": results, "summary": summarize_gebench_results(results)}
+        write_json(self.output_root / "evaluations" / f"{data_type}.json", payload)
+        return payload
+
+    def _evaluate_one(self, data_type: str, sample_dir: Path, sample_spec: GEBenchSampleSpec) -> dict[str, Any] | None:
+        lang_device = sample_dir.parent.name
+        sample_name = sample_dir.name
+        dataset_sample = sample_spec.sample_path
+        metadata = sample_spec.metadata
+
+        if data_type == "type1":
+            source = _resolve_referenced_image(
+                metadata=metadata, sample_path=dataset_sample, dataset_root=self.dataset_root, data_type=data_type
+            )
+            generated_path = find_first_image(sample_dir)
+            if source is None or generated_path is None:
+                return None
+            generated = Image.open(generated_path).convert("RGB")
+            raw_scores = self.judge.evaluate(prompt=_type1_prompt(metadata), images=[source, generated])
+        elif data_type == "type2":
+            frames = [Image.open(sample_dir / f"frame{i}.png").convert("RGB") for i in range(6)]
+            goal = _text_or_default(metadata.get("question") or metadata.get("caption"), "Complete the task.")
+            prompt_suffix, judge_images = _trajectory_judge_payload(frames)
+            raw_scores = self.judge.evaluate(
+                prompt=f"Evaluate a six-frame GUI trajectory.\nTask: {goal}\n{prompt_suffix}",
+                images=judge_images,
+            )
+        elif data_type in {"type3", "type4"}:
+            frames = [Image.open(sample_dir / f"frame{i}.png").convert("RGB") for i in range(6)]
+            instruction = _text_or_default(metadata.get("instruction") or metadata.get("caption"), "Complete the task.")
+            prompt_suffix, judge_images = _trajectory_judge_payload(frames)
+            raw_scores = self.judge.evaluate(
+                prompt=f"Evaluate a six-frame GUI trajectory.\nInstruction: {instruction}\n{prompt_suffix}",
+                images=judge_images,
+            )
+        elif data_type == "type5":
+            source = _resolve_referenced_image(
+                metadata=metadata, sample_path=dataset_sample, dataset_root=self.dataset_root, data_type=data_type
+            )
+            generated_path = find_first_image(sample_dir)
+            if source is None or generated_path is None:
+                return None
+            generated = Image.open(generated_path).convert("RGB")
+            raw_scores = self.judge.evaluate(prompt=_type5_prompt(metadata), images=[source, generated])
+        else:
+            raise ValueError(f"Unsupported data type: {data_type}")
+
+        scores = _normalize_scores(raw_scores)
+        return {
+            "sample_name": f"{lang_device}/{sample_name}",
+            "data_type": data_type,
+            "scores": scores,
+            "overall": _compute_overall(scores),
+            "raw_scores": raw_scores,
+        }
+
+
+def _data_types_arg(value: str) -> list[str]:
+    return list(TYPE_TO_FOLDER.keys()) if value == "all" else [value]
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Run local GEBench generation and scoring against vLLM-Omni.")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    generate = subparsers.add_parser("generate")
+    generate.add_argument("--dataset-root", type=Path, required=True)
+    generate.add_argument("--output-root", type=Path, required=True)
+    generate.add_argument("--base-url", type=str, required=True)
+    generate.add_argument("--model", type=str, required=True)
+    generate.add_argument("--data-type", choices=["all", *TYPE_TO_FOLDER.keys()], default="all")
+    generate.add_argument("--api-key", type=str, default="EMPTY")
+    generate.add_argument("--width", type=int, default=768)
+    generate.add_argument("--height", type=int, default=576)
+    generate.add_argument("--num-inference-steps", type=int, default=8)
+    generate.add_argument("--output-compression", type=int, default=98)
+    generate.add_argument("--guidance-scale", type=float, default=None)
+    generate.add_argument("--seed", type=int, default=42)
+    generate.add_argument("--workers", type=int, default=1)
+    generate.add_argument("--max-samples", type=int, default=None)
+    generate.add_argument("--samples-per-type", type=int, default=None)
+
+    evaluate = subparsers.add_parser("evaluate")
+    evaluate.add_argument("--dataset-root", type=Path, required=True)
+    evaluate.add_argument("--output-root", type=Path, required=True)
+    evaluate.add_argument("--data-type", choices=["all", *TYPE_TO_FOLDER.keys()], default="all")
+    evaluate.add_argument("--judge-base-url", type=str, required=True)
+    evaluate.add_argument("--judge-model", type=str, required=True)
+    evaluate.add_argument("--judge-api-key", type=str, default="EMPTY")
+    evaluate.add_argument("--workers", type=int, default=1)
+    evaluate.add_argument("--max-samples", type=int, default=None)
+    evaluate.add_argument("--samples-per-type", type=int, default=None)
+
+    summarize = subparsers.add_parser("summarize")
+    summarize.add_argument("--output-root", type=Path, required=True)
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+
+    if args.command == "generate":
+        runner = GEBenchRunner(
+            dataset_root=args.dataset_root,
+            output_root=args.output_root,
+            base_url=args.base_url,
+            model=args.model,
+            api_key=args.api_key,
+            width=args.width,
+            height=args.height,
+            num_inference_steps=args.num_inference_steps,
+            output_compression=args.output_compression,
+            guidance_scale=args.guidance_scale,
+            seed=args.seed,
+        )
+        records: list[dict[str, Any]] = []
+        for data_type in _data_types_arg(args.data_type):
+            records.extend(
+                runner.generate(
+                    data_type=data_type,
+                    workers=args.workers,
+                    max_samples=args.max_samples,
+                    samples_per_type=args.samples_per_type,
+                )
+            )
+        payload = {"records": records, "summary": summarize_generated_records(records)}
+        write_json(args.output_root / "generation_manifest.json", payload)
+        return 0
+
+    if args.command == "evaluate":
+        judge = LocalJudgeClient(
+            base_url=args.judge_base_url,
+            api_key=args.judge_api_key,
+            model=args.judge_model,
+        )
+        evaluator = GEBenchEvaluator(dataset_root=args.dataset_root, output_root=args.output_root, judge=judge)
+        combined_results: list[dict[str, Any]] = []
+        for data_type in _data_types_arg(args.data_type):
+            payload = evaluator.evaluate(
+                data_type=data_type,
+                workers=args.workers,
+                max_samples=args.max_samples,
+                samples_per_type=args.samples_per_type,
+            )
+            combined_results.extend(payload["results"])
+        _write_json_with_timestamp(
+            args.output_root / "evaluations" / "summary.json",
+            {"summary": summarize_gebench_results(combined_results)},
+        )
+        return 0
+
+    if args.command == "summarize":
+        generation_summary = collect_gebench_generation_summary(args.output_root)
+        evaluation_dir = args.output_root / "evaluations"
+        result_records: list[dict[str, Any]] = []
+        if evaluation_dir.exists():
+            for file_path in sorted(evaluation_dir.glob("type*.json")):
+                payload = load_json(file_path)
+                result_records.extend(payload.get("results", []))
+        payload: dict[str, Any] = {"generation": generation_summary}
+        if result_records:
+            payload["evaluation"] = summarize_gebench_results(result_records)
+        _write_json_with_timestamp(args.output_root / "summary.json", payload)
+        print(json.dumps(payload, indent=2, ensure_ascii=False))
+        return 0
+
+    parser.error(f"Unknown command: {args.command}")
+    return 1

benchmarks/accuracy/text_to_image/run_gebench.py

@@ -0,0 +1,13 @@
+# ruff: noqa: E402, I001
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from benchmarks.accuracy.text_to_image.gbench import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pyproject.toml

@@ -41,6 +41,7 @@ dev = [
     "pytest-asyncio>=0.21.0",
     "pytest-cov>=4.0.0",
     "pytest-mock>=3.10.0",
+    "datasets>=2.14.0",
     "mypy==1.11.1",
     "pre-commit==4.0.1",
     "openai-whisper>=20250625",
@@ -211,3 +212,5 @@ extend-ignore-identifiers-re = [
 ]
 [tool.typos.default.extend-words]
 ue = "ue"
+semantics = "semantics"
+fullset = "fullset"

tests/benchmarks/test_accuracy_bench_utils.py

@@ -0,0 +1,685 @@
+# ruff: noqa: E402, I001
+import math
+import sys
+from pathlib import Path
+
+import pytest
+from PIL import Image
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from benchmarks.accuracy.common import VllmOmniImageClient
+from benchmarks.accuracy.image_to_image.gedit_bench import (
+    GROUPS as GEDIT_GROUPS,
+    GEditBenchEvaluator,
+    GEditBenchRunner,
+    _load_gedit_dataset,
+    _resolve_gedit_split,
+    infer_model_name,
+    resolve_model_name,
+    select_balanced_gedit_rows,
+    parse_score_payload,
+    summarize_generated_records as summarize_gedit_generated_records,
+    summarize_gedit_rows,
+    summarize_gedit_rows_with_backbone,
+)
+from benchmarks.accuracy.text_to_image.gbench import (
+    _expand_sample_path,
+    _trajectory_judge_payload,
+    _write_json_with_timestamp,
+    LocalJudgeClient,
+    GEBenchEvaluator,
+    TYPE_TO_FOLDER,
+    select_balanced_gebench_samples,
+    summarize_generated_records as summarize_gebench_generated_records,
+    summarize_gebench_results,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
+
+
+def test_summarize_gebench_generated_records_groups_by_type():
+    records = [
+        {"data_type": "type1", "sample_name": "english_phone/folder_1", "output_path": "a.png"},
+        {"data_type": "type1", "sample_name": "english_phone/folder_2", "output_path": "b.png"},
+        {"data_type": "type2", "sample_name": "english_phone/folder_3", "output_path": "c.png"},
+    ]
+
+    summary = summarize_gebench_generated_records(records)
+
+    assert summary["count"] == 3
+    assert summary["by_type"]["type1"]["count"] == 2
+    assert summary["by_type"]["type2"]["count"] == 1
+    assert "samples" not in summary["by_type"]["type1"]
+
+
+def test_summarize_gebench_results_computes_type_and_global_means():
+    results = [
+        {"data_type": "type1", "overall": 0.8, "scores": {"goal": 5, "logic": 4}},
+        {"data_type": "type1", "overall": 0.6, "scores": {"goal": 3, "logic": 4}},
+        {"data_type": "type2", "overall": 0.5, "scores": {"goal": 2, "logic": 3}},
+    ]
+
+    summary = summarize_gebench_results(results)
+
+    assert math.isclose(summary["overall_mean"], (0.8 + 0.6 + 0.5) / 3)
+    assert math.isclose(summary["by_type"]["type1"]["overall_mean"], 0.7)
+    assert math.isclose(summary["by_type"]["type2"]["overall_mean"], 0.5)
+    assert math.isclose(summary["by_type"]["type1"]["score_means"]["goal"], 4.0)
+
+
+def test_write_json_with_timestamp_writes_stable_and_timestamped_files(monkeypatch, tmp_path: Path):
+    monkeypatch.setattr(
+        "benchmarks.accuracy.text_to_image.gbench._utc_timestamp",
+        lambda: "20260325T130000Z",
+    )
+
+    timestamped_path = _write_json_with_timestamp(tmp_path / "summary.json", {"ok": True})
+
+    assert (tmp_path / "summary.json").exists()
+    assert timestamped_path == tmp_path / "summary_20260325T130000Z.json"
+    assert timestamped_path.exists()
+
+
+def test_select_balanced_gebench_samples_limits_each_type_independently():
+    sample_paths_by_type = {
+        "type1": [Path(f"/tmp/type1_{idx}") for idx in range(12)],
+        "type2": [Path(f"/tmp/type2_{idx}") for idx in range(8)],
+        "type3": [Path(f"/tmp/type3_{idx}") for idx in range(15)],
+    }
+
+    selected = select_balanced_gebench_samples(sample_paths_by_type, samples_per_type=10)
+
+    assert len(selected["type1"]) == 10
+    assert len(selected["type2"]) == 8
+    assert len(selected["type3"]) == 10
+    assert selected["type1"][0].name == "type1_0"
+    assert selected["type3"][-1].name == "type3_9"
+
+
+def test_expand_sample_path_flattens_json_list_samples(tmp_path: Path):
+    sample_path = tmp_path / "trajectories.json"
+    sample_path.write_text(
+        """
+[
+  {"id": "sample_a", "lang_device": "english_phone", "instruction": "do a"},
+  {"id": "sample_b", "lang_device": "english_phone", "instruction": "do b"}
+]
+""".strip(),
+        encoding="utf-8",
+    )
+
+    specs = _expand_sample_path(sample_path)
+
+    assert len(specs) == 2
+    assert specs[0].sample_name == "sample_a"
+    assert specs[1].sample_name == "sample_b"
+    assert specs[0].lang_device == "english_phone"
+
+
+def test_gebench_evaluate_skips_missing_output_folder(tmp_path: Path):
+    dataset_type_root = tmp_path / TYPE_TO_FOLDER["type3"] / "english_phone"
+    sample_dir = dataset_type_root / "sample_a"
+    sample_dir.mkdir(parents=True)
+    (sample_dir / "meta_data.json").write_text("{}", encoding="utf-8")
+
+    judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge")
+    evaluator = GEBenchEvaluator(dataset_root=tmp_path, output_root=tmp_path / "outputs", judge=judge)
+
+    payload = evaluator.evaluate(data_type="type3")
+
+    assert payload["results"] == []
+    assert payload["summary"]["count"] == 0
+
+
+def test_local_judge_client_retries_when_first_response_is_not_json(monkeypatch):
+    responses = iter(
+        [
+            "The image looks like a GUI screenshot with several controls.",
+            '{"goal": 4, "logic": 4, "cons": 5, "ui": 4, "qual": 4, "reasoning": "mostly correct"}',
+        ]
+    )
+
+    def fake_request_text(self, prompt, images):
+        return next(responses)
+
+    monkeypatch.setattr(LocalJudgeClient, "_request_text", fake_request_text)
+
+    judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge")
+    result = judge.evaluate(prompt="Evaluate this GUI trajectory.", images=[Image.new("RGB", (2, 2), color="white")])
+
+    assert result["goal"] == 4
+    assert result["cons"] == 5
+
+
+def test_local_judge_client_returns_zero_scores_when_retry_is_still_invalid(monkeypatch):
+    responses = iter(
+        [
+            "not json",
+            "still not json",
+        ]
+    )
+
+    def fake_request_text(self, prompt, images):
+        return next(responses)
+
+    monkeypatch.setattr(LocalJudgeClient, "_request_text", fake_request_text)
+
+    judge = LocalJudgeClient(base_url="http://127.0.0.1:8094", api_key="EMPTY", model="judge")
+    result = judge.evaluate(prompt="Evaluate this GUI trajectory.", images=[Image.new("RGB", (2, 2), color="white")])
+
+    assert result["goal"] == 0
+    assert result["logic"] == 0
+    assert result["cons"] == 0
+    assert result["ui"] == 0
+    assert result["qual"] == 0
+    assert result["reasoning"] == "still not json"
+
+
+def test_trajectory_judge_payload_collapses_six_frames_into_single_storyboard():
+    frames = [Image.new("RGB", (8, 6), color=(idx * 10, idx * 10, idx * 10)) for idx in range(6)]
+
+    prompt_suffix, judge_images = _trajectory_judge_payload(frames)
+
+    assert "frame0" in prompt_suffix
+    assert len(judge_images) == 1
+    assert judge_images[0].size == (24, 12)
+
+
+def test_image_edit_client_uses_openai_image_edit_endpoint(monkeypatch):
+    captured = {}
+
+    class FakeResponse:
+        status_code = 200
+
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {
+                "data": [
+                    {
+                        "b64_json": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aY0cAAAAASUVORK5CYII="
+                    }
+                ]
+            }
+
+    def fake_post(url, data=None, files=None, headers=None, timeout=None, **kwargs):
+        captured["url"] = url
+        captured["data"] = data
+        captured["files"] = files
+        captured["headers"] = headers
+        captured["timeout"] = timeout
+        return FakeResponse()
+
+    monkeypatch.setattr("benchmarks.accuracy.common.requests.post", fake_post)
+
+    client = VllmOmniImageClient(base_url="http://127.0.0.1:8093", api_key="EMPTY")
+    image = Image.new("RGB", (2, 2), color="white")
+    output = client.generate_image_edit(
+        model="Qwen/Qwen-Image-Edit",
+        prompt="edit this image",
+        images=image,
+        width=512,
+        height=512,
+    )
+
+    assert output.size == (1, 1)
+    assert captured["url"] == "http://127.0.0.1:8093/v1/images/edits"
+    assert captured["data"]["prompt"] == "edit this image"
+    assert captured["data"]["size"] == "512x512"
+    assert captured["files"][0][0] == "image"
+
+
+def test_text_to_image_client_forwards_output_compression(monkeypatch):
+    captured = {}
+
+    class FakeResponse:
+        status_code = 200
+
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {
+                "data": [
+                    {
+                        "b64_json": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aY0cAAAAASUVORK5CYII="
+                    }
+                ]
+            }
+
+    def fake_post(url, json=None, headers=None, timeout=None, **kwargs):
+        captured["url"] = url
+        captured["json"] = json
+        return FakeResponse()
+
+    monkeypatch.setattr("benchmarks.accuracy.common.requests.post", fake_post)
+
+    client = VllmOmniImageClient(base_url="http://127.0.0.1:8093", api_key="EMPTY")
+    output = client.generate_text_to_image(
+        model="Qwen/Qwen-Image",
+        prompt="generate a gui",
+        width=768,
+        height=576,
+        num_inference_steps=8,
+        output_compression=98,
+    )
+
+    assert output.size == (1, 1)
+    assert captured["url"] == "http://127.0.0.1:8093/v1/images/generations"
+    assert captured["json"]["size"] == "768x576"
+    assert captured["json"]["num_inference_steps"] == 8
+    assert captured["json"]["output_compression"] == 98
+
+
+def test_parse_score_payload_handles_raw_json_and_delimited_json():
+    raw = '{"score": [7, 8], "reasoning": "ok"}'
+    wrapped = 'prefix ||V^=^V|| {"score": [6], "reasoning": "fine"} ||V^=^V|| suffix'
+
+    assert parse_score_payload(raw)["score"] == [7, 8]
+    assert parse_score_payload(wrapped)["score"] == [6]
+
+
+def test_summarize_gedit_generated_records_groups_by_task_and_language():
+    records = []
+    for group in GEDIT_GROUPS[:2]:
+        records.append(
+            {
+                "task_type": group,
+                "instruction_language": "en",
+                "key": f"{group}_en",
+                "output_path": f"{group}_en.png",
+            }
+        )
+        records.append(
+            {
+                "task_type": group,
+                "instruction_language": "cn",
+                "key": f"{group}_cn",
+                "output_path": f"{group}_cn.png",
+            }
+        )
+
+    summary = summarize_gedit_generated_records(records)
+
+    assert summary["count"] == 4
+    assert summary["by_task"][GEDIT_GROUPS[0]]["count"] == 2
+    assert summary["by_language"]["en"]["count"] == 2
+    assert summary["by_language"]["cn"]["samples"] == [
+        f"{GEDIT_GROUPS[0]}_cn",
+        f"{GEDIT_GROUPS[1]}_cn",
+    ]
+
+
+def test_select_balanced_gedit_rows_limits_each_group_independently():
+    rows = []
+    for idx in range(12):
+        rows.append(
+            {
+                "task_type": "background_change",
+                "instruction_language": "en",
+                "key": f"background_change_{idx}",
+            }
+        )
+    for idx in range(7):
+        rows.append(
+            {
+                "task_type": "color_alter",
+                "instruction_language": "en",
+                "key": f"color_alter_{idx}",
+            }
+        )
+
+    selected = select_balanced_gedit_rows(
+        rows,
+        task_type="all",
+        instruction_language="en",
+        samples_per_group=10,
+    )
+
+    selected_background = [row for row in selected if row["task_type"] == "background_change"]
+    selected_color = [row for row in selected if row["task_type"] == "color_alter"]
+
+    assert len(selected_background) == 10
+    assert len(selected_color) == 7
+    assert selected_background[0]["key"] == "background_change_0"
+    assert selected_background[-1]["key"] == "background_change_9"
+
+
+def test_select_balanced_gedit_rows_balances_languages_when_all_requested():
+    rows = []
+    for idx in range(10):
+        rows.append(
+            {
+                "task_type": "background_change",
+                "instruction_language": "cn",
+                "key": f"background_change_cn_{idx}",
+            }
+        )
+    for idx in range(10):
+        rows.append(
+            {
+                "task_type": "background_change",
+                "instruction_language": "en",
+                "key": f"background_change_en_{idx}",
+            }
+        )
+
+    selected = select_balanced_gedit_rows(
+        rows,
+        task_type="all",
+        instruction_language="all",
+        samples_per_group=10,
+    )
+
+    selected_background = [row for row in selected if row["task_type"] == "background_change"]
+
+    assert len(selected_background) == 10
+    assert sum(1 for row in selected_background if row["instruction_language"] == "en") == 5
+    assert sum(1 for row in selected_background if row["instruction_language"] == "cn") == 5
+
+
+def test_infer_model_name_uses_last_path_segment():
+    assert infer_model_name("/workspace/models/Qwen/Qwen-Image-Edit") == "Qwen-Image-Edit"
+
+
+def test_resolve_model_name_prefers_explicit_value_then_model_then_output_root(tmp_path: Path):
+    assert (
+        resolve_model_name(
+            model_name="explicit_name",
+            model="/workspace/models/Qwen/Qwen-Image-Edit",
+        )
+        == "explicit_name"
+    )
+    assert (
+        resolve_model_name(
+            model_name=None,
+            model="/workspace/models/Qwen/Qwen-Image-Edit",
+        )
+        == "Qwen-Image-Edit"
+    )
+
+    output_root = tmp_path / "results"
+    (output_root / "qwen_image_edit").mkdir(parents=True)
+    assert resolve_model_name(model_name=None, output_root=output_root) == "qwen_image_edit"
+
+
+def test_resolve_gedit_split_accepts_dataset_dict_like_input():
+    train_rows = [{"key": "a"}]
+    dataset = {"train": train_rows}
+
+    assert _resolve_gedit_split(dataset) == train_rows
+
+
+def test_resolve_gedit_split_accepts_dataset_like_input():
+    rows = [{"key": "a"}]
+
+    assert _resolve_gedit_split(rows) == rows
+
+
+def test_load_gedit_dataset_uses_load_from_disk_for_saved_dataset(monkeypatch, tmp_path: Path):
+    (tmp_path / "state.json").write_text("{}", encoding="utf-8")
+    (tmp_path / "dataset_info.json").write_text("{}", encoding="utf-8")
+    captured = {}
+
+    def fake_load_dataset(path):
+        captured["load_dataset"] = path
+        return "load_dataset"
+
+    def fake_load_from_disk(path):
+        captured["load_from_disk"] = path
+        return "load_from_disk"
+
+    monkeypatch.setattr(
+        "benchmarks.accuracy.image_to_image.gedit_bench._require_datasets",
+        lambda: (fake_load_dataset, fake_load_from_disk),
+    )
+
+    result = _load_gedit_dataset(str(tmp_path))
+
+    assert result == "load_from_disk"
+    assert captured["load_from_disk"] == str(tmp_path)
+    assert "load_dataset" not in captured
+
+
+def test_load_gedit_dataset_uses_load_dataset_for_local_snapshot_path(monkeypatch, tmp_path: Path):
+    (tmp_path / "README.md").write_text("dataset repo snapshot", encoding="utf-8")
+    captured = {}
+
+    def fake_load_dataset(path):
+        captured["load_dataset"] = path
+        return "load_dataset"
+
+    def fake_load_from_disk(path):
+        captured["load_from_disk"] = path
+        return "load_from_disk"
+
+    monkeypatch.setattr(
+        "benchmarks.accuracy.image_to_image.gedit_bench._require_datasets",
+        lambda: (fake_load_dataset, fake_load_from_disk),
+    )
+
+    result = _load_gedit_dataset(str(tmp_path))
+
+    assert result == "load_dataset"
+    assert captured["load_dataset"] == str(tmp_path)
+    assert "load_from_disk" not in captured
+
+
+def test_gedit_runner_generate_skips_failed_samples(monkeypatch, tmp_path: Path):
+    rows = [
+        {"key": "ok", "task_type": "background_change", "instruction_language": "en"},
+        {"key": "bad", "task_type": "background_change", "instruction_language": "en"},
+    ]
+
+    monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows)
+    runner = GEditBenchRunner(
+        dataset_ref="dataset",
+        output_root=tmp_path,
+        base_url="http://127.0.0.1:8093",
+        model="model",
+    )
+
+    def fake_generate_one(self, model_name, item):
+        if item["key"] == "bad":
+            raise RuntimeError("boom")
+        return {
+            "key": item["key"],
+            "task_type": item["task_type"],
+            "instruction_language": item["instruction_language"],
+        }
+
+    monkeypatch.setattr(GEditBenchRunner, "_generate_one", fake_generate_one)
+
+    outputs = runner.generate(model_name="demo", workers=1)
+
+    assert outputs == [{"key": "ok", "task_type": "background_change", "instruction_language": "en"}]
+
+
+def test_gedit_runner_uses_tqdm_progress(monkeypatch, tmp_path: Path):
+    rows = [
+        {"key": "one", "task_type": "background_change", "instruction_language": "en"},
+        {"key": "two", "task_type": "background_change", "instruction_language": "en"},
+    ]
+    updates = []
+
+    monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows)
+    runner = GEditBenchRunner(
+        dataset_ref="dataset",
+        output_root=tmp_path,
+        base_url="http://127.0.0.1:8093",
+        model="model",
+    )
+
+    def fake_generate_one(self, model_name, item):
+        return {
+            "key": item["key"],
+            "task_type": item["task_type"],
+            "instruction_language": item["instruction_language"],
+        }
+
+    monkeypatch.setattr(GEditBenchRunner, "_generate_one", fake_generate_one)
+
+    class FakeTqdm:
+        def __init__(self, total, desc, unit):
+            self.total = total
+            self.desc = desc
+            self.unit = unit
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def update(self, value):
+            updates.append(value)
+
+    monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench.tqdm", FakeTqdm)
+
+    runner.generate(model_name="demo", workers=1)
+
+    assert updates == [1, 1]
+
+
+def test_gedit_evaluator_skips_failed_samples(monkeypatch, tmp_path: Path):
+    rows = [
+        {"key": "ok", "task_type": "background_change", "instruction_language": "en"},
+        {"key": "bad", "task_type": "background_change", "instruction_language": "en"},
+    ]
+
+    monkeypatch.setattr("benchmarks.accuracy.image_to_image.gedit_bench._load_gedit_dataset", lambda ref: rows)
+    evaluator = GEditBenchEvaluator(dataset_ref="dataset", output_root=tmp_path / "results", scorer=object())
+
+    def fake_evaluate_one(self, model_name, item):
+        if item["key"] == "bad":
+            raise RuntimeError("boom")
+        return {
+            "key": item["key"],
+            "task_type": item["task_type"],
+            "edited_image": "ok.png",
+            "instruction": "edit",
+            "semantics_score": 8.0,
+            "quality_score": 7.0,
+            "overall_score": math.sqrt(56.0),
+            "intersection_exist": True,
+            "instruction_language": item["instruction_language"],
+        }
+
+    monkeypatch.setattr(GEditBenchEvaluator, "_evaluate_one", fake_evaluate_one)
+    monkeypatch.setattr(
+        "benchmarks.accuracy.image_to_image.gedit_bench._utc_timestamp",
+        lambda: "20260325T120000Z",
+    )
+
+    payload = evaluator.evaluate(
+        model_name="demo",
+        save_dir=tmp_path / "scores",
+        instruction_language="en",
+        workers=1,
+    )
+
+    assert len(payload["results"]) == 1
+    assert payload["results"][0]["key"] == "ok"
+    assert payload["summary"]["overall"]["count"] == 1
+    assert Path(payload["csv_path"]).name == "demo_all_en_vie_score.csv"
+    assert Path(payload["summary_path"]).name == "demo_all_en_summary.json"
+    assert Path(payload["timestamped_csv_path"]).name == "demo_all_en_vie_score_20260325T120000Z.csv"
+    assert Path(payload["timestamped_summary_path"]).name == "demo_all_en_summary_20260325T120000Z.json"
+    assert Path(payload["timestamped_csv_path"]).exists()
+    assert Path(payload["timestamped_summary_path"]).exists()
+
+
+def test_summarize_gedit_rows_computes_group_and_intersection_means():
+    rows = []
+    for group in GEDIT_GROUPS:
+        rows.append(
+            {
+                "task_type": group,
+                "instruction_language": "en",
+                "semantics_score": 8.0,
+                "quality_score": 9.0,
+                "intersection_exist": True,
+            }
+        )
+        rows.append(
+            {
+                "task_type": group,
+                "instruction_language": "en",
+                "semantics_score": 6.0,
+                "quality_score": 4.0,
+                "intersection_exist": False,
+            }
+        )
+
+    summary = summarize_gedit_rows(rows, language="en")
+
+    expected_overall = (math.sqrt(8.0 * 9.0) + math.sqrt(6.0 * 4.0)) / 2
+    assert math.isclose(summary["overall"]["Q_SC"], 7.0)
+    assert math.isclose(summary["overall"]["Q_PQ"], 6.5)
+    assert math.isclose(summary["overall"]["Q_O"], expected_overall)
+    assert math.isclose(summary["intersection"]["Q_SC"], 8.0)
+
+
+def test_summarize_gedit_rows_uses_macro_average_across_groups():
+    rows = []
+    for idx in range(10):
+        rows.append(
+            {
+                "task_type": "background_change",
+                "instruction_language": "en",
+                "semantics_score": 10.0,
+                "quality_score": 10.0,
+                "intersection_exist": True,
+            }
+        )
+    for group in GEDIT_GROUPS[1:]:
+        rows.append(
+            {
+                "task_type": group,
+                "instruction_language": "en",
+                "semantics_score": 1.0,
+                "quality_score": 1.0,
+                "intersection_exist": True,
+            }
+        )
+
+    summary = summarize_gedit_rows_with_backbone(rows, language="en")
+
+    expected_macro = (10.0 + 10.0 * 1.0) / 11
+    assert math.isclose(summary["overall"]["Q_SC"], expected_macro)
+    assert math.isclose(summary["overall"]["Q_O"], expected_macro)
+    assert math.isclose(summary["by_group"]["background_change"]["Q_SC"], 10.0)
+
+
+def test_summarize_gedit_rows_with_all_language_splits_en_and_cn():
+    rows = []
+    for group in GEDIT_GROUPS:
+        rows.append(
+            {
+                "task_type": group,
+                "instruction_language": "en",
+                "semantics_score": 8.0,
+                "quality_score": 6.0,
+                "intersection_exist": True,
+            }
+        )
+        rows.append(
+            {
+                "task_type": group,
+                "instruction_language": "cn",
+                "semantics_score": 4.0,
+                "quality_score": 2.0,
+                "intersection_exist": True,
+            }
+        )
+
+    summary = summarize_gedit_rows_with_backbone(rows, language="all")
+
+    assert set(summary["languages"]) == {"en", "cn"}
+    assert math.isclose(summary["languages"]["en"]["overall"]["Q_SC"], 8.0)
+    assert math.isclose(summary["languages"]["en"]["overall"]["Q_PQ"], 6.0)
+    assert math.isclose(summary["languages"]["cn"]["overall"]["Q_O"], math.sqrt(8.0))

tests/conftest.py

@@ -54,6 +54,8 @@ class OmniServerParams(NamedTuple):
     port: int | None = None
     stage_config_path: str | None = None
     server_args: list[str] | None = None
+    env_dict: dict[str, str] | None = None
+    use_omni: bool = True
 
 
 def assert_image_valid(image: Path | Image.Image, *, width: int | None = None, height: int | None = None):
@@ -1076,6 +1078,7 @@ class OmniServer:
         *,
         port: int | None = None,
         env_dict: dict[str, str] | None = None,
+        use_omni: bool = True,
     ) -> None:
         _run_pre_test_cleanup(enable_force=True)
         _run_post_test_cleanup(enable_force=True)
@@ -1083,6 +1086,7 @@ class OmniServer:
         self.model = model
         self.serve_args = serve_args
         self.env_dict = env_dict
+        self.use_omni = use_omni
         self.proc: subprocess.Popen | None = None
         self.host = "127.0.0.1"
         if port is None:
@@ -1103,12 +1107,14 @@ class OmniServer:
             "vllm_omni.entrypoints.cli.main",
             "serve",
             self.model,
-            "--omni",
             "--host",
             self.host,
             "--port",
             str(self.port),
-        ] + self.serve_args
+        ]
+        if self.use_omni:
+            cmd.append("--omni")
+        cmd += self.serve_args
 
         print(f"Launching OmniServer with: {' '.join(cmd)}")
         self.proc = subprocess.Popen(
@@ -1237,19 +1243,35 @@ def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: st
         stage_config_path = params.stage_config_path
         if run_level == "advanced_model" and stage_config_path is not None:
             with open(stage_config_path, encoding="utf-8") as f:
-                _cfg = yaml.safe_load(f) or {}
-            _stage_ids = [s["stage_id"] for s in _cfg.get("stage_args", []) if "stage_id" in s]
+                cfg = yaml.safe_load(f) or {}
+            stage_ids = [stage["stage_id"] for stage in cfg.get("stage_args", []) if "stage_id" in stage]
             stage_config_path = modify_stage_config(
                 stage_config_path,
-                deletes={"stage_args": {sid: ["engine_args.load_format"] for sid in _stage_ids}},
+                deletes={"stage_args": {stage_id: ["engine_args.load_format"] for stage_id in stage_ids}},
             )
 
         server_args = params.server_args or []
-        server_args = ["--stage-init-timeout", "120", *server_args]
+        if params.use_omni:
+            server_args = ["--stage-init-timeout", "120", *server_args]
         if stage_config_path is not None:
             server_args += ["--stage-configs-path", stage_config_path]
 
-        with OmniServer(model, server_args, port=port) if port else OmniServer(model, server_args) as server:
+        with (
+            OmniServer(
+                model,
+                server_args,
+                port=port,
+                env_dict=params.env_dict,
+                use_omni=params.use_omni,
+            )
+            if port
+            else OmniServer(
+                model,
+                server_args,
+                env_dict=params.env_dict,
+                use_omni=params.use_omni,
+            )
+        ) as server:
             print("OmniServer started successfully")
             yield server
             print("OmniServer stopping...")

tests/e2e/accuracy/conftest.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+import torch
+
+from tests.conftest import OmniServer, OmniServerParams
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("accuracy-e2e")
+    group.addoption("--gebench-root", action="store", default=None, help="Local GEBench dataset root")
+    group.addoption("--gedit-root", action="store", default=None, help="Local GEdit-Bench dataset root")
+    group.addoption(
+        "--gebench-model", action="store", default="Qwen/Qwen-Image-2512", help="Generate model for GEBench smoke"
+    )
+    group.addoption(
+        "--gedit-model", action="store", default="Qwen/Qwen-Image-Edit", help="Generate model for GEdit-Bench smoke"
+    )
+    group.addoption(
+        "--accuracy-judge-model",
+        action="store",
+        default="QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ",
+        help="Judge model path",
+    )
+    group.addoption("--accuracy-gpu", action="store", default="0", help="Single GPU id used sequentially")
+    group.addoption("--gebench-port", action="store", type=int, default=8093, help="Generate port for GEBench")
+    group.addoption("--gedit-port", action="store", type=int, default=8093, help="Generate port for GEdit-Bench")
+    group.addoption(
+        "--gebench-samples-per-type",
+        action="store",
+        type=int,
+        default=10,
+        help="Balanced sample count per GEBench type",
+    )
+    group.addoption(
+        "--gedit-samples-per-group",
+        action="store",
+        type=int,
+        default=20,
+        help="Balanced sample count per GEdit task group",
+    )
+    group.addoption("--accuracy-workers", action="store", type=int, default=1, help="Worker count for accuracy benches")
+
+
+def _hf_cache_root() -> Path:
+    return Path(os.environ.get("HF_HOME", "/root/.cache/huggingface"))
+
+
+def _dataset_cache_dirs(dataset_id: str) -> list[Path]:
+    cache_root = _hf_cache_root() / "hub" / f"datasets--{dataset_id.replace('/', '--')}" / "snapshots"
+    if not cache_root.exists():
+        return []
+    return sorted(
+        (path for path in cache_root.iterdir() if path.is_dir()), key=lambda path: path.stat().st_mtime, reverse=True
+    )
+
+
+def _ensure_dataset_snapshot(dataset_id: str) -> Path:
+    candidates = _dataset_cache_dirs(dataset_id)
+    if candidates:
+        return candidates[0]
+
+    subprocess.run(
+        ["huggingface-cli", "download", "--repo-type", "dataset", dataset_id],
+        check=True,
+    )
+    candidates = _dataset_cache_dirs(dataset_id)
+    if not candidates:
+        raise FileNotFoundError(
+            f"Dataset {dataset_id} was downloaded but no snapshot was found under {_hf_cache_root()}"
+        )
+    return candidates[0]
+
+
+def _resolve_dataset_root(request: pytest.FixtureRequest, option_name: str, dataset_id: str) -> Path:
+    value = request.config.getoption(option_name)
+    if value:
+        path = Path(value)
+        if not path.exists():
+            pytest.skip(f"Dataset path does not exist: {path}")
+        return path
+    return _ensure_dataset_snapshot(dataset_id)
+
+
+@dataclass
+class AccuracyServerConfig:
+    generate_params: OmniServerParams
+    judge_params: OmniServerParams
+    run_level: str
+    model_prefix: str
+
+    @contextmanager
+    def generate_server(self):
+        params = self.generate_params
+        model = self.model_prefix + params.model
+        server_args = params.server_args or []
+        if params.use_omni:
+            server_args = ["--stage-init-timeout", "120", *server_args]
+        with OmniServer(
+            model,
+            server_args,
+            port=params.port,
+            env_dict=params.env_dict,
+            use_omni=params.use_omni,
+        ) as server:
+            yield server
+
+    @contextmanager
+    def judge_server(self):
+        params = self.judge_params
+        model = self.model_prefix + params.model
+        server_args = params.server_args or []
+        with OmniServer(
+            model,
+            server_args,
+            port=params.port,
+            env_dict=params.env_dict,
+            use_omni=params.use_omni,
+        ) as server:
+            yield server
+
+
+@pytest.fixture(scope="session")
+def gebench_dataset_root(request: pytest.FixtureRequest) -> Path:
+    return _resolve_dataset_root(request, "gebench_root", "stepfun-ai/GEBench")
+
+
+@pytest.fixture(scope="session")
+def gedit_dataset_root(request: pytest.FixtureRequest) -> Path:
+    return _resolve_dataset_root(request, "gedit_root", "stepfun-ai/GEdit-Bench")
+
+
+@pytest.fixture(scope="session")
+def accuracy_workers(request: pytest.FixtureRequest) -> int:
+    return int(request.config.getoption("accuracy_workers"))
+
+
+@pytest.fixture(scope="session")
+def gebench_samples_per_type(request: pytest.FixtureRequest) -> int:
+    return int(request.config.getoption("gebench_samples_per_type"))
+
+
+@pytest.fixture(scope="session")
+def gedit_samples_per_group(request: pytest.FixtureRequest) -> int:
+    return int(request.config.getoption("gedit_samples_per_group"))
+
+
+@pytest.fixture(scope="session")
+def accuracy_artifact_root() -> Path:
+    root = Path(__file__).resolve().parent / "artifacts"
+    root.mkdir(parents=True, exist_ok=True)
+    return root
+
+
+def reset_artifact_dir(path: Path) -> Path:
+    if path.exists():
+        shutil.rmtree(path)
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def infer_model_label(model: str) -> str:
+    label = Path(model.rstrip("/\\")).name or "model"
+    return "".join(char if char.isalnum() or char in {"-", "_"} else "_" for char in label)
+
+
+def _build_accuracy_server_config(
+    *,
+    generate_model: str,
+    judge_model: str,
+    shared_gpu: str,
+    port: int,
+    run_level: str,
+    model_prefix: str,
+) -> AccuracyServerConfig:
+    if torch.cuda.device_count() < 1:
+        pytest.skip("Need at least 1 CUDA GPU for accuracy benchmark smoke tests.")
+
+    if not generate_model:
+        pytest.skip("No generate model configured for accuracy benchmark test.")
+    generate_server_args = ["--num-gpus", "1"]
+    judge_server_args = [
+        "--max-model-len",
+        "32768",
+        "--gpu-memory-utilization",
+        "0.8",
+    ]
+
+    judge_env = {"CUDA_VISIBLE_DEVICES": shared_gpu}
+
+    return AccuracyServerConfig(
+        generate_params=OmniServerParams(
+            model=generate_model,
+            port=port,
+            server_args=generate_server_args,
+            env_dict={"CUDA_VISIBLE_DEVICES": shared_gpu},
+            use_omni=True,
+        ),
+        judge_params=OmniServerParams(
+            model=judge_model,
+            port=port,
+            server_args=judge_server_args,
+            env_dict=judge_env,
+            use_omni=False,
+        ),
+        run_level=run_level,
+        model_prefix=model_prefix,
+    )
+
+
+@pytest.fixture
+def gebench_accuracy_servers(
+    request: pytest.FixtureRequest,
+    run_level: str,
+    model_prefix: str,
+) -> AccuracyServerConfig:
+    return _build_accuracy_server_config(
+        generate_model=request.config.getoption("gebench_model"),
+        judge_model=request.config.getoption("accuracy_judge_model"),
+        shared_gpu=str(request.config.getoption("accuracy_gpu")),
+        port=int(request.config.getoption("gebench_port")),
+        run_level=run_level,
+        model_prefix=model_prefix,
+    )
+
+
+@pytest.fixture
+def gedit_accuracy_servers(
+    request: pytest.FixtureRequest,
+    run_level: str,
+    model_prefix: str,
+) -> AccuracyServerConfig:
+    return _build_accuracy_server_config(
+        generate_model=request.config.getoption("gedit_model"),
+        judge_model=request.config.getoption("accuracy_judge_model"),
+        shared_gpu=str(request.config.getoption("accuracy_gpu")),
+        port=int(request.config.getoption("gedit_port")),
+        run_level=run_level,
+        model_prefix=model_prefix,
+    )

tests/e2e/accuracy/test_gebench_h100_smoke.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from benchmarks.accuracy.text_to_image.gbench import main as gbench_main
+from tests.e2e.accuracy.conftest import infer_model_label, reset_artifact_dir
+from tests.utils import hardware_test
+
+
+@pytest.mark.advanced_model
+@pytest.mark.benchmark
+@pytest.mark.diffusion
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+def test_gebench_h100_smoke(
+    gebench_accuracy_servers,
+    accuracy_artifact_root: Path,
+    gebench_dataset_root: Path,
+    accuracy_workers: int,
+) -> None:
+    model_label = infer_model_label(gebench_accuracy_servers.generate_params.model).lower()
+    output_root = reset_artifact_dir(accuracy_artifact_root / f"gebench_{model_label}")
+
+    with gebench_accuracy_servers.generate_server() as generate_server:
+        for data_type in ("type3", "type4"):
+            assert (
+                gbench_main(
+                    [
+                        "generate",
+                        "--dataset-root",
+                        str(gebench_dataset_root),
+                        "--output-root",
+                        str(output_root),
+                        "--base-url",
+                        f"http://{generate_server.host}:{generate_server.port}",
+                        "--model",
+                        generate_server.model,
+                        "--data-type",
+                        data_type,
+                        "--width",
+                        "768",
+                        "--height",
+                        "576",
+                        "--output-compression",
+                        "98",
+                        "--num-inference-steps",
+                        "8",
+                        "--workers",
+                        str(accuracy_workers),
+                    ]
+                )
+                == 0
+            )
+
+    with gebench_accuracy_servers.judge_server() as judge_server:
+        for data_type in ("type3", "type4"):
+            assert (
+                gbench_main(
+                    [
+                        "evaluate",
+                        "--dataset-root",
+                        str(gebench_dataset_root),
+                        "--output-root",
+                        str(output_root),
+                        "--data-type",
+                        data_type,
+                        "--judge-base-url",
+                        f"http://{judge_server.host}:{judge_server.port}",
+                        "--judge-model",
+                        judge_server.model,
+                        "--judge-api-key",
+                        "EMPTY",
+                        "--workers",
+                        str(accuracy_workers),
+                    ]
+                )
+                == 0
+            )
+
+    assert gbench_main(["summarize", "--output-root", str(output_root)]) == 0
+
+    summary = json.loads((output_root / "summary.json").read_text(encoding="utf-8"))
+    assert "generation" in summary
+    assert "evaluation" in summary
+
+    for data_type in ("type3", "type4"):
+        assert data_type in summary["generation"]["by_type"]
+        assert summary["generation"]["by_type"][data_type]["count"] > 0
+        assert data_type in summary["evaluation"]["by_type"]
+        assert summary["evaluation"]["by_type"][data_type]["count"] > 0
+
+    assert summary["evaluation"]["overall_mean"] >= 0.45
+    assert summary["evaluation"]["by_type"]["type3"]["overall_mean"] >= 0.45
+    assert summary["evaluation"]["by_type"]["type4"]["overall_mean"] >= 0.45

tests/e2e/accuracy/test_gedit_bench_h100_smoke.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from benchmarks.accuracy.image_to_image.gedit_bench import GROUPS
+from benchmarks.accuracy.image_to_image.gedit_bench import main as gedit_main
+from tests.e2e.accuracy.conftest import infer_model_label, reset_artifact_dir
+from tests.utils import hardware_test
+
+
+@pytest.mark.advanced_model
+@pytest.mark.benchmark
+@pytest.mark.diffusion
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+def test_gedit_bench_h100_smoke(
+    gedit_accuracy_servers,
+    accuracy_artifact_root: Path,
+    gedit_dataset_root: Path,
+    gedit_samples_per_group: int,
+    accuracy_workers: int,
+) -> None:
+    model_label = infer_model_label(gedit_accuracy_servers.generate_params.model).lower()
+    output_root = reset_artifact_dir(accuracy_artifact_root / f"gedit_results_{model_label}")
+    score_root = reset_artifact_dir(accuracy_artifact_root / f"gedit_scores_{model_label}")
+    model_name = model_label
+
+    with gedit_accuracy_servers.generate_server() as generate_server:
+        assert (
+            gedit_main(
+                [
+                    "generate",
+                    "--dataset-ref",
+                    str(gedit_dataset_root),
+                    "--output-root",
+                    str(output_root),
+                    "--base-url",
+                    f"http://{generate_server.host}:{generate_server.port}",
+                    "--model",
+                    generate_server.model,
+                    "--model-name",
+                    model_name,
+                    "--task-type",
+                    "all",
+                    "--instruction-language",
+                    "all",
+                    "--samples-per-group",
+                    str(gedit_samples_per_group),
+                    "--workers",
+                    str(accuracy_workers),
+                ]
+            )
+            == 0
+        )
+
+    with gedit_accuracy_servers.judge_server() as judge_server:
+        assert (
+            gedit_main(
+                [
+                    "evaluate",
+                    "--dataset-ref",
+                    str(gedit_dataset_root),
+                    "--output-root",
+                    str(output_root),
+                    "--model-name",
+                    model_name,
+                    "--save-dir",
+                    str(score_root),
+                    "--task-type",
+                    "all",
+                    "--instruction-language",
+                    "all",
+                    "--judge-base-url",
+                    f"http://{judge_server.host}:{judge_server.port}",
+                    "--judge-model",
+                    judge_server.model,
+                    "--judge-api-key",
+                    "EMPTY",
+                    "--samples-per-group",
+                    str(gedit_samples_per_group),
+                    "--workers",
+                    str(accuracy_workers),
+                ]
+            )
+            == 0
+        )
+
+    csv_path = score_root / f"{model_name}_all_all_vie_score.csv"
+    assert gedit_main(["summarize", "--csv-path", str(csv_path), "--language", "all"]) == 0
+
+    summary_path = score_root / f"{model_name}_all_all_summary.json"
+    summary = json.loads(summary_path.read_text(encoding="utf-8"))
+    assert set(summary["languages"]) == {"en", "cn"}
+
+    for language in ["en", "cn"]:
+        language_summary = summary["languages"][language]
+        assert language_summary["overall"]["count"] is not None
+        assert language_summary["intersection"]["count"] is not None
+        assert language_summary["overall"]["Q_SC"] is not None
+        assert language_summary["overall"]["Q_PQ"] is not None
+        assert language_summary["overall"]["Q_O"] is not None
+
+        for group in GROUPS:
+            group_summary = language_summary["by_group"][group]
+            assert set(group_summary) == {"count", "Q_SC", "Q_PQ", "Q_O"}
+
+    assert summary["languages"]["en"]["overall"]["Q_SC"] >= 7.1
+    assert summary["languages"]["en"]["overall"]["Q_PQ"] >= 5.9
+    assert summary["languages"]["en"]["overall"]["Q_O"] >= 6.3
+    assert summary["languages"]["cn"]["overall"]["Q_SC"] >= 7.0
+    assert summary["languages"]["cn"]["overall"]["Q_PQ"] >= 5.8
+    assert summary["languages"]["cn"]["overall"]["Q_O"] >= 6.2