Upload GPT_TO_HF_CityResponsesBatch.py

GPT_TO_HF_CityResponsesBatch.py

@@ -0,0 +1,2175 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+import re
+import secrets
+import shutil
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime, timezone
+from io import BytesIO
+from pathlib import Path
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
+
+try:
+    import folder_paths
+except ImportError:  # Allows offline tests outside ComfyUI.
+    folder_paths = None
+
+try:
+    from PIL import Image
+except ImportError:  # Raised with a useful message only when an image is decoded.
+    Image = None
+
+
+NODE_VERSION = "2.1.0"
+PURPOSE = "salia-city-responses-batch-waves"
+OPENAI_BATCH_ENDPOINT = "/v1/responses"
+OPENAI_MODEL = "gpt-5.6-sol"
+OPENAI_IMAGE_MODEL = "gpt-image-2.5-sunburst"
+OPENAI_REASONING: Dict[str, str] = {
+    "effort": "medium",
+    "mode": "standard",
+    "summary": "auto",
+}
+RESPONSE_MODERATION: Dict[str, Any] = {
+    "model": "omni-moderation-latest",
+    "policy": {
+        "input": {"mode": "score"},
+        "output": {"mode": "score"},
+    },
+}
+OPENAI_IMAGE_TOOL: Dict[str, Any] = {
+    "type": "image_generation",
+    "model": OPENAI_IMAGE_MODEL,
+    "action": "generate",
+    "size": "1536x1024",
+    "quality": "auto",
+    "output_format": "png",
+    "background": "opaque",
+    "moderation": "low",
+    "partial_images": 0,
+}
+
+SOURCE_REPO_ID = "saliacoel/maps_imgs"
+GENERATED_REPO_ID = "saliacoel/map_imgs_gpt"
+REFERENCE_REPO_ID = "saliacoel/x"
+REPO_TYPE = "model"
+REPO_REVISION = "main"
+MANIFEST_FILE_NAME = "gpt_navigation_manifest.json"
+STATE_DIRECTORY_NAME = "Salia_City_Responses_Batch"
+EXPECTED_MANIFEST_VERSION = 5
+EXPECTED_BATCH_PLAN_VERSION = 3
+EXPECTED_BATCH_STRATEGY = (
+    "entrance_fronts_first_scene_tree_with_deferred_cycles_v3"
+)
+EXPECTED_PROMPT_VERSION = "city_responses_continuity_prompt_v2"
+EXPECTED_REFERENCE_TYPES = (
+    "4_way_Junction",
+    "Left_Forward",
+    "Forward_Right",
+    "Left_Right",
+    "Forward_Only",
+    "Left_Only",
+    "Right_Only",
+    "Dead_End",
+    "straight_door",
+    "door_Left",
+    "door_Right",
+    "station_gate_Front_Direct",
+    "station_gate_End_Left",
+    "station_gate_End_Right",
+    "station_gate_Forward_Left",
+    "station_gate_Forward_Right",
+    "station_gate_Distant_Front",
+    "station_gate_Distant_Left",
+    "station_gate_Distant_Right",
+)
+
+ACTION_PREVIEW = "PREVIEW_PROMPTS_ONLY"
+ACTION_RUN = "RUN_OR_CONTINUE"
+_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
+_SAFE_BATCH_ID = re.compile(r"^batch_[A-Za-z0-9_-]{1,200}$")
+_SAFE_FILE_ID = re.compile(r"^file-[A-Za-z0-9_-]{1,200}$")
+_TERMINAL_BATCH_STATUSES = {"completed", "failed", "expired", "cancelled"}
+_WAITING_BATCH_STATUSES = {
+    "validating",
+    "in_progress",
+    "finalizing",
+    "cancelling",
+}
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _output_root() -> Path:
+    if folder_paths is not None:
+        return Path(folder_paths.get_output_directory())
+    return Path.cwd() / "output"
+
+
+def _optional_secret(value: str) -> Optional[str]:
+    cleaned = str(value or "").strip()
+    if cleaned.lower() in {"", "none", "xxx", "hf_xxx", "sk_xxx", "sk-xxx"}:
+        return None
+    return cleaned
+
+
+def _required_secret(value: str, label: str) -> str:
+    cleaned = _optional_secret(value)
+    if not cleaned:
+        raise ValueError(f"A real {label} is required for RUN_OR_CONTINUE.")
+    return cleaned
+
+
+def _safe_district_name(value: str) -> str:
+    name = str(value or "").strip().strip("\"'")
+    if (
+        not name
+        or name in {".", ".."}
+        or "/" in name
+        or "\\" in name
+        or len(name) > 180
+        or any(ord(character) < 32 or ord(character) == 127 for character in name)
+    ):
+        raise ValueError("DISTRICT_NAME must be one safe Hugging Face folder name.")
+    return name
+
+
+def _safe_file_name(value: Any, label: str) -> str:
+    name = str(value or "").strip()
+    if (
+        not name
+        or name in {".", ".."}
+        or "/" in name
+        or "\\" in name
+        or len(name) > 240
+        or any(ord(character) < 32 or ord(character) == 127 for character in name)
+    ):
+        raise ValueError(f"Unsafe {label}: {name!r}.")
+    return name
+
+
+def _write_json(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".part")
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    os.replace(temporary, path)
+
+
+def _read_json(path: Path) -> Dict[str, Any]:
+    parsed = json.loads(path.read_text(encoding="utf-8-sig"))
+    if not isinstance(parsed, dict):
+        raise ValueError(f"Expected a JSON object in {path}.")
+    return parsed
+
+
+def _write_jsonl(path: Path, values: Sequence[Mapping[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".part")
+    with temporary.open("w", encoding="utf-8", newline="\n") as handle:
+        for value in values:
+            handle.write(
+                json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+                + "\n"
+            )
+    os.replace(temporary, path)
+
+
+def _load_hf_tools():
+    try:
+        from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
+    except ImportError as exception:
+        raise RuntimeError(
+            "huggingface_hub is missing. Install it with ComfyUI's Python and "
+            "restart ComfyUI."
+        ) from exception
+    return CommitOperationAdd, HfApi, hf_hub_download
+
+
+def _repo_commit(api: Any, repo_id: str, token: Optional[str]) -> str:
+    info = api.repo_info(
+        repo_id=repo_id,
+        repo_type=REPO_TYPE,
+        revision=REPO_REVISION,
+        token=token,
+    )
+    commit = str(getattr(info, "sha", "") or "").strip()
+    if not commit:
+        raise ValueError(f"Could not resolve the current commit of {repo_id}.")
+    return commit
+
+
+def _hf_url(repo_id: str, revision: str, path_in_repo: str) -> str:
+    return (
+        "https://huggingface.co/"
+        + repo_id
+        + "/resolve/"
+        + urllib.parse.quote(str(revision), safe="")
+        + "/"
+        + urllib.parse.quote(str(path_in_repo), safe="/")
+    )
+
+
+def _hf_blob_url(repo_id: str, revision: str, path_in_repo: str) -> str:
+    return (
+        "https://huggingface.co/"
+        + repo_id
+        + "/blob/"
+        + urllib.parse.quote(str(revision), safe="")
+        + "/"
+        + urllib.parse.quote(str(path_in_repo), safe="/")
+    )
+
+
+def _repo_path(district: str, file_name: str) -> str:
+    return district + "/" + _safe_file_name(file_name, "repository filename")
+
+
+def _download_manifest(
+    district: str,
+    hf_token: Optional[str],
+) -> Tuple[Dict[str, Any], str, str, Path]:
+    _, HfApi, hf_hub_download = _load_hf_tools()
+    api = HfApi(token=hf_token)
+    source_commit = _repo_commit(api, SOURCE_REPO_ID, hf_token)
+    path_in_repo = _repo_path(district, MANIFEST_FILE_NAME)
+    local = Path(
+        hf_hub_download(
+            repo_id=SOURCE_REPO_ID,
+            filename=path_in_repo,
+            repo_type=REPO_TYPE,
+            revision=source_commit,
+            token=hf_token,
+            force_download=True,
+        )
+    )
+    raw = local.read_bytes()
+    manifest_sha256 = hashlib.sha256(raw).hexdigest()
+    parsed = json.loads(raw.decode("utf-8-sig"))
+    if not isinstance(parsed, dict):
+        raise ValueError("The downloaded navigation manifest is not a JSON object.")
+    return parsed, source_commit, manifest_sha256, local
+
+
+def _image_label(zero_based_index: int) -> str:
+    value = int(zero_based_index) + 1
+    if value <= 0:
+        raise ValueError("Image label index must be non-negative.")
+    letters = ""
+    while value:
+        value, remainder = divmod(value - 1, 26)
+        letters = chr(ord("A") + remainder) + letters
+    return "Image_" + letters
+
+
+def _segmentation_file_name(reference_type: str) -> str:
+    return (
+        "4_way_Junction_Segmentation.png"
+        if reference_type == "4_way_Junction"
+        else reference_type + "_Segment.png"
+    )
+
+
+def _require_mapping(value: Any, label: str) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        raise ValueError(f"{label} must be a JSON object.")
+    return value
+
+
+def _require_list(value: Any, label: str) -> List[Any]:
+    if not isinstance(value, list):
+        raise ValueError(f"{label} must be a JSON array.")
+    return value
+
+
+def _validate_manifest(
+    manifest: Mapping[str, Any],
+    requested_district: str,
+) -> List[Dict[str, Any]]:
+    if int(manifest.get("formatVersion") or 0) != EXPECTED_MANIFEST_VERSION:
+        raise ValueError(
+            f"Unsupported navigation manifest formatVersion; expected "
+            f"{EXPECTED_MANIFEST_VERSION}. Recapture with Unity V6.0.22's v5 exporter."
+        )
+    if int(manifest.get("batchPlanVersion") or 0) != EXPECTED_BATCH_PLAN_VERSION:
+        raise ValueError(
+            f"Unsupported batchPlanVersion; expected {EXPECTED_BATCH_PLAN_VERSION}."
+        )
+    if str(manifest.get("districtName") or "") != requested_district:
+        raise ValueError(
+            "The manifest districtName does not equal the requested Hugging Face "
+            "folder name."
+        )
+    expected_top_level = {
+        "sourceImageRepoId": SOURCE_REPO_ID,
+        "generatedImageRepoId": GENERATED_REPO_ID,
+        "referenceImageRepoId": REFERENCE_REPO_ID,
+        "repoRevision": REPO_REVISION,
+        "batchStrategy": EXPECTED_BATCH_STRATEGY,
+        "promptVersion": EXPECTED_PROMPT_VERSION,
+    }
+    for key, expected in expected_top_level.items():
+        actual = str(manifest.get(key) or "")
+        if actual != expected:
+            raise ValueError(f"Manifest {key} is {actual!r}; expected {expected!r}.")
+
+    catalog = _require_list(manifest.get("referenceCatalog"), "referenceCatalog")
+    catalog_by_type: Dict[str, Mapping[str, Any]] = {}
+    for index, raw_entry in enumerate(catalog):
+        entry = _require_mapping(raw_entry, f"referenceCatalog[{index}]")
+        reference_type = str(entry.get("referenceType") or "")
+        if reference_type in catalog_by_type:
+            raise ValueError(f"Duplicate reference type {reference_type!r}.")
+        catalog_by_type[reference_type] = entry
+    if tuple(catalog_by_type.keys()) != EXPECTED_REFERENCE_TYPES:
+        raise ValueError(
+            f"referenceCatalog must contain the {len(EXPECTED_REFERENCE_TYPES)} "
+            "canonical types in canonical order."
+        )
+    for reference_type in EXPECTED_REFERENCE_TYPES:
+        entry = catalog_by_type[reference_type]
+        expected_names = {
+            "depthFileName": reference_type + "_Depth.png",
+            "segmentationFileName": _segmentation_file_name(reference_type),
+            "lineartFileName": reference_type + "_Lineart.png",
+        }
+        for key, expected in expected_names.items():
+            if str(entry.get(key) or "") != expected:
+                raise ValueError(
+                    f"Reference {reference_type!r} has invalid {key}; expected "
+                    f"{expected!r}."
+                )
+
+    raw_sets = _require_list(manifest.get("batchSets"), "batchSets")
+    if int(manifest.get("batchSetCount") or -1) != len(raw_sets) or not raw_sets:
+        raise ValueError("batchSetCount does not match the non-empty batchSets array.")
+
+    jobs: List[Dict[str, Any]] = []
+    jobs_by_id: Dict[str, Dict[str, Any]] = {}
+    output_names = set()
+    set_ids = set()
+    declared_dependencies_by_set: Dict[int, List[int]] = {}
+    for set_index, raw_set in enumerate(raw_sets):
+        batch_set = _require_mapping(raw_set, f"batchSets[{set_index}]")
+        set_number = int(batch_set.get("batchSetNumber") or 0)
+        if set_number != set_index + 1:
+            raise ValueError("Batch-set numbers must be consecutive and one-based.")
+        set_id = str(batch_set.get("id") or "").strip()
+        if not set_id or set_id in set_ids:
+            raise ValueError(f"Missing or duplicate Batch-set id {set_id!r}.")
+        set_ids.add(set_id)
+        raw_depends_on = _require_list(
+            batch_set.get("dependsOnBatchSets"),
+            f"batchSets[{set_index}].dependsOnBatchSets",
+        )
+        depends_on = [int(value) for value in raw_depends_on]
+        if (
+            depends_on != sorted(set(depends_on))
+            or any(value < 1 or value >= set_number for value in depends_on)
+        ):
+            raise ValueError(
+                f"Batch set {set_number} has invalid dependsOnBatchSets."
+            )
+        declared_dependencies_by_set[set_number] = depends_on
+        raw_images = _require_list(
+            batch_set.get("images"), f"batchSets[{set_index}].images"
+        )
+        if int(batch_set.get("imageCount") or -1) != len(raw_images) or not raw_images:
+            raise ValueError(f"Batch set {set_number} has an invalid imageCount.")
+        for image_index, raw_image in enumerate(raw_images):
+            image = dict(
+                _require_mapping(
+                    raw_image,
+                    f"batchSets[{set_index}].images[{image_index}]",
+                )
+            )
+            custom_id = str(image.get("customId") or "").strip()
+            if not custom_id or custom_id in jobs_by_id:
+                raise ValueError(f"Missing or duplicate image customId {custom_id!r}.")
+            if str(image.get("kind") or "").lower() == "bench" or "bench" in str(
+                image.get("topology") or ""
+            ).lower():
+                raise ValueError("Bench images are forbidden in the city Batch plan.")
+            source_name = _safe_file_name(
+                image.get("sourceScreenshotFileName"), "Unity screenshot filename"
+            )
+            target_name = _safe_file_name(
+                image.get("targetGeneratedFileName"), "generated filename"
+            )
+            if not target_name.lower().endswith(".png") or target_name in output_names:
+                raise ValueError(f"Invalid or duplicate generated filename {target_name!r}.")
+            output_names.add(target_name)
+            reference_type = str(image.get("referenceType") or "")
+            if reference_type not in catalog_by_type:
+                raise ValueError(
+                    f"Image {custom_id!r} uses unknown referenceType {reference_type!r}."
+                )
+            reference_images = _require_mapping(
+                image.get("referenceImages"), f"referenceImages for {custom_id}"
+            )
+            if str(reference_images.get("referenceType") or "") != reference_type:
+                raise ValueError(
+                    f"Image {custom_id!r} referenceImages.referenceType does not "
+                    "match referenceType."
+                )
+            catalog_entry = catalog_by_type[reference_type]
+            for key in ("depthFileName", "segmentationFileName", "lineartFileName"):
+                if str(reference_images.get(key) or "") != str(
+                    catalog_entry.get(key) or ""
+                ):
+                    raise ValueError(
+                        f"Image {custom_id!r} does not use the catalog's {key}."
+                    )
+            dependencies = _require_list(
+                image.get("dependencies"), f"dependencies for {custom_id}"
+            )
+            if int(image.get("inputImageCount") or -1) != 4 + len(dependencies):
+                raise ValueError(f"Image {custom_id!r} has an invalid inputImageCount.")
+            if str(image.get("promptVersion") or "") != EXPECTED_PROMPT_VERSION:
+                raise ValueError(f"Image {custom_id!r} has the wrong promptVersion.")
+            # Do not normalize this string: Unity authored the complete request prompt,
+            # and the Batch request must receive it byte-for-byte as decoded JSON text.
+            prompt = str(image.get("prompt") or "")
+            if not prompt.strip():
+                raise ValueError(f"Image {custom_id!r} has an empty prompt.")
+            for required_text in (
+                "Image_A is the authoritative raw Unity render",
+                "Image_B is the",
+                "Image_C is the",
+                "Image_D is the",
+                "Do not swap left and right",
+            ):
+                if required_text not in prompt:
+                    raise ValueError(
+                        f"Image {custom_id!r} prompt is missing {required_text!r}."
+                    )
+            image["_batchSetNumber"] = set_number
+            image["_batchSetId"] = set_id
+            image["_sourceName"] = source_name
+            image["_targetName"] = target_name
+            image["_referenceImages"] = dict(reference_images)
+            image["_dependencies"] = [
+                dict(_require_mapping(dep, f"dependency for {custom_id}"))
+                for dep in dependencies
+            ]
+            image["_prompt"] = prompt
+            jobs.append(image)
+            jobs_by_id[custom_id] = image
+
+    if int(manifest.get("batchImageCount") or -1) != len(jobs):
+        raise ValueError("batchImageCount does not match the Batch image total.")
+
+    actual_dependencies_by_set: Dict[int, set] = {
+        number: set() for number in declared_dependencies_by_set
+    }
+
+    for job in jobs:
+        custom_id = str(job["customId"])
+        dependency_ids = set()
+        for dependency_index, dependency in enumerate(job["_dependencies"]):
+            dependency_id = str(dependency.get("customId") or "")
+            if not dependency_id or dependency_id in dependency_ids:
+                raise ValueError(
+                    f"Image {custom_id!r} has a missing or duplicate dependency."
+                )
+            dependency_ids.add(dependency_id)
+            source = jobs_by_id.get(dependency_id)
+            if source is None:
+                raise ValueError(
+                    f"Image {custom_id!r} depends on unknown image {dependency_id!r}."
+                )
+            if int(source["_batchSetNumber"]) >= int(job["_batchSetNumber"]):
+                raise ValueError(
+                    f"Image {custom_id!r} has a same-wave or forward dependency."
+                )
+            actual_dependencies_by_set[int(job["_batchSetNumber"])].add(
+                int(source["_batchSetNumber"])
+            )
+            expected_label = _image_label(4 + dependency_index)
+            if str(dependency.get("imageLabel") or "") != expected_label:
+                raise ValueError(
+                    f"Image {custom_id!r} dependency {dependency_index} must be "
+                    f"labeled {expected_label}."
+                )
+            if str(dependency.get("generatedFileName") or "") != str(
+                source["_targetName"]
+            ):
+                raise ValueError(
+                    f"Image {custom_id!r} has a mismatched dependency filename."
+                )
+            if str(dependency.get("locationId") or "") != str(
+                source.get("locationId") or ""
+            ):
+                raise ValueError(
+                    f"Image {custom_id!r} has a mismatched dependency locationId."
+                )
+            if str(dependency.get("sourceReferenceType") or "") != str(
+                source.get("referenceType") or ""
+            ):
+                raise ValueError(
+                    f"Image {custom_id!r} has a mismatched dependency "
+                    "sourceReferenceType."
+                )
+            if not str(dependency.get("relationship") or "").strip():
+                raise ValueError(
+                    f"Image {custom_id!r} has a dependency without a relationship."
+                )
+            movement = str(dependency.get("movementInstruction") or "").strip()
+            if not movement or movement not in str(job["_prompt"]):
+                raise ValueError(
+                    f"Image {custom_id!r} is missing its exact movement instruction."
+                )
+        expected_primary = (
+            str(job["_dependencies"][0].get("customId") or "")
+            if job["_dependencies"]
+            else ""
+        )
+        if str(job.get("primaryContinuityCustomId") or "") != expected_primary:
+            raise ValueError(
+                f"Image {custom_id!r} has an invalid primaryContinuityCustomId."
+            )
+        if int(job["_batchSetNumber"]) > 1 and not job["_dependencies"]:
+            raise ValueError(
+                f"Image {custom_id!r} is outside Batch set 1 but has no dependency."
+            )
+
+    for set_number, declared in declared_dependencies_by_set.items():
+        actual = sorted(actual_dependencies_by_set[set_number])
+        if declared != actual:
+            raise ValueError(
+                f"Batch set {set_number} declares dependencies {declared}, but its "
+                f"images actually require {actual}."
+            )
+
+    actual_reference_counts: Dict[str, int] = {
+        reference_type: 0 for reference_type in EXPECTED_REFERENCE_TYPES
+    }
+    for job in jobs:
+        actual_reference_counts[str(job["referenceType"])] += 1
+    for reference_type, entry in catalog_by_type.items():
+        if int(entry.get("usedByImageCount") or 0) != actual_reference_counts[
+            reference_type
+        ]:
+            raise ValueError(
+                f"Reference {reference_type!r} has an invalid usedByImageCount."
+            )
+    required_reference_types = sum(
+        1 for count in actual_reference_counts.values() if count > 0
+    )
+    expected_reference_counts = {
+        "referenceTypeCount": len(EXPECTED_REFERENCE_TYPES),
+        "referenceFileCount": len(EXPECTED_REFERENCE_TYPES) * 3,
+        "requiredReferenceTypeCount": required_reference_types,
+        "requiredReferenceFileCount": required_reference_types * 3,
+    }
+    for key, expected in expected_reference_counts.items():
+        if int(manifest.get(key) or -1) != expected:
+            raise ValueError(f"Manifest {key} is invalid; expected {expected}.")
+
+    raw_locations = _require_list(manifest.get("locations"), "locations")
+    if len(raw_locations) != len(jobs):
+        raise ValueError("locations must contain exactly one entry per Batch image.")
+    locations_by_custom_id: Dict[str, Mapping[str, Any]] = {}
+    for index, raw_location in enumerate(raw_locations):
+        location_entry = _require_mapping(raw_location, f"locations[{index}]")
+        custom_id = str(location_entry.get("customId") or "")
+        if not custom_id or custom_id in locations_by_custom_id:
+            raise ValueError(f"Missing or duplicate location customId {custom_id!r}.")
+        locations_by_custom_id[custom_id] = location_entry
+    if set(locations_by_custom_id) != set(jobs_by_id):
+        raise ValueError("locations and batchSets do not contain the same customIds.")
+    for custom_id, job in jobs_by_id.items():
+        location_entry = locations_by_custom_id[custom_id]
+        location = _require_mapping(
+            location_entry.get("location"), f"location object for {custom_id}"
+        )
+        if str(location.get("id") or "") != str(job.get("locationId") or ""):
+            raise ValueError(f"Location id mismatch for image {custom_id!r}.")
+        if str(location_entry.get("sourceScreenshotFileName") or "") != str(
+            job["_sourceName"]
+        ):
+            raise ValueError(f"Screenshot filename mismatch for image {custom_id!r}.")
+
+    first_set = [job for job in jobs if int(job["_batchSetNumber"]) == 1]
+    if not first_set or any(job["_dependencies"] for job in first_set):
+        raise ValueError(
+            "Batch set 1 must contain dependency-free entrance identity seeds."
+        )
+    allowed_seed_types = {"straight_door", "station_gate_Front_Direct"}
+    if any(str(job.get("referenceType") or "") not in allowed_seed_types for job in first_set):
+        raise ValueError(
+            "Batch set 1 may contain only straight-door and station-front seeds."
+        )
+    if int(manifest.get("entranceSeedCount") or -1) != len(first_set):
+        raise ValueError("entranceSeedCount does not match Batch set 1.")
+    raw_seed_ids = _require_list(
+        manifest.get("entranceSeedCustomIds"), "entranceSeedCustomIds"
+    )
+    seed_ids = [str(value or "") for value in raw_seed_ids]
+    if len(seed_ids) != len(set(seed_ids)) or set(seed_ids) != {
+        str(job["customId"]) for job in first_set
+    }:
+        raise ValueError("entranceSeedCustomIds does not identify all Batch-set-1 seeds.")
+    root_id = str(manifest.get("batchRootCustomId") or "")
+    root = jobs_by_id.get(root_id)
+    if root is None or int(root["_batchSetNumber"]) != 1:
+        raise ValueError("batchRootCustomId must identify a Batch-set-1 entrance seed.")
+    if str(manifest.get("batchRootLocationId") or "") != str(
+        root.get("locationId") or ""
+    ):
+        raise ValueError("batchRootLocationId does not match batchRootCustomId.")
+    if int(manifest.get("excludedBenchCount", -1)) != 0:
+        raise ValueError("The v5 city Batch manifest must exclude all benches.")
+    return jobs
+
+
+def _workflow_root(district: str, manifest_sha256: str) -> Path:
+    district_key = re.sub(r"[^A-Za-z0-9._-]+", "_", district).strip("._")
+    return (
+        _output_root()
+        / STATE_DIRECTORY_NAME
+        / (district_key or "district")
+        / manifest_sha256[:20]
+    )
+
+
+def _preview_inputs(
+    manifest: Mapping[str, Any],
+    job: Mapping[str, Any],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    generated_revision: str,
+) -> List[Dict[str, str]]:
+    references = _require_mapping(job.get("_referenceImages"), "referenceImages")
+    inputs: List[Dict[str, str]] = [
+        {
+            "label": "Image_A",
+            "role": "authoritative Unity target view",
+            "repoId": SOURCE_REPO_ID,
+            "revision": source_commit,
+            "path": _repo_path(district, str(job["_sourceName"])),
+        },
+        {
+            "label": "Image_B",
+            "role": "Depth structural reference",
+            "repoId": REFERENCE_REPO_ID,
+            "revision": reference_commit,
+            "path": str(references["depthFileName"]),
+        },
+        {
+            "label": "Image_C",
+            "role": "Segmentation structural reference",
+            "repoId": REFERENCE_REPO_ID,
+            "revision": reference_commit,
+            "path": str(references["segmentationFileName"]),
+        },
+        {
+            "label": "Image_D",
+            "role": "Lineart structural reference",
+            "repoId": REFERENCE_REPO_ID,
+            "revision": reference_commit,
+            "path": str(references["lineartFileName"]),
+        },
+    ]
+    for dependency in job["_dependencies"]:
+        inputs.append(
+            {
+                "label": str(dependency["imageLabel"]),
+                "role": "previously generated continuity view",
+                "repoId": GENERATED_REPO_ID,
+                "revision": generated_revision,
+                "path": _repo_path(district, str(dependency["generatedFileName"])),
+            }
+        )
+    for item in inputs:
+        item["url"] = _hf_url(item["repoId"], item["revision"], item["path"])
+    return inputs
+
+
+def _write_prompt_preview(
+    manifest: Mapping[str, Any],
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    manifest_sha256: str,
+) -> Tuple[Path, Path, str]:
+    root = _workflow_root(district, manifest_sha256)
+    entries: List[Dict[str, Any]] = []
+    text_parts: List[str] = []
+    for job in jobs:
+        inputs = _preview_inputs(
+            manifest,
+            job,
+            district,
+            source_commit,
+            reference_commit,
+            REPO_REVISION,
+        )
+        entry = {
+            "batchSetNumber": int(job["_batchSetNumber"]),
+            "batchSetId": str(job["_batchSetId"]),
+            "customId": str(job["customId"]),
+            "locationId": str(job.get("locationId") or ""),
+            "referenceType": str(job.get("referenceType") or ""),
+            "targetGeneratedFileName": str(job["_targetName"]),
+            "inputs": inputs,
+            "prompt": str(job["_prompt"]),
+        }
+        entries.append(entry)
+        text_parts.append(
+            "=" * 80
+            + f"\nBATCH SET {entry['batchSetNumber']:03d} | {entry['customId']}"
+            + f"\nTARGET: {entry['targetGeneratedFileName']}"
+            + "\n\nINPUTS\n"
+            + "\n".join(
+                f"{item['label']} = {item['url']} ({item['role']})"
+                for item in inputs
+            )
+            + "\n\nPROMPT\n"
+            + entry["prompt"]
+            + "\n"
+        )
+    preview = {
+        "formatVersion": 1,
+        "nodeVersion": NODE_VERSION,
+        "createdUtc": _utc_now(),
+        "districtName": district,
+        "manifestSha256": manifest_sha256,
+        "sourceCommit": source_commit,
+        "referenceCommit": reference_commit,
+        "generatedPreviewRevision": REPO_REVISION,
+        "batchSetCount": int(manifest.get("batchSetCount") or 0),
+        "imageCount": len(jobs),
+        "entries": entries,
+    }
+    json_path = root / "all_prompt_previews.json"
+    text_path = root / "all_prompt_previews.txt"
+    _write_json(json_path, preview)
+    full_text = "\n".join(text_parts)
+    text_path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = text_path.with_suffix(".txt.part")
+    temporary.write_text(full_text, encoding="utf-8")
+    os.replace(temporary, text_path)
+    return json_path, text_path, full_text
+
+
+def _openai_error(exception: urllib.error.HTTPError) -> str:
+    try:
+        body = exception.read().decode("utf-8", errors="replace")
+        parsed = json.loads(body)
+        error = parsed.get("error") if isinstance(parsed, dict) else None
+        if isinstance(error, Mapping):
+            return str(error.get("message") or body)[:1600]
+        return body[:1600]
+    except Exception:
+        return str(exception)
+
+
+def _openai_json(
+    method: str,
+    path: str,
+    openai_key: str,
+    payload: Optional[Mapping[str, Any]] = None,
+    timeout: int = 300,
+) -> Dict[str, Any]:
+    data = None
+    headers = {
+        "Authorization": "Bearer " + openai_key,
+        "User-Agent": f"Salia-City-Responses-Batch/{NODE_VERSION}",
+    }
+    if payload is not None:
+        data = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(
+        "https://api.openai.com" + path,
+        data=data,
+        headers=headers,
+        method=method,
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            parsed = json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exception:
+        raise RuntimeError(
+            f"OpenAI API {method} {path} failed: {_openai_error(exception)}"
+        ) from exception
+    if not isinstance(parsed, dict):
+        raise RuntimeError(f"OpenAI API {path} returned a non-object response.")
+    return parsed
+
+
+def _upload_openai_batch_file(path: Path, openai_key: str) -> Dict[str, Any]:
+    boundary = "----SaliaCityResponses" + secrets.token_hex(16)
+    body = bytearray()
+    body.extend(f"--{boundary}\r\n".encode("ascii"))
+    body.extend(b'Content-Disposition: form-data; name="purpose"\r\n\r\n')
+    body.extend(b"batch\r\n")
+    body.extend(f"--{boundary}\r\n".encode("ascii"))
+    body.extend(
+        (
+            'Content-Disposition: form-data; name="file"; filename="'
+            + path.name
+            + '"\r\n'
+        ).encode("utf-8")
+    )
+    body.extend(b"Content-Type: application/jsonl\r\n\r\n")
+    body.extend(path.read_bytes())
+    body.extend(b"\r\n")
+    body.extend(f"--{boundary}--\r\n".encode("ascii"))
+    request = urllib.request.Request(
+        "https://api.openai.com/v1/files",
+        data=bytes(body),
+        headers={
+            "Authorization": "Bearer " + openai_key,
+            "Content-Type": "multipart/form-data; boundary=" + boundary,
+            "User-Agent": f"Salia-City-Responses-Batch/{NODE_VERSION}",
+        },
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=300) as response:
+            parsed = json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exception:
+        raise RuntimeError(
+            "OpenAI Batch file upload failed: " + _openai_error(exception)
+        ) from exception
+    file_id = str(parsed.get("id") or "") if isinstance(parsed, dict) else ""
+    if not _SAFE_FILE_ID.fullmatch(file_id):
+        raise RuntimeError("OpenAI did not return a valid Batch input file id.")
+    return parsed
+
+
+def _download_openai_file(file_id: str, destination: Path, openai_key: str) -> Path:
+    if not _SAFE_FILE_ID.fullmatch(file_id):
+        raise ValueError(f"Invalid OpenAI file id {file_id!r}.")
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    temporary = destination.with_suffix(destination.suffix + ".part")
+    request = urllib.request.Request(
+        "https://api.openai.com/v1/files/"
+        + urllib.parse.quote(file_id, safe="")
+        + "/content",
+        headers={
+            "Authorization": "Bearer " + openai_key,
+            "User-Agent": f"Salia-City-Responses-Batch/{NODE_VERSION}",
+        },
+        method="GET",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=3600) as response, temporary.open(
+            "wb"
+        ) as output:
+            shutil.copyfileobj(response, output, length=1024 * 1024)
+        os.replace(temporary, destination)
+    except Exception:
+        if temporary.exists():
+            temporary.unlink()
+        raise
+    return destination
+
+
+def _batch_status(batch_id: str, openai_key: str) -> Dict[str, Any]:
+    if not _SAFE_BATCH_ID.fullmatch(batch_id):
+        raise ValueError("Saved OpenAI Batch id is invalid.")
+    return _openai_json(
+        "GET",
+        "/v1/batches/" + urllib.parse.quote(batch_id, safe=""),
+        openai_key,
+    )
+
+
+def _request_counts(batch: Mapping[str, Any]) -> Tuple[int, int, int]:
+    counts = batch.get("request_counts")
+    if not isinstance(counts, Mapping):
+        return 0, 0, 0
+    return (
+        int(counts.get("total") or 0),
+        int(counts.get("completed") or 0),
+        int(counts.get("failed") or 0),
+    )
+
+
+def _assistant_messages(body: Mapping[str, Any]) -> List[str]:
+    result: List[str] = []
+    output = body.get("output")
+    if not isinstance(output, list):
+        return result
+    for item in output:
+        if not isinstance(item, Mapping) or str(item.get("type") or "") != "message":
+            continue
+        content = item.get("content")
+        if not isinstance(content, list):
+            continue
+        for part in content:
+            if isinstance(part, Mapping):
+                text = str(part.get("text") or "").strip()
+                if text:
+                    result.append(text)
+    return result
+
+
+def _response_diagnostic(
+    record: Mapping[str, Any],
+    request_mapping: Mapping[str, Any],
+    line_number: int,
+) -> Dict[str, Any]:
+    response = record.get("response")
+    response_map = response if isinstance(response, Mapping) else {}
+    body_value = response_map.get("body")
+    body = body_value if isinstance(body_value, Mapping) else {}
+    output = body.get("output") if isinstance(body.get("output"), list) else []
+    calls: List[Dict[str, Any]] = []
+    reasoning_items: List[Dict[str, Any]] = []
+    for output_index, item in enumerate(output):
+        if not isinstance(item, Mapping):
+            continue
+        item_type = str(item.get("type") or "")
+        if item_type == "image_generation_call":
+            result_value = item.get("result")
+            calls.append(
+                {
+                    "outputIndex": output_index,
+                    "id": str(item.get("id") or ""),
+                    "status": str(item.get("status") or "unknown"),
+                    "revised_prompt": item.get("revised_prompt"),
+                    "revisedPromptPresent": "revised_prompt" in item,
+                    "resultPresent": isinstance(result_value, str)
+                    and bool(result_value),
+                    "resultBase64CharacterCount": len(result_value)
+                    if isinstance(result_value, str)
+                    else 0,
+                    "availableFields": sorted(str(key) for key in item.keys()),
+                    "allFieldsExceptResult": {
+                        str(key): value
+                        for key, value in item.items()
+                        if str(key) != "result"
+                    },
+                }
+            )
+        elif item_type == "reasoning":
+            reasoning_items.append(
+                {
+                    "outputIndex": output_index,
+                    "id": str(item.get("id") or ""),
+                    "status": str(item.get("status") or ""),
+                    "summary": item.get("summary"),
+                    "content": item.get("content"),
+                    "encryptedContentPresent": bool(item.get("encrypted_content")),
+                    "availableFields": sorted(str(key) for key in item.keys()),
+                }
+            )
+    reasoning = body.get("reasoning") if isinstance(body.get("reasoning"), Mapping) else {}
+    usage = body.get("usage") if isinstance(body.get("usage"), Mapping) else {}
+    output_details = (
+        usage.get("output_tokens_details")
+        if isinstance(usage.get("output_tokens_details"), Mapping)
+        else {}
+    )
+    tool_usage = (
+        body.get("tool_usage") if isinstance(body.get("tool_usage"), Mapping) else {}
+    )
+    image_usage = (
+        tool_usage.get("image_gen")
+        if isinstance(tool_usage.get("image_gen"), Mapping)
+        else {}
+    )
+    messages = _assistant_messages(body)
+    completed_image_present = any(
+        call.get("status") == "completed" and call.get("resultPresent")
+        for call in calls
+    )
+    return {
+        "batchOutputLine": line_number,
+        "batchCustomId": str(record.get("custom_id") or ""),
+        "jobCustomId": str(request_mapping.get("jobCustomId") or ""),
+        "batchSetNumber": int(request_mapping.get("batchSetNumber") or 0),
+        "attempt": int(request_mapping.get("attempt") or 0),
+        "targetPath": str(request_mapping.get("targetPath") or ""),
+        "responseStatusCode": int(response_map.get("status_code") or 0),
+        "batchRequestId": str(response_map.get("request_id") or ""),
+        "responseId": str(body.get("id") or ""),
+        "responseStatus": str(body.get("status") or ""),
+        "responseModel": str(body.get("model") or ""),
+        "responseError": body.get("error"),
+        "responseIncompleteDetails": body.get("incomplete_details"),
+        "responseModeration": body.get("moderation"),
+        "responseReasoning": dict(reasoning),
+        "reasoningEffort": reasoning.get("effort"),
+        "reasoningMode": reasoning.get("mode"),
+        "reasoningSummarySetting": reasoning.get("summary"),
+        "reasoningTokenCount": int(output_details.get("reasoning_tokens") or 0),
+        "responseUsage": dict(usage),
+        "imageGenUsage": dict(image_usage),
+        "imageGenerationCalls": calls,
+        "reasoningOutputItems": reasoning_items,
+        "assistantMessages": messages,
+        "assistantFailureMessage": "" if completed_image_present else "\n".join(messages),
+        "topLevelBatchError": record.get("error"),
+    }
+
+
+def _extract_single_png(body: Mapping[str, Any]) -> bytes:
+    if str(body.get("status") or "") != "completed" or body.get("error"):
+        raise ValueError(
+            "Responses request did not complete: "
+            + str(body.get("error") or body.get("status") or "unknown")
+        )
+    output = body.get("output")
+    if not isinstance(output, list):
+        raise ValueError("Responses result has no output array.")
+    encoded_results: List[str] = []
+    statuses: List[str] = []
+    for item in output:
+        if not isinstance(item, Mapping) or str(item.get("type") or "") != "image_generation_call":
+            continue
+        statuses.append(str(item.get("status") or "unknown"))
+        encoded = item.get("result")
+        if str(item.get("status") or "") == "completed" and isinstance(encoded, str) and encoded:
+            encoded_results.append(encoded)
+    if len(encoded_results) != 1:
+        raise ValueError(
+            f"Expected exactly one completed image result; received "
+            f"{len(encoded_results)}. Call statuses: {', '.join(statuses) or 'none'}."
+        )
+    raw = base64.b64decode(encoded_results[0], validate=True)
+    if not raw.startswith(_PNG_SIGNATURE):
+        raise ValueError("Generated result is not a PNG.")
+    if Image is None:
+        raise RuntimeError("Pillow is required to validate generated image dimensions.")
+    with Image.open(BytesIO(raw)) as image:
+        if image.size != (1536, 1024):
+            raise ValueError(
+                f"Generated PNG dimensions are {image.size}; expected (1536, 1024)."
+            )
+        image.verify()
+    return raw
+
+
+def _validate_remote_inputs(
+    api: Any,
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    generated_commit: Optional[str],
+    hf_token: str,
+) -> None:
+    missing: List[str] = []
+    checked = set()
+    for job in jobs:
+        candidates = [
+            (SOURCE_REPO_ID, source_commit, _repo_path(district, str(job["_sourceName"]))),
+            (
+                REFERENCE_REPO_ID,
+                reference_commit,
+                str(job["_referenceImages"]["depthFileName"]),
+            ),
+            (
+                REFERENCE_REPO_ID,
+                reference_commit,
+                str(job["_referenceImages"]["segmentationFileName"]),
+            ),
+            (
+                REFERENCE_REPO_ID,
+                reference_commit,
+                str(job["_referenceImages"]["lineartFileName"]),
+            ),
+        ]
+        for dependency in job["_dependencies"]:
+            if not generated_commit:
+                missing.append(
+                    GENERATED_REPO_ID
+                    + ":"
+                    + _repo_path(district, str(dependency["generatedFileName"]))
+                    + " (no generated repository commit)"
+                )
+            else:
+                candidates.append(
+                    (
+                        GENERATED_REPO_ID,
+                        generated_commit,
+                        _repo_path(district, str(dependency["generatedFileName"])),
+                    )
+                )
+        for repo_id, revision, path_in_repo in candidates:
+            key = (repo_id, revision, path_in_repo)
+            if key in checked:
+                continue
+            checked.add(key)
+            if not api.file_exists(
+                repo_id=repo_id,
+                filename=path_in_repo,
+                repo_type=REPO_TYPE,
+                revision=revision,
+                token=hf_token,
+            ):
+                missing.append(f"{repo_id}@{revision}:{path_in_repo}")
+    if missing:
+        raise FileNotFoundError(
+            "Cannot start this wave because required input files are missing: "
+            + "; ".join(missing[:30])
+        )
+
+
+def _batch_line(
+    request_mapping: Mapping[str, Any],
+    job: Mapping[str, Any],
+    inputs: Sequence[Mapping[str, str]],
+) -> Dict[str, Any]:
+    content: List[Dict[str, Any]] = [
+        {"type": "input_text", "text": str(job["_prompt"])}
+    ]
+    for item in inputs:
+        content.append(
+            {
+                "type": "input_text",
+                "text": str(item["label"])
+                + " follows: "
+                + str(item["role"])
+                + ".",
+            }
+        )
+        content.append({"type": "input_image", "image_url": str(item["url"])})
+    return {
+        "custom_id": str(request_mapping["batchCustomId"]),
+        "method": "POST",
+        "url": OPENAI_BATCH_ENDPOINT,
+        "body": {
+            "model": OPENAI_MODEL,
+            "input": [{"role": "user", "content": content}],
+            "reasoning": dict(OPENAI_REASONING),
+            "moderation": json.loads(json.dumps(RESPONSE_MODERATION)),
+            "tools": [dict(OPENAI_IMAGE_TOOL)],
+            "tool_choice": {"type": "image_generation"},
+            "store": False,
+        },
+    }
+
+
+def _build_wave_requests(
+    state: Mapping[str, Any],
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    generated_commit: str,
+    maximum_attempts: int,
+) -> Tuple[int, Dict[str, int], List[Dict[str, Any]], List[Dict[str, Any]]]:
+    if not jobs:
+        raise ValueError("Cannot build an empty city Batch wave.")
+    set_numbers = {int(job["_batchSetNumber"]) for job in jobs}
+    if len(set_numbers) != 1:
+        raise ValueError("One OpenAI Batch may contain only one logical Batch set.")
+    set_number = next(iter(set_numbers))
+    completed = set(str(value) for value in state.get("completedCustomIds") or [])
+    for job in jobs:
+        for dependency in job["_dependencies"]:
+            if str(dependency["customId"]) not in completed:
+                raise ValueError(
+                    f"Image {job['customId']!r} is not ready; dependency "
+                    f"{dependency['customId']!r} is unpublished."
+                )
+
+    attempt_counts = dict(state.get("attemptCounts") or {})
+    request_mappings: List[Dict[str, Any]] = []
+    lines: List[Dict[str, Any]] = []
+    for job_index, job in enumerate(jobs):
+        job_id = str(job["customId"])
+        attempt = int(attempt_counts.get(job_id) or 0) + 1
+        if attempt > maximum_attempts:
+            raise RuntimeError(
+                f"Image {job_id!r} reached MAX_ATTEMPTS={maximum_attempts}. "
+                "Inspect its saved diagnostics before trying again."
+            )
+        fingerprint = (
+            str(state["manifestSha256"])
+            + "\0"
+            + job_id
+            + "\0"
+            + str(attempt)
+        )
+        batch_custom_id = (
+            "city_"
+            + str(set_number).zfill(3)
+            + "_"
+            + str(job_index).zfill(4)
+            + "_a"
+            + str(attempt).zfill(2)
+            + "_"
+            + hashlib.sha256(fingerprint.encode("utf-8")).hexdigest()[:12]
+        )
+        mapping = {
+            "batchCustomId": batch_custom_id,
+            "jobCustomId": job_id,
+            "batchSetNumber": set_number,
+            "attempt": attempt,
+            "targetPath": _repo_path(district, str(job["_targetName"])),
+            "prompt": str(job["_prompt"]),
+        }
+        inputs = _preview_inputs(
+            state,
+            job,
+            district,
+            source_commit,
+            reference_commit,
+            generated_commit or REPO_REVISION,
+        )
+        mapping["inputs"] = inputs
+        request_mappings.append(mapping)
+        lines.append(_batch_line(mapping, job, inputs))
+        attempt_counts[job_id] = attempt
+    return set_number, attempt_counts, request_mappings, lines
+
+
+def _start_wave(
+    state: Dict[str, Any],
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    hf_token: str,
+    openai_key: str,
+    maximum_attempts: int,
+) -> Tuple[str, str, int, str]:
+    _, HfApi, _ = _load_hf_tools()
+    api = HfApi(token=hf_token)
+    generated_commit: Optional[str] = _repo_commit(
+        api, GENERATED_REPO_ID, hf_token
+    )
+    source_commit = str(state["sourceCommit"])
+    reference_commit = str(state["referenceCommit"])
+    (
+        set_number,
+        attempt_counts,
+        request_mappings,
+        lines,
+    ) = _build_wave_requests(
+        state,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        generated_commit or REPO_REVISION,
+        maximum_attempts,
+    )
+    _validate_remote_inputs(
+        api,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        generated_commit,
+        hf_token,
+    )
+
+    wave_sequence = int(state.get("openAiBatchSequence") or 0) + 1
+    batches_root = Path(str(state["workflowRoot"])) / "openai_batches"
+    while True:
+        run_root = batches_root / (
+            "batch_"
+            + str(wave_sequence).zfill(3)
+            + "_set_"
+            + str(set_number).zfill(3)
+        )
+        if not run_root.exists():
+            break
+        # A prior upload/create attempt may have failed before its Batch id could
+        # be persisted. Keep that evidence and use a fresh local run directory.
+        wave_sequence += 1
+    run_root.mkdir(parents=True, exist_ok=False)
+    jsonl_path = run_root / "city_responses_batch.jsonl"
+    request_map_path = run_root / "request_map.json"
+    _write_jsonl(jsonl_path, lines)
+    _write_json(
+        request_map_path,
+        {
+            "formatVersion": 1,
+            "nodeVersion": NODE_VERSION,
+            "districtName": district,
+            "manifestSha256": str(state["manifestSha256"]),
+            "batchSetNumber": set_number,
+            "requestCount": len(request_mappings),
+            "responsesModel": OPENAI_MODEL,
+            "reasoning": OPENAI_REASONING,
+            "responseModeration": RESPONSE_MODERATION,
+            "imageTool": OPENAI_IMAGE_TOOL,
+            "requests": request_mappings,
+        },
+    )
+    uploaded = _upload_openai_batch_file(jsonl_path, openai_key)
+    input_file_id = str(uploaded["id"])
+    batch = _openai_json(
+        "POST",
+        "/v1/batches",
+        openai_key,
+        {
+            "input_file_id": input_file_id,
+            "endpoint": OPENAI_BATCH_ENDPOINT,
+            "completion_window": "24h",
+            "metadata": {
+                "purpose": PURPOSE,
+                "node_version": NODE_VERSION,
+                "batch_set": str(set_number),
+                "manifest": str(state["manifestSha256"])[:32],
+            },
+        },
+    )
+    batch_id = str(batch.get("id") or "")
+    if not _SAFE_BATCH_ID.fullmatch(batch_id):
+        raise RuntimeError("OpenAI did not return a valid Batch id.")
+    state["attemptCounts"] = attempt_counts
+    state["openAiBatchSequence"] = wave_sequence
+    state["activeBatch"] = {
+        "batchId": batch_id,
+        "inputFileId": input_file_id,
+        "batchSetNumber": set_number,
+        "requestCount": len(request_mappings),
+        "requests": request_mappings,
+        "runRoot": str(run_root.resolve()),
+        "jsonlPath": str(jsonl_path.resolve()),
+        "requestMapPath": str(request_map_path.resolve()),
+        "createdUtc": _utc_now(),
+        "generatedInputCommit": generated_commit or "",
+    }
+    _write_json(Path(str(state["statePath"])), state)
+    return (
+        f"Started OpenAI Responses Batch {batch_id}: logical Batch set "
+        f"{set_number}, {len(jobs)} image request(s), attempts "
+        + ", ".join(
+            f"{mapping['jobCustomId']}={mapping['attempt']}"
+            for mapping in request_mappings[:8]
+        )
+        + (" ..." if len(request_mappings) > 8 else "")
+        + ".",
+        batch_id,
+        set_number,
+        str(jsonl_path.resolve()),
+    )
+
+
+def _upload_jsonl_artifact(
+    jsonl_path: Path,
+    district: str,
+    set_number: int,
+    manifest_sha256: str,
+    hf_token: str,
+) -> Tuple[str, str, str]:
+    raw = jsonl_path.read_bytes()
+    jsonl_sha256 = hashlib.sha256(raw).hexdigest()
+    file_name = (
+        "city_responses_"
+        + manifest_sha256[:12]
+        + "_set_"
+        + str(set_number).zfill(3)
+        + "_"
+        + jsonl_sha256[:12]
+        + ".jsonl"
+    )
+    path_in_repo = (
+        district
+        + "/openai_batch_jsonl/"
+        + _safe_file_name(file_name, "JSONL artifact filename")
+    )
+    CommitOperationAdd, HfApi, _ = _load_hf_tools()
+    api = HfApi(token=hf_token)
+    parent_commit = _repo_commit(api, SOURCE_REPO_ID, hf_token)
+    if api.file_exists(
+        repo_id=SOURCE_REPO_ID,
+        filename=path_in_repo,
+        repo_type=REPO_TYPE,
+        revision=parent_commit,
+        token=hf_token,
+    ):
+        commit_oid = parent_commit
+    else:
+        commit = api.create_commit(
+            repo_id=SOURCE_REPO_ID,
+            repo_type=REPO_TYPE,
+            revision=REPO_REVISION,
+            operations=[
+                CommitOperationAdd(
+                    path_in_repo=path_in_repo,
+                    path_or_fileobj=str(jsonl_path),
+                )
+            ],
+            commit_message=(
+                "City Responses JSONL only: Batch set "
+                + str(set_number)
+                + " ("
+                + jsonl_sha256[:12]
+                + ")"
+            ),
+            parent_commit=parent_commit,
+            token=hf_token,
+        )
+        commit_oid = str(getattr(commit, "oid", "") or "").strip()
+        if not commit_oid:
+            raise RuntimeError("Hugging Face JSONL commit returned no oid.")
+    return (
+        _hf_blob_url(SOURCE_REPO_ID, commit_oid, path_in_repo),
+        commit_oid,
+        path_in_repo,
+    )
+
+
+def _create_jsonl_only(
+    state: Dict[str, Any],
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    hf_token: str,
+    maximum_attempts: int,
+) -> Tuple[str, int, str, str]:
+    _, HfApi, _ = _load_hf_tools()
+    api = HfApi(token=hf_token)
+    generated_commit = _repo_commit(api, GENERATED_REPO_ID, hf_token)
+    source_commit = str(state["sourceCommit"])
+    reference_commit = str(state["referenceCommit"])
+    (
+        set_number,
+        _unused_attempt_counts,
+        request_mappings,
+        lines,
+    ) = _build_wave_requests(
+        state,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        generated_commit,
+        maximum_attempts,
+    )
+    _validate_remote_inputs(
+        api,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        generated_commit,
+        hf_token,
+    )
+
+    artifact_root = Path(str(state["workflowRoot"])) / "jsonl_only"
+    artifact_root.mkdir(parents=True, exist_ok=True)
+    jsonl_path = artifact_root / (
+        "city_responses_set_" + str(set_number).zfill(3) + ".jsonl"
+    )
+    request_map_path = artifact_root / (
+        "city_responses_set_" + str(set_number).zfill(3) + "_request_map.json"
+    )
+    _write_jsonl(jsonl_path, lines)
+    _write_json(
+        request_map_path,
+        {
+            "formatVersion": 1,
+            "nodeVersion": NODE_VERSION,
+            "onlyCreateJsonl": True,
+            "districtName": district,
+            "manifestSha256": str(state["manifestSha256"]),
+            "batchSetNumber": set_number,
+            "requestCount": len(request_mappings),
+            "responsesModel": OPENAI_MODEL,
+            "reasoning": OPENAI_REASONING,
+            "responseModeration": RESPONSE_MODERATION,
+            "imageTool": OPENAI_IMAGE_TOOL,
+            "requests": request_mappings,
+        },
+    )
+    url, commit_oid, path_in_repo = _upload_jsonl_artifact(
+        jsonl_path,
+        district,
+        set_number,
+        str(state["manifestSha256"]),
+        hf_token,
+    )
+    state["lastJsonlOnly"] = {
+        "createdUtc": _utc_now(),
+        "batchSetNumber": set_number,
+        "requestCount": len(request_mappings),
+        "localPath": str(jsonl_path.resolve()),
+        "requestMapPath": str(request_map_path.resolve()),
+        "huggingFaceRepoPath": path_in_repo,
+        "huggingFaceCommit": commit_oid,
+        "huggingFaceUrl": url,
+    }
+    _write_json(Path(str(state["statePath"])), state)
+    status = (
+        "Only_Create_Jsonl is ON. Created and uploaded logical Batch set "
+        + str(set_number)
+        + " with "
+        + str(len(lines))
+        + " /v1/responses request(s). No OpenAI file was uploaded and no "
+        + "OpenAI Batch was started. Image_A=Unity screenshot; Image_B=Depth; "
+        + "Image_C=Segmentation; Image_D=Lineart; Image_E and later=published "
+        + "generated continuity dependencies. Hugging Face JSONL: "
+        + url
+    )
+    return status, set_number, str(jsonl_path.resolve()), url
+
+
+def _save_png(raw: bytes, directory: Path, job_custom_id: str) -> Path:
+    directory.mkdir(parents=True, exist_ok=True)
+    name = hashlib.sha256(job_custom_id.encode("utf-8")).hexdigest()[:20] + ".png"
+    path = directory / name
+    temporary = path.with_suffix(".png.part")
+    temporary.write_bytes(raw)
+    os.replace(temporary, path)
+    return path
+
+
+def _parse_error_file(
+    path: Optional[Path],
+    request_by_batch_id: Mapping[str, Mapping[str, Any]],
+) -> Dict[str, Dict[str, Any]]:
+    result: Dict[str, Dict[str, Any]] = {}
+    if path is None or not path.is_file():
+        return result
+    with path.open("r", encoding="utf-8-sig") as handle:
+        for line_number, line in enumerate(handle, start=1):
+            if not line.strip():
+                continue
+            record = json.loads(line)
+            if not isinstance(record, Mapping):
+                continue
+            batch_custom_id = str(record.get("custom_id") or "")
+            if batch_custom_id not in request_by_batch_id:
+                continue
+            response = record.get("response")
+            response_map = response if isinstance(response, Mapping) else {}
+            error = record.get("error") or response_map.get("body") or response_map
+            result[batch_custom_id] = {
+                "batchErrorLine": line_number,
+                "statusCode": int(response_map.get("status_code") or 0),
+                "error": error,
+            }
+    return result
+
+
+def _publish_results(
+    state: Dict[str, Any],
+    results: Sequence[Mapping[str, Any]],
+    hf_token: str,
+) -> str:
+    if not results:
+        return ""
+    CommitOperationAdd, HfApi, _ = _load_hf_tools()
+    api = HfApi(token=hf_token)
+    parent_commit = _repo_commit(api, GENERATED_REPO_ID, hf_token)
+    operations = [
+        CommitOperationAdd(
+            path_in_repo=str(result["targetPath"]),
+            path_or_fileobj=str(result["localPath"]),
+        )
+        for result in results
+    ]
+    commit = api.create_commit(
+        repo_id=GENERATED_REPO_ID,
+        repo_type=REPO_TYPE,
+        revision=REPO_REVISION,
+        operations=operations,
+        commit_message=(
+            "City Responses Batch set "
+            + str(results[0]["batchSetNumber"])
+            + ": publish "
+            + str(len(results))
+            + " generated views"
+        ),
+        parent_commit=parent_commit,
+        token=hf_token,
+    )
+    oid = str(getattr(commit, "oid", "") or "").strip()
+    if not oid:
+        raise RuntimeError("Hugging Face commit returned no oid.")
+    state["lastGeneratedCommit"] = oid
+    return oid
+
+
+def _process_terminal_batch(
+    state: Dict[str, Any],
+    batch: Mapping[str, Any],
+    hf_token: str,
+    openai_key: str,
+) -> Tuple[int, int, str]:
+    active = _require_mapping(state.get("activeBatch"), "activeBatch")
+    run_root = Path(str(active["runRoot"]))
+    requests = _require_list(active.get("requests"), "activeBatch.requests")
+    request_by_batch_id: Dict[str, Mapping[str, Any]] = {
+        str(request["batchCustomId"]): _require_mapping(request, "request mapping")
+        for request in requests
+    }
+    output_path: Optional[Path] = None
+    output_file_id = str(batch.get("output_file_id") or "")
+    if _SAFE_FILE_ID.fullmatch(output_file_id):
+        output_path = _download_openai_file(
+            output_file_id,
+            run_root / "openai_batch_output.jsonl",
+            openai_key,
+        )
+    error_path: Optional[Path] = None
+    error_file_id = str(batch.get("error_file_id") or "")
+    if _SAFE_FILE_ID.fullmatch(error_file_id):
+        error_path = _download_openai_file(
+            error_file_id,
+            run_root / "openai_batch_errors.jsonl",
+            openai_key,
+        )
+    error_records = _parse_error_file(error_path, request_by_batch_id)
+    results: List[Dict[str, Any]] = []
+    failures: List[Dict[str, Any]] = []
+    diagnostics: List[Dict[str, Any]] = []
+    seen = set()
+    if output_path is not None:
+        with output_path.open("r", encoding="utf-8-sig") as handle:
+            for line_number, line in enumerate(handle, start=1):
+                if not line.strip():
+                    continue
+                record = json.loads(line)
+                if not isinstance(record, Mapping):
+                    raise ValueError(f"Batch output line {line_number} is not an object.")
+                batch_custom_id = str(record.get("custom_id") or "")
+                mapping = request_by_batch_id.get(batch_custom_id)
+                if mapping is None or batch_custom_id in seen:
+                    raise ValueError(
+                        f"Unexpected or duplicate Batch custom_id {batch_custom_id!r}."
+                    )
+                seen.add(batch_custom_id)
+                diagnostic = _response_diagnostic(record, mapping, line_number)
+                diagnostic["diagnosticIndex"] = len(diagnostics)
+                diagnostics.append(diagnostic)
+                try:
+                    response = _require_mapping(record.get("response"), "Batch response")
+                    status_code = int(response.get("status_code") or 0)
+                    if status_code < 200 or status_code >= 300:
+                        raise ValueError(f"Responses HTTP status {status_code}.")
+                    body = _require_mapping(response.get("body"), "Responses body")
+                    raw = _extract_single_png(body)
+                    local_path = _save_png(
+                        raw,
+                        run_root / "generated_images",
+                        str(mapping["jobCustomId"]),
+                    )
+                    results.append(
+                        {
+                            "batchCustomId": batch_custom_id,
+                            "jobCustomId": str(mapping["jobCustomId"]),
+                            "batchSetNumber": int(mapping["batchSetNumber"]),
+                            "attempt": int(mapping["attempt"]),
+                            "targetPath": str(mapping["targetPath"]),
+                            "localPath": str(local_path.resolve()),
+                            "sha256": hashlib.sha256(raw).hexdigest(),
+                            "byteCount": len(raw),
+                            "diagnosticIndex": int(diagnostic["diagnosticIndex"]),
+                        }
+                    )
+                except Exception as exception:
+                    failures.append(
+                        {
+                            "batchCustomId": batch_custom_id,
+                            "jobCustomId": str(mapping["jobCustomId"]),
+                            "batchSetNumber": int(mapping["batchSetNumber"]),
+                            "attempt": int(mapping["attempt"]),
+                            "failureKind": "response_or_image_failure",
+                            "reason": str(exception)[:1600],
+                            "diagnosticIndex": int(diagnostic["diagnosticIndex"]),
+                        }
+                    )
+    for batch_custom_id, mapping in request_by_batch_id.items():
+        if batch_custom_id in seen:
+            continue
+        error = error_records.get(batch_custom_id)
+        failures.append(
+            {
+                "batchCustomId": batch_custom_id,
+                "jobCustomId": str(mapping["jobCustomId"]),
+                "batchSetNumber": int(mapping["batchSetNumber"]),
+                "attempt": int(mapping["attempt"]),
+                "failureKind": "batch_request_error",
+                "reason": str(error.get("error") if error else batch.get("errors") or batch.get("status"))[:1600],
+                "batchError": error,
+            }
+        )
+
+    commit_oid = _publish_results(state, results, hf_token)
+    completed = set(str(value) for value in state.get("completedCustomIds") or [])
+    result_history = list(state.get("resultHistory") or [])
+    for result in results:
+        completed.add(str(result["jobCustomId"]))
+        result_history.append(dict(result, publishedCommit=commit_oid))
+    failure_history = list(state.get("failureHistory") or [])
+    failure_history.extend(failures)
+    state["completedCustomIds"] = sorted(completed)
+    state["resultHistory"] = result_history
+    state["failureHistory"] = failure_history
+    state["lastProcessedBatchId"] = str(active["batchId"])
+    state["activeBatch"] = None
+    _write_json(
+        run_root / "openai_response_diagnostics.json",
+        {
+            "formatVersion": 1,
+            "nodeVersion": NODE_VERSION,
+            "batchId": str(active["batchId"]),
+            "batchSetNumber": int(active["batchSetNumber"]),
+            "rawOutputPath": str(output_path.resolve()) if output_path else "",
+            "rawErrorPath": str(error_path.resolve()) if error_path else "",
+            "returnedOutputCount": len(seen),
+            "successfulImageCount": len(results),
+            "failureCount": len(failures),
+            "requestConfiguration": {
+                "responsesModel": OPENAI_MODEL,
+                "reasoning": OPENAI_REASONING,
+                "responseModeration": RESPONSE_MODERATION,
+                "imageTool": OPENAI_IMAGE_TOOL,
+                "toolChoice": {"type": "image_generation"},
+                "store": False,
+            },
+            "entries": diagnostics,
+            "failures": failures,
+        },
+    )
+    _write_json(Path(str(state["statePath"])), state)
+    return len(results), len(failures), commit_oid
+
+
+def _initialize_state(
+    manifest: Mapping[str, Any],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    manifest_sha256: str,
+    jobs: Sequence[Mapping[str, Any]],
+) -> Dict[str, Any]:
+    root = _workflow_root(district, manifest_sha256)
+    root.mkdir(parents=True, exist_ok=True)
+    state_path = root / "workflow_state.json"
+    state: Dict[str, Any] = {
+        "formatVersion": 1,
+        "nodeVersion": NODE_VERSION,
+        "createdUtc": _utc_now(),
+        "districtName": district,
+        "manifestSha256": manifest_sha256,
+        "sourceCommit": source_commit,
+        "referenceCommit": reference_commit,
+        "sourceRepoId": SOURCE_REPO_ID,
+        "generatedRepoId": GENERATED_REPO_ID,
+        "referenceRepoId": REFERENCE_REPO_ID,
+        "workflowRoot": str(root.resolve()),
+        "statePath": str(state_path.resolve()),
+        "logicalBatchSetCount": int(manifest.get("batchSetCount") or 0),
+        "totalImageCount": len(jobs),
+        "completedCustomIds": [],
+        "attemptCounts": {},
+        "openAiBatchSequence": 0,
+        "activeBatch": None,
+        "resultHistory": [],
+        "failureHistory": [],
+    }
+    _write_json(root / "uploaded_navigation_manifest.json", manifest)
+    _write_json(state_path, state)
+    return state
+
+
+def _load_or_initialize_state(
+    manifest: Mapping[str, Any],
+    jobs: Sequence[Mapping[str, Any]],
+    district: str,
+    source_commit: str,
+    reference_commit: str,
+    preview_json: Path,
+) -> Dict[str, Any]:
+    root = preview_json.parent
+    state_path = root / "workflow_state.json"
+    preview_data = _read_json(preview_json)
+    manifest_sha256 = str(preview_data["manifestSha256"])
+    if state_path.is_file():
+        state = _read_json(state_path)
+        if str(state.get("manifestSha256") or "") != manifest_sha256:
+            raise ValueError("Saved workflow identity does not match this manifest.")
+        has_started = bool(
+            state.get("activeBatch")
+            or state.get("completedCustomIds")
+            or state.get("attemptCounts")
+            or int(state.get("openAiBatchSequence") or 0) > 0
+        )
+        if not has_started and not isinstance(state.get("lastJsonlOnly"), Mapping):
+            # Refresh an untouched state, but keep commits pinned after a JSONL
+            # export. Uploading the artifact advances the source repository; if
+            # that new commit were substituted into Image_A URLs, an otherwise
+            # identical second export would produce different JSONL and would no
+            # longer match the file the user inspected.
+            state["sourceCommit"] = source_commit
+            state["referenceCommit"] = reference_commit
+            state["nodeVersion"] = NODE_VERSION
+            _write_json(state_path, state)
+        return state
+    return _initialize_state(
+        manifest,
+        district,
+        source_commit,
+        reference_commit,
+        manifest_sha256,
+        jobs,
+    )
+
+
+def _next_incomplete_set(
+    jobs: Sequence[Mapping[str, Any]],
+    completed_ids: Sequence[str],
+) -> Tuple[int, List[Mapping[str, Any]]]:
+    completed = set(str(value) for value in completed_ids)
+    incomplete = [job for job in jobs if str(job["customId"]) not in completed]
+    if not incomplete:
+        return 0, []
+    set_number = min(int(job["_batchSetNumber"]) for job in incomplete)
+    return (
+        set_number,
+        [job for job in incomplete if int(job["_batchSetNumber"]) == set_number],
+    )
+
+
+def _prepare(
+    district_name: str,
+    hf_key: str,
+) -> Tuple[
+    str,
+    Optional[str],
+    Dict[str, Any],
+    List[Dict[str, Any]],
+    str,
+    str,
+    Path,
+    Path,
+    str,
+]:
+    district = _safe_district_name(district_name)
+    hf_token = _optional_secret(hf_key)
+    manifest, source_commit, manifest_sha256, _ = _download_manifest(
+        district, hf_token
+    )
+    jobs = _validate_manifest(manifest, district)
+    _, HfApi, _ = _load_hf_tools()
+    reference_commit = _repo_commit(
+        HfApi(token=hf_token), REFERENCE_REPO_ID, hf_token
+    )
+    preview_json, preview_text, all_prompts = _write_prompt_preview(
+        manifest,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        manifest_sha256,
+    )
+    return (
+        district,
+        hf_token,
+        manifest,
+        jobs,
+        source_commit,
+        reference_commit,
+        preview_json,
+        preview_text,
+        all_prompts,
+    )
+
+
+def run_city_responses_batch(
+    action: str,
+    district_name: str,
+    hf_key: str,
+    openai_api_key: str,
+    maximum_attempts: int,
+    only_create_jsonl: bool = False,
+) -> Tuple[str, str, int, int, int, str, str]:
+    (
+        district,
+        optional_hf_token,
+        manifest,
+        jobs,
+        source_commit,
+        reference_commit,
+        preview_json,
+        preview_text,
+        all_prompts,
+    ) = _prepare(district_name, hf_key)
+    total = len(jobs)
+    if action not in {ACTION_PREVIEW, ACTION_RUN}:
+        raise ValueError(f"Unknown ACTION {action!r}.")
+    maximum_attempts = int(maximum_attempts)
+    if maximum_attempts < 1 or maximum_attempts > 50:
+        raise ValueError("MAX_ATTEMPTS must be between 1 and 50.")
+    if action == ACTION_PREVIEW and not only_create_jsonl:
+        return (
+            f"Preview only. Validated {total} prompts in "
+            f"{manifest['batchSetCount']} logical Batch sets. No OpenAI file or "
+            f"Batch was created. Read {preview_text.resolve()} before changing "
+            "ACTION to RUN_OR_CONTINUE.",
+            "",
+            0,
+            0,
+            total,
+            str(preview_text.resolve()),
+            all_prompts,
+        )
+    hf_token = _required_secret(hf_key, "Hugging Face token")
+    state = _load_or_initialize_state(
+        manifest,
+        jobs,
+        district,
+        source_commit,
+        reference_commit,
+        preview_json,
+    )
+
+    if only_create_jsonl:
+        completed_count = len(state.get("completedCustomIds") or [])
+        active_value = state.get("activeBatch")
+        if isinstance(active_value, Mapping):
+            jsonl_path = Path(str(active_value.get("jsonlPath") or ""))
+            if not jsonl_path.is_file():
+                raise FileNotFoundError(
+                    "The active Batch JSONL is missing locally: " + str(jsonl_path)
+                )
+            set_number = int(active_value.get("batchSetNumber") or 0)
+            url, commit_oid, path_in_repo = _upload_jsonl_artifact(
+                jsonl_path,
+                district,
+                set_number,
+                str(state["manifestSha256"]),
+                hf_token,
+            )
+            state["lastJsonlOnly"] = {
+                "createdUtc": _utc_now(),
+                "batchSetNumber": set_number,
+                "requestCount": int(active_value.get("requestCount") or 0),
+                "localPath": str(jsonl_path.resolve()),
+                "huggingFaceRepoPath": path_in_repo,
+                "huggingFaceCommit": commit_oid,
+                "huggingFaceUrl": url,
+                "source": "existing_active_batch",
+            }
+            _write_json(Path(str(state["statePath"])), state)
+            return (
+                "Only_Create_Jsonl is ON. Uploaded the exact JSONL already used "
+                "by active Batch "
+                + str(active_value.get("batchId") or "")
+                + ". No OpenAI API call was made. Image_A=Unity screenshot; "
+                "Image_B=Depth; Image_C=Segmentation; Image_D=Lineart; Image_E "
+                "and later=published generated continuity dependencies. "
+                "Hugging Face JSONL: "
+                + url,
+                str(active_value.get("batchId") or ""),
+                set_number,
+                completed_count,
+                total,
+                str(preview_text.resolve()),
+                all_prompts,
+            )
+        set_number, ready_jobs = _next_incomplete_set(
+            jobs, list(state.get("completedCustomIds") or [])
+        )
+        if not ready_jobs:
+            return (
+                "Only_Create_Jsonl is ON, but all city images are already "
+                "complete; there is no pending logical Batch set to export. "
+                "No OpenAI API call was made.",
+                "",
+                int(manifest["batchSetCount"]),
+                total,
+                total,
+                str(preview_text.resolve()),
+                all_prompts,
+            )
+        status, set_number, local_jsonl, _url = _create_jsonl_only(
+            state,
+            ready_jobs,
+            district,
+            hf_token,
+            maximum_attempts,
+        )
+        return (
+            status + " Local JSONL: " + local_jsonl,
+            "",
+            set_number,
+            completed_count,
+            total,
+            str(preview_text.resolve()),
+            all_prompts,
+        )
+
+    openai_key = _required_secret(openai_api_key, "OpenAI API key")
+
+    progress_note = ""
+    active_value = state.get("activeBatch")
+    if isinstance(active_value, Mapping):
+        batch_id = str(active_value.get("batchId") or "")
+        batch = _batch_status(batch_id, openai_key)
+        if str(batch.get("id") or "") != batch_id:
+            raise ValueError("OpenAI returned a different Batch identity.")
+        if str(batch.get("endpoint") or "") != OPENAI_BATCH_ENDPOINT:
+            raise ValueError("Saved OpenAI Batch does not target /v1/responses.")
+        if str(batch.get("input_file_id") or "") != str(
+            active_value.get("inputFileId") or ""
+        ):
+            raise ValueError("OpenAI returned a different Batch input file identity.")
+        metadata = batch.get("metadata")
+        if (
+            not isinstance(metadata, Mapping)
+            or str(metadata.get("purpose") or "") != PURPOSE
+            or str(metadata.get("batch_set") or "")
+            != str(active_value.get("batchSetNumber") or "")
+        ):
+            raise ValueError("OpenAI Batch metadata does not identify this city wave.")
+        status = str(batch.get("status") or "unknown")
+        total_requests, completed_requests, failed_requests = _request_counts(batch)
+        if status in _WAITING_BATCH_STATUSES:
+            return (
+                f"Waiting for Responses Batch {batch_id}: status={status}, "
+                f"total={total_requests}, completed={completed_requests}, "
+                f"failed={failed_requests}. Logical Batch set "
+                f"{active_value['batchSetNumber']} contains "
+                f"{active_value['requestCount']} request(s).",
+                batch_id,
+                int(active_value["batchSetNumber"]),
+                len(state.get("completedCustomIds") or []),
+                total,
+                str(preview_text.resolve()),
+                all_prompts,
+            )
+        if status not in _TERMINAL_BATCH_STATUSES:
+            raise ValueError(f"Unknown OpenAI Batch status {status!r}.")
+        success_count, failure_count, commit_oid = _process_terminal_batch(
+            state,
+            batch,
+            hf_token,
+            openai_key,
+        )
+        progress_note = (
+            f"Processed Batch {batch_id} ({status}): published {success_count}, "
+            f"will retry {failure_count}."
+            + (f" Hugging Face commit {commit_oid}." if commit_oid else "")
+            + " "
+        )
+
+    completed_ids = list(state.get("completedCustomIds") or [])
+    next_set_number, ready_jobs = _next_incomplete_set(jobs, completed_ids)
+    if not ready_jobs:
+        state["completedUtc"] = str(state.get("completedUtc") or _utc_now())
+        _write_json(Path(str(state["statePath"])), state)
+        return (
+            progress_note
+            + f"All done. Published all {total} city images across "
+            f"{manifest['batchSetCount']} logical Batch sets.",
+            str(state.get("lastProcessedBatchId") or ""),
+            int(manifest["batchSetCount"]),
+            total,
+            total,
+            str(preview_text.resolve()),
+            all_prompts,
+        )
+
+    start_status, batch_id, started_set, artifact = _start_wave(
+        state,
+        ready_jobs,
+        district,
+        hf_token,
+        openai_key,
+        maximum_attempts,
+    )
+    return (
+        progress_note
+        + start_status
+        + f" Progress: {len(completed_ids)}/{total}. JSONL: {artifact}",
+        batch_id,
+        started_set,
+        len(completed_ids),
+        total,
+        str(preview_text.resolve()),
+        all_prompts,
+    )
+
+
+class GPT_TO_HF_CityResponsesBatch:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "ACTION": (
+                    [ACTION_PREVIEW, ACTION_RUN],
+                    {"default": ACTION_PREVIEW},
+                ),
+                "Only_Create_Jsonl": (
+                    "BOOLEAN",
+                    {"default": False},
+                ),
+                "DISTRICT_NAME": (
+                    "STRING",
+                    {"default": "District_19_I", "multiline": False},
+                ),
+                "hf_key": (
+                    "STRING",
+                    {"default": "hf_xxx", "multiline": False},
+                ),
+                "openAi_key": (
+                    "STRING",
+                    {"default": "sk-xxx", "multiline": False},
+                ),
+                "MAX_ATTEMPTS": (
+                    "INT",
+                    {"default": 8, "min": 1, "max": 50, "step": 1},
+                ),
+            }
+        }
+
+    RETURN_TYPES = ("STRING", "STRING", "INT", "INT", "INT", "STRING", "STRING")
+    RETURN_NAMES = (
+        "status",
+        "batch_id",
+        "logical_batch_set",
+        "completed_images",
+        "total_images",
+        "prompt_preview_path",
+        "all_prompts",
+    )
+    FUNCTION = "run"
+    CATEGORY = "Salia/OpenAI Batch"
+    OUTPUT_NODE = True
+
+    @classmethod
+    def IS_CHANGED(cls, **kwargs):
+        return float("nan")
+
+    def run(
+        self,
+        ACTION: str,
+        DISTRICT_NAME: str,
+        hf_key: str,
+        openAi_key: str,
+        MAX_ATTEMPTS: int,
+        Only_Create_Jsonl: bool,
+    ):
+        return run_city_responses_batch(
+            ACTION,
+            DISTRICT_NAME,
+            hf_key,
+            openAi_key,
+            MAX_ATTEMPTS,
+            Only_Create_Jsonl,
+        )
+
+
+NODE_CLASS_MAPPINGS = {
+    "GPT_TO_HF_CityResponsesBatch": GPT_TO_HF_CityResponsesBatch,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "GPT_TO_HF_CityResponsesBatch": (
+        "GPT TO HF — City Responses Batch Sets (Preview First)"
+    ),
+}