Skip to content

fix(image-gen): Codex-auth images use the native images endpoints, no pinned chat model (#105398, #107076) #111000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
teknium1 merged 1 commit into from
Sep 14, 2026
Merged

fix(image-gen): Codex-auth images use the native images endpoints, no pinned chat model (#105398, #107076) #111000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
292 changes: 84 additions & 208 deletions plugins/image_gen/openai-codex/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,18 @@
"""OpenAI image generation — ChatGPT/Codex OAuth variant.

Same catalog/tiers as the ``openai`` plugin (``gpt-image-2`` low/medium/high), routed
through the Codex Responses API ``image_generation`` tool, so no ``OPENAI_API_KEY`` is
needed. Output is PNG; source images travel as Responses ``input_image`` parts.

Do NOT reintroduce an "account capability" classifier keyed on ``Tool choice
'image_generation' not found in 'tools' parameter``: that 400 is a request-shape
rejection for every account, fixed by omitting tool_choice (``_build_responses_payload``);
any remaining HTTP error must surface verbatim.
Same catalog/tiers as the ``openai`` plugin (``gpt-image-2`` low/medium/high), posted to the
Codex backend's native ``images/generations`` and ``images/edits`` endpoints, the same route the
official Codex client uses (``codex-rs/ext/image-generation``). No ``OPENAI_API_KEY`` is needed.

There is deliberately NO chat/host model here. An earlier version rode a Responses call with a
hosted ``image_generation`` tool on a pinned chat model (``gpt-5.5``): when OpenAI withdrew that id
from an account cohort every image call 404'd while chat kept working (#105398, #107076), and the
host model was free to answer in text instead of calling the tool. The native route has neither
failure mode.

The backend does not enforce ``model``/``quality``/``size`` — it accepts unknown model ids and
returns its own quality/size (#107233). We send the catalog values and report what came back
(``reported_quality``/``reported_size``) so a request that was not honoured is diagnosable.
"""

from __future__ import annotations
Expand All @@ -16,6 +21,7 @@
import json
import logging
import os
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

Expand All @@ -27,34 +33,14 @@

logger = logging.getLogger(__name__)

# NOTE: do NOT reintroduce an "account capability" classifier keyed on ``Tool choice 'image_generation' not
# found in 'tools' parameter``. That HTTP 400 is a *request-shape* rejection (the Codex backend resolves
# tool_choice as a function-tool name and never recognizes hosted-tool entries) — it is emitted for every
# account, including accounts where image generation works. A previous version of this file translated that
# 400 into "Image generation is not enabled for the current Codex account. Switch the image provider to
# OpenAI API key, FAL, or xAI.", which reported a universal bug in our own payload as the user's entitlement
# problem and sent people away from a provider that was never actually tried. The request-shape bug is fixed
# by omitting tool_choice (see ``_build_responses_payload``); any remaining HTTP error must surface verbatim
# so it stays diagnosable. See issues #19505, #49008 and #31335.
_MAX_ERROR_BODY_CHARS = 500

# Hosts the ``image_generation`` tool call; ``API_MODEL`` does the image work.
_CODEX_CHAT_MODEL = "gpt-5.5"
_CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
_CODEX_INSTRUCTIONS = (
"You are an assistant that must fulfill image generation and image editing "
"requests by using the image_generation tool when provided.")
_MAX_ERROR_BODY_CHARS = 500

_MAX_REFERENCE_IMAGES = 16
_MAX_INPUT_IMAGE_BYTES = 25 * 1024 * 1024
# ``input_image`` accepts raster only; the shared sniffer also knows SVG/TIFF/ICO, which the API rejects.
# The edit endpoint accepts raster only; the shared sniffer also knows SVG/TIFF/ICO, which it rejects.
_ACCEPTED_INPUT_MIME = frozenset({"image/png", "image/jpeg", "image/gif", "image/webp"})

# Progressive frames (partial_image_b64) saved as finals produced the "smear" failure mode:
# request 0 partials, never let a partial overwrite a final, only deliver source=final.
_PARTIAL_IMAGES_REQUESTED = 0
_NONFINAL_RETRIES = 1 # content-agnostic retries when the stream yields no final result

_NO_AUTH = (
"No Codex/ChatGPT OAuth credentials available. Run "
"`hermes auth codex` (or `hermes setup` → Codex) to sign in.")
Expand Down Expand Up @@ -129,17 +115,24 @@ def _data_url_to_input_image_url(value: str) -> str:
"Image data URL does not contain supported image bytes")


def _remote_image_to_data_url(value: str) -> str:
"""The edit endpoint takes inline data URLs only (as the official client sends), so fetch."""
import httpx

response = httpx.get(value, timeout=60.0, follow_redirects=True)
response.raise_for_status()
return _encode_input_image(
response.content,
f"Image URL exceeds 25MB cap: {value}",
f"Image URL did not return a supported image: {value}")


def _local_image_to_data_url(value: str) -> str:
try:
from agent.file_safety import get_read_block_error
from agent.file_safety import get_read_block_error

blocked = get_read_block_error(value)
if blocked:
raise ValueError(blocked)
except ValueError:
raise
except Exception as exc:
logger.debug("Codex image input read guard unavailable: %s", exc)
blocked = get_read_block_error(value)
if blocked:
raise ValueError(blocked)
path = Path(os.path.expanduser(value)).resolve()
if not path.is_file():
raise ValueError(f"Image input path does not exist or is not a file: {value}")
Expand All @@ -151,85 +144,71 @@ def _local_image_to_data_url(value: str) -> str:
f"Image input path is not a supported image: {value}")


def _to_input_image_part(value: str) -> Dict[str, str]:
"""Convert a URL/data URL/local path into a Responses input_image part."""
def _to_input_image(value: str) -> Dict[str, str]:
"""Convert a URL/data URL/local path into an ``images[]`` entry for ``images/edits``."""
candidate = (value or "").strip()
if not candidate:
raise ValueError("Blank image input")
lowered = candidate.lower()
if lowered.startswith(("http://", "https://")):
image_url = candidate
image_url = _remote_image_to_data_url(candidate)
elif lowered.startswith("data:"):
image_url = _data_url_to_input_image_url(candidate)
else:
image_url = _local_image_to_data_url(candidate)
return {"type": "input_image", "image_url": image_url}
return {"image_url": image_url}


def _normalize_input_images(
image_url: Optional[str], reference_image_urls: Optional[List[str]]
) -> List[Dict[str, str]]:
values = collect_source_images(image_url, reference_image_urls, limit=_MAX_REFERENCE_IMAGES)
return [_to_input_image_part(value) for value in values]
return [_to_input_image(value) for value in values]


def _build_responses_payload(
def _build_image_request(
*, prompt: str, size: str, quality: str, input_images: Optional[List[Dict[str, str]]] = None
) -> Dict[str, Any]:
"""Responses body for an image_generation call. No ``tool_choice``: Codex rejects every shape
for forcing the hosted tool (looks it up as a *function* name), so the host model decides,
nudged by ``instructions``."""
content: List[Dict[str, Any]] = [{"type": "input_text", "text": prompt}, *(input_images or [])]
return {
"model": _CODEX_CHAT_MODEL,
"store": False,
"instructions": _CODEX_INSTRUCTIONS,
"input": [{"type": "message", "role": "user", "content": content}],
"tools": [{
"type": "image_generation",
"model": API_MODEL,
"size": size,
"quality": quality,
"output_format": "png",
"background": "opaque",
"partial_images": _PARTIAL_IMAGES_REQUESTED,
}],
# No ``tool_choice`` is sent: the chatgpt.com/backend-api/codex backend rejects every shape we have
# for forcing the hosted ``image_generation`` tool. ``{"type": "allowed_tools", "mode": "required",
# "tools": [{"type": "image_generation"}]}`` (and the simpler ``{"type": "image_generation"}`` form)
# both 400 with ``Tool choice 'image_generation' not found in 'tools' parameter`` — the backend
# looks up tool_choice as a *function* name and never recognizes hosted-tool entries. Letting the
# host model decide is the only shape Codex currently accepts; the ``instructions`` above are what
# nudge it toward the tool. See issue #19505.
"stream": True,
) -> Tuple[str, Dict[str, Any]]:
"""``(endpoint_path, json_body)`` — ``images/edits`` when sources are present, else
``images/generations``. Field set mirrors the official client's ``ImageGenerationRequest`` /
``ImageEditRequest``."""
body: Dict[str, Any] = {
"prompt": prompt, "model": API_MODEL, "n": 1, "quality": quality, "size": size,
"background": "opaque",
}
if input_images:
body["images"] = input_images
return "images/edits", body
return "images/generations", body


def _extract_image_candidates(value: Any) -> Tuple[Optional[str], Optional[str]]:
"""``(final_result_b64, latest_partial_b64)`` from a payload tree; a partial never overwrites
a final."""
result_b64: Optional[str] = None
partial_b64: Optional[str] = None

def walk(node: Any) -> None:
nonlocal result_b64, partial_b64
if isinstance(node, dict):
result = node.get("result") if node.get("type") == "image_generation_call" else None
if isinstance(result, str) and result:
result_b64 = result
partial = node.get("partial_image_b64")
if isinstance(partial, str) and partial:
partial_b64 = partial
for child in node.values() if isinstance(node, dict) else node if isinstance(node, list) else ():
walk(child)

walk(value)
return result_b64, partial_b64

def _post_image_request(
token: str, *, prompt: str, size: str, quality: str, input_images: Optional[List[Dict[str, str]]] = None
) -> Dict[str, Any]:
"""POST to the native Codex images endpoint; return the decoded JSON body plus
``imagegen_request_id`` (backend correlation id, for support tickets)."""
import httpx
from agent.codex_headers import codex_cloudflare_headers

def _extract_image_b64(value: Any) -> Optional[str]:
"""Image b64 from a payload, preferring a final result over a partial."""
return next((b64 for b64 in _extract_image_candidates(value) if b64), None)
headers = codex_cloudflare_headers(token)
headers.update({
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"x-codex-image-turn-id": str(uuid.uuid4()),
})
path, body = _build_image_request(prompt=prompt, size=size, quality=quality, input_images=input_images)
timeout = httpx.Timeout(300.0, connect=30.0, read=300.0, write=60.0, pool=30.0)
with httpx.Client(timeout=timeout, headers=headers) as http:
response = http.post(f"{_CODEX_BASE_URL}/{path}", json=body)
if response.status_code >= 400:
raise RuntimeError(
f"Codex images API returned HTTP {response.status_code}: "
f"{_summarize_error_body(response.text)}")
payload = response.json()
if not isinstance(payload, dict):
raise RuntimeError("Codex images API returned a non-object body")
payload["imagegen_request_id"] = response.headers.get("x-codex-imagegen-request-id")
return payload


def _png_pixel_size(raw: bytes) -> Optional[str]:
Expand All @@ -242,81 +221,6 @@ def _png_pixel_size(raw: bytes) -> Optional[str]:
return f"{width}x{height}"


def _iter_sse_json(response: Any):
"""JSON payloads from an SSE response, without SDK parsing (events newer than the SDK still parse)."""
event_name: Optional[str] = None
data_lines: List[str] = []

def flush():
nonlocal event_name, data_lines
if not data_lines:
event_name = None
return None
raw = "\n".join(data_lines).strip()
event, event_name, data_lines = event_name, None, []
if not raw or raw == "[DONE]":
return None
payload = json.loads(raw)
if isinstance(payload, dict) and event and "type" not in payload:
payload["type"] = event
return payload

for line in response.iter_lines():
if isinstance(line, bytes):
line = line.decode("utf-8", errors="replace")
line = str(line)
if line == "":
payload = flush()
if payload is not None:
yield payload
elif line.startswith("event:"):
event_name = line[len("event:"):].strip()
elif line.startswith("data:"):
data_lines.append(line[len("data:"):].lstrip())
payload = flush()
if payload is not None:
yield payload


def _collect_image_b64(
token: str, *, prompt: str, size: str, quality: str, input_images: Optional[List[Dict[str, str]]] = None
) -> Optional[Dict[str, str]]:
"""Stream a Codex Responses image_generation call → ``{"b64", "source": "final"|"partial"}`` or
``None``. A partial is kept only when no final arrives; callers must not treat it as success."""
import httpx
from agent.codex_headers import codex_cloudflare_headers

headers = codex_cloudflare_headers(token)
headers.update({
"Accept": "text/event-stream",
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
})
payload = _build_responses_payload(
prompt=prompt, size=size, quality=quality, input_images=input_images)
timeout = httpx.Timeout(300.0, connect=30.0, read=300.0, write=30.0, pool=30.0)

final_b64: Optional[str] = None
partial_b64: Optional[str] = None
with httpx.Client(timeout=timeout, headers=headers) as http:
with http.stream("POST", f"{_CODEX_BASE_URL}/responses", json=payload) as response:
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
exc.response.read()
raise RuntimeError(
f"Codex Responses API returned HTTP {exc.response.status_code}: "
f"{_summarize_error_body(exc.response.text)}"
) from exc
for event in _iter_sse_json(response):
result_b64, event_partial = _extract_image_candidates(event)
final_b64 = result_b64 or final_b64
partial_b64 = event_partial or partial_b64
if final_b64:
return {"b64": final_b64, "source": "final"}
return {"b64": partial_b64, "source": "partial"} if partial_b64 else None


class OpenAICodexImageGenProvider(StaticImageGenProvider):
"""gpt-image-2 routed through ChatGPT/Codex OAuth instead of an API key."""

Expand Down Expand Up @@ -362,52 +266,22 @@ def generate(
tier_id, meta = _resolve_model()
size = size_for(aspect)
fail = error_factory("openai-codex", aspect, model=tier_id, prompt=prompt)
attempts = _NONFINAL_RETRIES + 1
try:
input_images = _normalize_input_images(image_url, reference_image_urls)
except Exception as exc:
return fail(f"Invalid image input for Codex image editing: {exc}", "invalid_image_input")

try:
collected: Optional[Dict[str, str]] = None
for attempt in range(attempts):
collected = _collect_image_b64(
token, prompt=prompt, size=size, quality=meta["quality"],
input_images=input_images or None)
if collected and collected.get("source") == "final" and collected.get("b64"):
break
if attempt < _NONFINAL_RETRIES:
partial = collected and collected.get("source") == "partial"
logger.warning(
"Codex image stream ended with %s (attempt %s/%s); "
"retrying once before failing closed.",
"progressive-only partial frame" if partial else "no image_generation_call result",
attempt + 1, attempts)
payload = _post_image_request(
token, prompt=prompt, size=size, quality=meta["quality"], input_images=input_images or None)
except Exception as exc:
logger.debug("Codex image generation failed", exc_info=True)
return fail(f"OpenAI image generation via Codex auth failed: {exc}", "api_error")

if not collected or not collected.get("b64"):
return fail(
f"Codex response contained no image_generation_call result after {attempts} attempt(s)",
"empty_response")
image_source = collected.get("source") or "unknown"
b64 = collected["b64"]
# Never deliver a progressive-only frame as success (smeared previews).
if image_source != "final":
try:
pixel_hint = _png_pixel_size(base64.b64decode(b64, validate=False))
except Exception:
pixel_hint = None
detail = (
"Codex returned only a progressive partial image frame after "
f"{attempts} attempt(s); refusing to save it as a final deliverable.")
if pixel_hint:
detail = f"{detail} partial_pixel_size={pixel_hint}."
return {
**fail(detail, "incomplete_image"), "image_source": image_source, "requested_size": size,
"partial_pixel_size": pixel_hint, "nonfinal_retries": _NONFINAL_RETRIES,
}
data = payload.get("data")
b64 = data[0].get("b64_json") if isinstance(data, list) and data and isinstance(data[0], dict) else None
if not isinstance(b64, str) or not b64:
return fail("Codex images API response contained no image data", "empty_response")

try:
pixel_size = _png_pixel_size(base64.b64decode(b64))
Expand All @@ -419,7 +293,9 @@ def generate(
provider="openai-codex", modality="image" if input_images else "text",
extra={
"size": size, "quality": meta["quality"], "input_image_count": len(input_images),
"image_source": image_source, "requested_size": size, "pixel_size": pixel_size,
"requested_size": size, "pixel_size": pixel_size,
"reported_quality": payload.get("quality"), "reported_size": payload.get("size"),
"imagegen_request_id": payload.get("imagegen_request_id"),
})


Expand Down
2 changes: 1 addition & 1 deletion plugins/image_gen/openai-codex/plugin.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name: openai-codex
version: 1.0.0
description: "OpenAI image generation backed by ChatGPT/Codex OAuth (gpt-image-2 via the Responses image_generation tool). Saves generated images to $HERMES_HOME/cache/images/."
description: "OpenAI image generation backed by ChatGPT/Codex OAuth (gpt-image-2 via the native Codex images/generations and images/edits endpoints). Saves generated images to $HERMES_HOME/cache/images/."
author: NousResearch
kind: backend
Loading
Loading