mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-28 18:05:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6
hermes-agent/plugins/image_gen/openai-codex/__init__.py
teknium1 064b6e40c5 fix(image-gen): stop reporting the Codex tool_choice 400 as an account limit 
The Codex image backend rejected our own request shape for every account, and
we then translated that rejection into "Image generation is not enabled for the
current Codex account. Switch the image provider to OpenAI API key, FAL, or
xAI." — telling every affected user to abandon a provider that had never
actually been tried. That message is why this reads as a setup failure rather
than a bug: the wire error was replaced with a confident, wrong diagnosis.

Removes the classifier and its exception, so any HTTP failure surfaces
verbatim. The paired request-shape fix (previous commit) is what makes the
400 stop happening; this commit makes the next one diagnosable.

Also fixes error-body truncation: bodies were head-truncated at 500 chars, and
Codex error payloads can carry hundreds of bytes of leading metadata, so the
user got a wall of padding and no message. _summarize_error_body() prefers the
parsed error.message and falls back to a truncated raw body.

Docs: drop the unqualified image-to-image claim for the Codex backend and note
that the hosted tool call cannot be forced, so it is best-effort.

Verified E2E against a local fake Codex backend: success path writes a real PNG
with no tool_choice on the wire; the 400 path now returns api_error carrying
"Tool choice 'image_generation' not found in 'tools' parameter" (148 chars)
instead of the entitlement message. Sabotage run confirms all 4 regression
tests fail when the old behavior is restored.

Refs #19505, #49008, #31335.
2026-07-25 16:41:57 -07:00

639 lines
22 KiB
Python
Raw Blame History

"""OpenAI image generation backend — ChatGPT/Codex OAuth variant.
Identical model catalog and tier semantics to the ``openai`` image-gen plugin
(``gpt-image-2`` at low/medium/high quality), but routes the request through
the Codex Responses API ``image_generation`` tool instead of the
``images.generate`` REST endpoint. This lets users who are already
authenticated with Codex/ChatGPT generate images without configuring a
separate ``OPENAI_API_KEY``.
Selection precedence for the tier (first hit wins):
1. ``OPENAI_IMAGE_MODEL`` env var (escape hatch for scripts / tests)
2. ``image_gen.openai-codex.model`` in ``config.yaml``
3. ``image_gen.model`` in ``config.yaml`` (when it's one of our tier IDs)
4. :data:`DEFAULT_MODEL` — ``gpt-image-2-medium``
Output is saved as PNG under ``$HERMES_HOME/cache/images/``. Source images for
image-to-image/editing are sent as Responses ``input_image`` content parts.
"""
from __future__ import annotations
import base64
import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from agent.image_gen_provider import (
DEFAULT_ASPECT_RATIO,
ImageGenProvider,
error_response,
normalize_reference_images,
resolve_aspect_ratio,
save_b64_image,
success_response,
)
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
def _summarize_error_body(body: str) -> str:
"""Return a bounded, information-preserving summary of an error body.
Prefers the parsed ``error.message`` field, because a blind head-truncation
of the raw body can cut the actual message off entirely — Codex error
payloads sometimes carry hundreds of bytes of leading metadata, so
``body[:500]`` yielded a wall of padding and no diagnosis. Falls back to a
truncated raw body for non-JSON responses.
"""
text = body or ""
try:
payload = json.loads(text)
error = payload.get("error") if isinstance(payload, dict) else None
message = error.get("message") if isinstance(error, dict) else None
if isinstance(message, str) and message.strip():
return message.strip()[:_MAX_ERROR_BODY_CHARS]
except (TypeError, ValueError):
pass
return text[:_MAX_ERROR_BODY_CHARS]
# ---------------------------------------------------------------------------
# Model catalog — mirrors the ``openai`` plugin so the picker UX is identical.
# ---------------------------------------------------------------------------
API_MODEL = "gpt-image-2"
_MODELS: Dict[str, Dict[str, Any]] = {
"gpt-image-2-low": {
"display": "GPT Image 2 (Low)",
"speed": "~15s",
"strengths": "Fast iteration, lowest cost",
"quality": "low",
},
"gpt-image-2-medium": {
"display": "GPT Image 2 (Medium)",
"speed": "~40s",
"strengths": "Balanced — default",
"quality": "medium",
},
"gpt-image-2-high": {
"display": "GPT Image 2 (High)",
"speed": "~2min",
"strengths": "Highest fidelity, strongest prompt adherence",
"quality": "high",
},
}
DEFAULT_MODEL = "gpt-image-2-medium"
_SIZES = {
"landscape": "1536x1024",
"square": "1024x1024",
"portrait": "1024x1536",
}
# Codex Responses surface used for the request. The chat model itself is only
# the host that calls the ``image_generation`` tool; the actual image work is
# done by ``API_MODEL``.
_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_REFERENCE_IMAGES = 16
_MAX_INPUT_IMAGE_BYTES = 25 * 1024 * 1024
# gpt-image-2's Responses ``input_image`` accepts raster formats only. The
# shared magic-byte sniffer also recognizes SVG/TIFF/ICO, which the API
# rejects server-side — gate to this allowlist so unsupported inputs fail
# locally with a clear error instead of an opaque HTTP 400.
_ACCEPTED_INPUT_MIME = frozenset(
{"image/png", "image/jpeg", "image/gif", "image/webp"}
)
# ---------------------------------------------------------------------------
# Config + auth helpers
# ---------------------------------------------------------------------------
def _load_image_gen_config() -> Dict[str, Any]:
"""Read ``image_gen`` from config.yaml (returns {} on any failure)."""
try:
from hermes_cli.config import load_config
cfg = load_config()
section = cfg.get("image_gen") if isinstance(cfg, dict) else None
return section if isinstance(section, dict) else {}
except Exception as exc:
logger.debug("Could not load image_gen config: %s", exc)
return {}
def _resolve_model() -> Tuple[str, Dict[str, Any]]:
"""Decide which tier to use and return ``(model_id, meta)``."""
import os
env_override = os.environ.get("OPENAI_IMAGE_MODEL")
if env_override and env_override in _MODELS:
return env_override, _MODELS[env_override]
cfg = _load_image_gen_config()
sub = cfg.get("openai-codex") if isinstance(cfg.get("openai-codex"), dict) else {}
candidate: Optional[str] = None
if isinstance(sub, dict):
value = sub.get("model")
if isinstance(value, str) and value in _MODELS:
candidate = value
if candidate is None:
top = cfg.get("model")
if isinstance(top, str) and top in _MODELS:
candidate = top
if candidate is not None:
return candidate, _MODELS[candidate]
return DEFAULT_MODEL, _MODELS[DEFAULT_MODEL]
def _read_codex_access_token() -> Optional[str]:
"""Return a usable Codex OAuth token, or None.
Delegates to the canonical reader in ``agent.auxiliary_client`` so token
expiry, credential pool selection, and JWT decoding stay in one place.
"""
try:
from agent.auxiliary_client import _read_codex_access_token as _reader
token = _reader()
if isinstance(token, str) and token.strip():
return token.strip()
return None
except Exception as exc:
logger.debug("Could not resolve Codex access token: %s", exc)
return None
def _sniff_image_mime(raw: bytes) -> Optional[str]:
"""Return a safe raster image MIME from magic bytes (not filename labels).
Delegates magic-byte detection to the shared sniffer in
``agent.image_routing`` (single source of truth), then gates the result
to :data:`_ACCEPTED_INPUT_MIME` — the raster formats gpt-image-2's
``input_image`` actually accepts. SVG/TIFF/ICO (which the shared sniffer
also recognizes) are rejected here so they fail locally with a clear
error instead of an opaque server-side HTTP 400.
"""
from agent.image_routing import _sniff_mime_from_bytes
mime = _sniff_mime_from_bytes(raw)
if mime in _ACCEPTED_INPUT_MIME:
return mime
return None
def _data_url_to_input_image_url(value: str) -> str:
"""Validate and canonicalize a data:image URL for Responses input_image."""
if "," not in value:
raise ValueError("Image data URL is missing a comma separator")
header, data = value.split(",", 1)
header_lc = header.lower()
if not header_lc.startswith("data:image/") or ";base64" not in header_lc:
raise ValueError("Only base64 data:image URLs are supported as Codex image inputs")
raw = base64.b64decode(data, validate=True)
if len(raw) > _MAX_INPUT_IMAGE_BYTES:
raise ValueError("Image data URL exceeds 25MB cap")
mime = _sniff_image_mime(raw)
if mime is None:
raise ValueError("Image data URL does not contain supported image bytes")
encoded = base64.b64encode(raw).decode("ascii")
return f"data:{mime};base64,{encoded}"
def _local_image_to_data_url(value: str) -> str:
"""Read a local image path and return a validated data:image URL."""
try:
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)
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}")
size = path.stat().st_size
if size <= 0:
raise ValueError(f"Image input path is empty: {value}")
if size > _MAX_INPUT_IMAGE_BYTES:
raise ValueError(f"Image input path exceeds 25MB cap: {value}")
raw = path.read_bytes()
mime = _sniff_image_mime(raw)
if mime is None:
raise ValueError(f"Image input path is not a supported image: {value}")
encoded = base64.b64encode(raw).decode("ascii")
return f"data:{mime};base64,{encoded}"
def _to_input_image_part(value: str) -> Dict[str, str]:
"""Convert a URL/data URL/local path into a Responses input_image part."""
candidate = (value or "").strip()
if not candidate:
raise ValueError("Blank image input")
lowered = candidate.lower()
if lowered.startswith("http://") or lowered.startswith("https://"):
image_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}
def _normalize_input_images(
image_url: Optional[str],
reference_image_urls: Optional[List[str]],
) -> List[Dict[str, str]]:
"""Collect primary + reference images as ordered Responses content parts."""
values: List[str] = []
if isinstance(image_url, str) and image_url.strip():
values.append(image_url.strip())
for ref in (normalize_reference_images(reference_image_urls) or []):
values.append(ref)
values = values[:_MAX_REFERENCE_IMAGES]
return [_to_input_image_part(value) for value in values]
def _build_responses_payload(
*,
prompt: str,
size: str,
quality: str,
input_images: Optional[List[Dict[str, str]]] = None,
) -> Dict[str, Any]:
"""Build the Codex Responses request body for an image_generation call."""
content: List[Dict[str, Any]] = [{"type": "input_text", "text": prompt}]
if input_images:
content.extend(input_images)
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": 1,
}],
# 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,
}
def _extract_image_b64(value: Any) -> Optional[str]:
"""Return the newest image b64 embedded in a Responses event payload."""
found: Optional[str] = None
if isinstance(value, dict):
if value.get("type") == "image_generation_call":
result = value.get("result")
if isinstance(result, str) and result:
found = result
partial = value.get("partial_image_b64")
if isinstance(partial, str) and partial:
found = partial
for child in value.values():
nested = _extract_image_b64(child)
if nested:
found = nested
elif isinstance(value, list):
for child in value:
nested = _extract_image_b64(child)
if nested:
found = nested
return found
def _iter_sse_json(response: Any):
"""Yield JSON payloads from an SSE response without OpenAI SDK parsing.
The ChatGPT/Codex backend can emit image-generation events newer than the
pinned Python SDK understands. Parsing raw SSE keeps this provider tolerant
of those event-shape changes.
"""
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
event_name = None
data_lines = []
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
continue
if line.startswith(":"):
continue
if 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[str]:
"""Stream a Codex Responses image_generation call and return the b64 image."""
import httpx
from agent.auxiliary_client 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)
image_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):
found = _extract_image_b64(event)
if found:
image_b64 = found
return image_b64
# ---------------------------------------------------------------------------
# Provider
# ---------------------------------------------------------------------------
class OpenAICodexImageGenProvider(ImageGenProvider):
"""gpt-image-2 routed through ChatGPT/Codex OAuth instead of an API key."""
@property
def name(self) -> str:
return "openai-codex"
@property
def display_name(self) -> str:
return "OpenAI (Codex auth)"
def is_available(self) -> bool:
if not _read_codex_access_token():
return False
try:
import httpx # noqa: F401
except ImportError:
return False
return True
def list_models(self) -> List[Dict[str, Any]]:
return [
{
"id": model_id,
"display": meta["display"],
"speed": meta["speed"],
"strengths": meta["strengths"],
"price": "varies",
}
for model_id, meta in _MODELS.items()
]
def default_model(self) -> Optional[str]:
return DEFAULT_MODEL
def get_setup_schema(self) -> Dict[str, Any]:
return {
"name": "OpenAI (Codex auth)",
"badge": "free",
"tag": "gpt-image-2 via ChatGPT/Codex OAuth — no API key required; supports text and image inputs",
"env_vars": [],
"post_setup_hint": (
"Sign in with `hermes auth codex` (or `hermes setup` → Codex) "
"if you haven't already. No API key needed."
),
}
def capabilities(self) -> Dict[str, Any]:
# The Codex Responses image_generation tool accepts source/reference
# images as `input_image` message content parts. Keep this capability
# honest so the dynamic `image_generate` schema encourages identity-
# preserving edits instead of unrelated text-to-image redraws.
return {"modalities": ["text", "image"], "max_reference_images": _MAX_REFERENCE_IMAGES}
def generate(
self,
prompt: str,
aspect_ratio: str = DEFAULT_ASPECT_RATIO,
*,
image_url: Optional[str] = None,
reference_image_urls: Optional[List[str]] = None,
**kwargs: Any,
) -> Dict[str, Any]:
prompt = (prompt or "").strip()
aspect = resolve_aspect_ratio(aspect_ratio)
if not prompt:
return error_response(
error="Prompt is required and must be a non-empty string",
error_type="invalid_argument",
provider="openai-codex",
aspect_ratio=aspect,
)
if not _read_codex_access_token():
return error_response(
error=(
"No Codex/ChatGPT OAuth credentials available. Run "
"`hermes auth codex` (or `hermes setup` → Codex) to sign in."
),
error_type="auth_required",
provider="openai-codex",
aspect_ratio=aspect,
)
try:
import httpx # noqa: F401
except ImportError:
return error_response(
error="httpx Python package not installed (pip install httpx)",
error_type="missing_dependency",
provider="openai-codex",
aspect_ratio=aspect,
)
tier_id, meta = _resolve_model()
size = _SIZES.get(aspect, _SIZES["square"])
token = _read_codex_access_token()
if not token:
return error_response(
error=(
"No Codex/ChatGPT OAuth credentials available. Run "
"`hermes auth codex` (or `hermes setup` → Codex) to sign in."
),
error_type="auth_required",
provider="openai-codex",
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
)
try:
input_images = _normalize_input_images(image_url, reference_image_urls)
except Exception as exc:
return error_response(
error=f"Invalid image input for Codex image editing: {exc}",
error_type="invalid_image_input",
provider="openai-codex",
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
)
try:
b64 = _collect_image_b64(
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 error_response(
error=f"OpenAI image generation via Codex auth failed: {exc}",
error_type="api_error",
provider="openai-codex",
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
)
if not b64:
return error_response(
error="Codex response contained no image_generation_call result",
error_type="empty_response",
provider="openai-codex",
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
)
try:
saved_path = save_b64_image(b64, prefix=f"openai_codex_{tier_id}")
except Exception as exc:
return error_response(
error=f"Could not save image to cache: {exc}",
error_type="io_error",
provider="openai-codex",
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
)
return success_response(
image=str(saved_path),
model=tier_id,
prompt=prompt,
aspect_ratio=aspect,
provider="openai-codex",
modality="image" if input_images else "text",
extra={"size": size, "quality": meta["quality"], "input_image_count": len(input_images)},
)
# ---------------------------------------------------------------------------
# Plugin entry point
# ---------------------------------------------------------------------------
def register(ctx) -> None:
"""Plugin entry point — register the Codex-backed image-gen provider."""
ctx.register_image_gen_provider(OpenAICodexImageGenProvider())
Reference in a new issue View git blame Copy permalink