diff --git a/acp_adapter/tools.py b/acp_adapter/tools.py index be4e49d013cea..303261e1e0ea7 100644 --- a/acp_adapter/tools.py +++ b/acp_adapter/tools.py @@ -50,6 +50,7 @@ "delegate_task": "execute", "vision_analyze": "read", "image_generate": "execute", + "image_edit": "execute", "text_to_speech": "execute", # Thinking / meta "_thinking": "think", @@ -65,7 +66,7 @@ "skill_view", "skills_list", "skill_manage", "web_search", "web_extract", "browser_navigate", "browser_click", "browser_type", "browser_press", "browser_scroll", "browser_back", "browser_snapshot", "browser_console", "browser_get_images", "browser_vision", - "vision_analyze", "image_generate", "text_to_speech", + "vision_analyze", "image_generate", "image_edit", "text_to_speech", # Schedulers / platform integrations "cronjob", "send_message", "clarify", "discord", "discord_admin", "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service", @@ -173,6 +174,9 @@ def build_tool_title(tool_name: str, args: Dict[str, Any]) -> str: if tool_name == "image_generate": prompt = str(args.get("prompt") or args.get("description") or "").strip() return f"generate image: {prompt[:50]}" if prompt else "generate image" + if tool_name == "image_edit": + prompt = str(args.get("prompt") or args.get("instruction") or "").strip() + return f"edit image: {prompt[:50]}" if prompt else "edit image" if tool_name == "cronjob": action = str(args.get("action") or "manage").strip() or "manage" job_id = str(args.get("job_id") or args.get("id") or "").strip() @@ -894,6 +898,7 @@ def _build_polished_completion_content( "browser_get_images": lambda: _format_browser_result(tool_name, result, function_args), "vision_analyze": lambda: _format_media_or_cron_result(tool_name, result), "image_generate": lambda: _format_media_or_cron_result(tool_name, result), + "image_edit": lambda: _format_media_or_cron_result(tool_name, result), "cronjob": lambda: _format_media_or_cron_result(tool_name, result), }.get(tool_name) if formatter is None and tool_name in _POLISHED_TOOLS: diff --git a/agent/display.py b/agent/display.py index 02880a83e0dc8..ac58e789927af 100644 --- a/agent/display.py +++ b/agent/display.py @@ -183,7 +183,7 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) - "read_file": "path", "write_file": "path", "patch": "path", "search_files": "pattern", "browser_navigate": "url", "browser_click": "ref", "browser_type": "text", - "image_generate": "prompt", "text_to_speech": "text", + "image_generate": "prompt", "image_edit": "prompt", "text_to_speech": "text", "vision_analyze": "question", "mixture_of_agents": "user_prompt", "skill_view": "name", "skills_list": "category", "cronjob": "action", @@ -999,6 +999,8 @@ def _wrap(line: str) -> str: return _wrap(f"┊ 📚 skill {_trunc(args.get('name', ''), 30)} {dur}") if tool_name == "image_generate": return _wrap(f"┊ 🎨 create {_trunc(args.get('prompt', ''), 35)} {dur}") + if tool_name == "image_edit": + return _wrap(f"┊ 🖌️ edit {_trunc(args.get('prompt', ''), 35)} {dur}") if tool_name == "text_to_speech": return _wrap(f"┊ 🔊 speak {_trunc(args.get('text', ''), 30)} {dur}") if tool_name == "vision_analyze": diff --git a/agent/image_gen_provider.py b/agent/image_gen_provider.py index a7f1b8c31ff95..ba178393117c3 100644 --- a/agent/image_gen_provider.py +++ b/agent/image_gen_provider.py @@ -142,6 +142,38 @@ def generate( should ignore unknown keys. """ + def supports_edit(self) -> bool: + """Return True when this backend supports reference-image editing. + + Providers that support image-to-image editing via :meth:`edit` should + override this to return ``True``. The default returns ``False`` so + existing providers stay source-compatible. + """ + return False + + def edit( + self, + prompt: str, + image: str, + aspect_ratio: str = DEFAULT_ASPECT_RATIO, + **kwargs: Any, + ) -> Dict[str, Any]: + """Edit an existing image using a text instruction. + + ``image`` is an HTTP(S) URL, data URL, or absolute local file path. + Providers that support image-to-image should override this method. + The default returns a uniform unsupported error so callers don't need + to probe :meth:`supports_edit` first. + """ + aspect = resolve_aspect_ratio(aspect_ratio) + return error_response( + error=f"Image editing is not supported by provider '{self.name}'", + error_type="unsupported_operation", + provider=self.name, + prompt=prompt or "", + aspect_ratio=aspect, + ) + # --------------------------------------------------------------------------- # Helpers diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index 9c36d205ac5bb..df15f5f032a9c 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -1254,6 +1254,7 @@ def build_nous_subscription_prompt(valid_tool_names: "set[str] | None" = None) - "browser_get_images", "browser_vision", "image_generate", + "image_edit", "text_to_speech", "terminal", "process", diff --git a/agent/transports/hermes_tools_mcp_server.py b/agent/transports/hermes_tools_mcp_server.py index 37f2d6179d117..c94be3b78364c 100644 --- a/agent/transports/hermes_tools_mcp_server.py +++ b/agent/transports/hermes_tools_mcp_server.py @@ -80,6 +80,7 @@ "browser_vision", "vision_analyze", "image_generate", + "image_edit", "skill_view", "skills_list", "text_to_speech", diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index cbe8a449d2731..14acfbdea6143 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -59,7 +59,7 @@ ("code_execution", "⚡ Code Execution", "execute_code"), ("vision", "👁️ Vision / Image Analysis", "vision_analyze"), ("video", "🎬 Video Analysis", "video_analyze (requires video-capable model)"), - ("image_gen", "🎨 Image Generation", "image_generate"), + ("image_gen", "🎨 Image Generation / Editing", "image_generate / image_edit"), ("video_gen", "🎬 Video Generation", "video_generate (text-to-video + image-to-video)"), ("x_search", "🐦 X (Twitter) Search", "x_search (requires xAI OAuth or XAI_API_KEY)"), ("moa", "🧠 Mixture of Agents", "mixture_of_agents"), diff --git a/model_tools.py b/model_tools.py index f461afff5ba4b..62070972d44c7 100644 --- a/model_tools.py +++ b/model_tools.py @@ -222,7 +222,7 @@ def _run_in_worker(): "terminal_tools": ["terminal"], "vision_tools": ["vision_analyze"], "moa_tools": ["mixture_of_agents"], - "image_tools": ["image_generate"], + "image_tools": ["image_generate", "image_edit"], "skills_tools": ["skills_list", "skill_view", "skill_manage"], "browser_tools": [ "browser_navigate", "browser_snapshot", "browser_click", diff --git a/plugins/image_gen/openai-codex/__init__.py b/plugins/image_gen/openai-codex/__init__.py index ab524dbdd7591..02bdf319cf7de 100644 --- a/plugins/image_gen/openai-codex/__init__.py +++ b/plugins/image_gen/openai-codex/__init__.py @@ -19,8 +19,13 @@ from __future__ import annotations +import base64 import logging +import mimetypes +import re +from pathlib import Path from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import urlparse from agent.image_gen_provider import ( DEFAULT_ASPECT_RATIO, @@ -78,6 +83,21 @@ "You are an assistant that must fulfill image generation requests by " "using the image_generation tool when provided." ) +_CODEX_EDIT_INSTRUCTIONS = ( + "You are an assistant that must edit the provided reference image by " + "using the image_generation tool when provided. Preserve visual details " + "the user did not ask to change." +) + +# Reference-image validation constants. +_ALLOWED_REFERENCE_IMAGE_MIME_TYPES = { + "image/png", + "image/jpeg", + "image/webp", + "image/gif", +} +_REFERENCE_IMAGE_MAX_BYTES = 20 * 1024 * 1024 # 20 MiB +_DATA_URL_HEADER_RE = re.compile(r"^data:([^;,]+)(?:;[^,]*)*$", re.IGNORECASE) # --------------------------------------------------------------------------- @@ -161,28 +181,45 @@ def _build_codex_client(): return None -def _collect_image_b64(client: Any, *, prompt: str, size: str, quality: str) -> Optional[str]: - """Stream a Codex Responses image_generation call and return the b64 image.""" +def _collect_image_b64_from_content( + client: Any, + *, + content: List[Dict[str, Any]], + size: str, + quality: str, + instructions: str, + action: Optional[str] = None, +) -> Optional[str]: + """Stream a Codex Responses image_generation call and return the b64 image. + + Shared implementation for both generation and editing — the caller + controls the input content, instructions, and optional ``action`` on the + image_generation tool. + """ image_b64: Optional[str] = None + tool: Dict[str, Any] = { + "type": "image_generation", + "model": API_MODEL, + "size": size, + "quality": quality, + "output_format": "png", + "background": "opaque", + "partial_images": 1, + } + if action: + tool["action"] = action + with client.responses.stream( model=_CODEX_CHAT_MODEL, store=False, - instructions=_CODEX_INSTRUCTIONS, + instructions=instructions, input=[{ "type": "message", "role": "user", - "content": [{"type": "input_text", "text": prompt}], - }], - tools=[{ - "type": "image_generation", - "model": API_MODEL, - "size": size, - "quality": quality, - "output_format": "png", - "background": "opaque", - "partial_images": 1, + "content": content, }], + tools=[tool], tool_choice={ "type": "allowed_tools", "mode": "required", @@ -214,6 +251,91 @@ def _collect_image_b64(client: Any, *, prompt: str, size: str, quality: str) -> return image_b64 +def _collect_image_b64(client: Any, *, prompt: str, size: str, quality: str) -> Optional[str]: + """Stream a Codex Responses image *generation* call and return the b64 image.""" + return _collect_image_b64_from_content( + client, + content=[{"type": "input_text", "text": prompt}], + size=size, + quality=quality, + instructions=_CODEX_INSTRUCTIONS, + ) + + +def _image_to_input_image_part(image: str) -> Dict[str, Any]: + """Convert a local path, HTTP(S) URL, or data URL into a Responses input_image part.""" + value = (image or "").strip() + if not value: + raise ValueError("image is required") + + parsed = urlparse(value) + if parsed.scheme in {"http", "https"}: + return {"type": "input_image", "image_url": value} + if parsed.scheme == "data": + return {"type": "input_image", "image_url": _validate_data_image_url(value)} + if parsed.scheme: + raise ValueError(f"Unsupported reference image URL scheme: {parsed.scheme}") + + path = Path(value).expanduser() + if not path.exists() or not path.is_file(): + raise FileNotFoundError(f"Reference image not found: {value}") + + raw, mime = _read_local_reference_image(path) + encoded = base64.b64encode(raw).decode("ascii") + return {"type": "input_image", "image_url": f"data:{mime};base64,{encoded}"} + + +def _validate_data_image_url(value: str) -> str: + """Validate a data URL header and return it unchanged, or raise.""" + m = _DATA_URL_HEADER_RE.match(value.split(",", 1)[0]) + if not m: + raise ValueError("Invalid data URL header") + mime = m.group(1).lower() + if mime not in _ALLOWED_REFERENCE_IMAGE_MIME_TYPES: + raise ValueError(f"Unsupported data URL MIME type: {mime}") + return value + + +def _read_local_reference_image(path: Path) -> tuple[bytes, str]: + """Read a local file, validate size and type, return (raw_bytes, mime_type).""" + if path.stat().st_size > _REFERENCE_IMAGE_MAX_BYTES: + raise ValueError( + f"Reference image exceeds {_REFERENCE_IMAGE_MAX_BYTES // (1024 * 1024)} MiB" + ) + + raw = path.read_bytes() + mime, _ = mimetypes.guess_type(str(path)) + if not mime or mime not in _ALLOWED_REFERENCE_IMAGE_MIME_TYPES: + raise ValueError( + f"Unsupported reference image type: {mime or 'unknown'}. " + f"Allowed: {', '.join(sorted(_ALLOWED_REFERENCE_IMAGE_MIME_TYPES))}" + ) + return raw, mime + + +def _collect_edited_image_b64( + client: Any, + *, + prompt: str, + image: str, + size: str, + quality: str, +) -> Optional[str]: + """Stream a Codex Responses image *edit* call and return the b64 image.""" + content = [ + {"type": "input_text", "text": prompt}, + _image_to_input_image_part(image), + ] + return _collect_image_b64_from_content( + client, + content=content, + size=size, + quality=quality, + instructions=_CODEX_EDIT_INSTRUCTIONS, + action="edit", + ) + + # --------------------------------------------------------------------------- # Provider # --------------------------------------------------------------------------- @@ -266,6 +388,9 @@ def get_setup_schema(self) -> Dict[str, Any]: ), } + def supports_edit(self) -> bool: + return True + def generate( self, prompt: str, @@ -367,6 +492,128 @@ def generate( extra={"size": size, "quality": meta["quality"]}, ) + def edit( + self, + prompt: str, + image: str, + aspect_ratio: str = DEFAULT_ASPECT_RATIO, + **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 isinstance(image, str) or not image.strip(): + return error_response( + error="A reference image path or URL is required", + error_type="invalid_argument", + provider="openai-codex", + prompt=prompt, + 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", + prompt=prompt, + aspect_ratio=aspect, + ) + + try: + import openai # noqa: F401 + except ImportError: + return error_response( + error="openai Python package not installed (pip install openai)", + error_type="missing_dependency", + provider="openai-codex", + prompt=prompt, + aspect_ratio=aspect, + ) + + tier_id, meta = _resolve_model() + size = _SIZES.get(aspect, _SIZES["square"]) + + client = _build_codex_client() + if client is None: + return error_response( + error="Could not initialize Codex image client", + error_type="auth_required", + provider="openai-codex", + model=tier_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + try: + b64 = _collect_edited_image_b64( + client, + prompt=prompt, + image=image, + size=size, + quality=meta["quality"], + ) + except (ValueError, FileNotFoundError, OSError) as exc: + return error_response( + error=str(exc), + error_type="invalid_argument", + provider="openai-codex", + model=tier_id, + prompt=prompt, + aspect_ratio=aspect, + ) + except Exception as exc: + logger.debug("Codex image edit failed", exc_info=True) + return error_response( + error=f"OpenAI image edit 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_edit_{tier_id}") + except Exception as exc: + return error_response( + error=f"Could not save edited 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", + extra={"size": size, "quality": meta["quality"], "source_image": image}, + ) + # --------------------------------------------------------------------------- # Plugin entry point diff --git a/tests/plugins/image_gen/test_openai_codex_provider.py b/tests/plugins/image_gen/test_openai_codex_provider.py index 3c8cf86c0a6fa..45609e9763b56 100644 --- a/tests/plugins/image_gen/test_openai_codex_provider.py +++ b/tests/plugins/image_gen/test_openai_codex_provider.py @@ -283,6 +283,106 @@ def _boom(**kwargs): assert "cloudflare 403" in result["error"] +# ── Edit ──────────────────────────────────────────────────────────────────── + + +class TestEdit: + def test_supports_edit(self, provider): + assert provider.supports_edit() is True + + def test_returns_auth_error_without_codex_token(self, provider, monkeypatch): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: None) + result = provider.edit("restore", image="/tmp/test.png") + assert result["success"] is False + assert result["error_type"] == "auth_required" + + def test_rejects_missing_image(self, provider, monkeypatch): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + result = provider.edit("restore", image="") + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + def test_rejects_non_string_image(self, provider, monkeypatch): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + result = provider.edit("restore", image=None) + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + def test_rejects_nonexistent_local_path(self, provider, monkeypatch, tmp_path): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + result = provider.edit("restore", image=str(tmp_path / "gone.png")) + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + def test_edit_uses_codex_stream_with_input_image(self, provider, monkeypatch, tmp_path): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + + # Create a tiny valid PNG as reference + source = tmp_path / "source.png" + source.write_bytes(bytes.fromhex(_PNG_HEX)) + + captured = {} + + def _stream(**kwargs): + captured.update(kwargs) + output_item = SimpleNamespace( + type="image_generation_call", + result=_b64_png(), + ) + done_event = SimpleNamespace(type="response.output_item.done", item=output_item) + return _FakeStream([done_event], SimpleNamespace(output=[])) + + fake_client = SimpleNamespace(responses=SimpleNamespace(stream=_stream)) + monkeypatch.setattr(codex_plugin, "_build_codex_client", lambda: fake_client) + + result = provider.edit("restore", image=str(source), aspect_ratio="landscape") + assert result["success"] is True + assert result["model"] == "gpt-image-2-medium" + assert result["provider"] == "openai-codex" + + # Verify edit-specific request shape + tool = captured["tools"][0] + assert tool["action"] == "edit" + assert tool["model"] == "gpt-image-2" + + content = captured["input"][0]["content"] + assert content[0]["type"] == "input_text" + assert "restore" in content[0]["text"] + assert content[1]["type"] == "input_image" + assert content[1]["image_url"].startswith("data:image/png;base64,") + + def test_edit_saves_with_edit_prefix(self, provider, monkeypatch, tmp_path): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + + source = tmp_path / "source.png" + source.write_bytes(bytes.fromhex(_PNG_HEX)) + + output_item = SimpleNamespace( + type="image_generation_call", + result=_b64_png(), + ) + done_event = SimpleNamespace(type="response.output_item.done", item=output_item) + fake_client = SimpleNamespace( + responses=SimpleNamespace( + stream=lambda **kwargs: _FakeStream([done_event], SimpleNamespace(output=[])) + ) + ) + monkeypatch.setattr(codex_plugin, "_build_codex_client", lambda: fake_client) + + result = provider.edit("restore", image=str(source)) + assert result["success"] is True + + saved = Path(result["image"]) + assert saved.exists() + assert saved.name.startswith("openai_codex_edit_") + + def test_empty_prompt_returns_error(self, provider, monkeypatch): + monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token") + result = provider.edit(" ", image="/tmp/test.png") + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + # ── Plugin entry point ────────────────────────────────────────────────────── diff --git a/tests/tools/test_image_edit_tool.py b/tests/tools/test_image_edit_tool.py new file mode 100644 index 0000000000000..3589e951adf48 --- /dev/null +++ b/tests/tools/test_image_edit_tool.py @@ -0,0 +1,195 @@ +"""Tests for the image_edit tool schema, handler, and registry integration.""" + +from __future__ import annotations + +import json +from typing import Any, Dict, cast + +import pytest + + +# ── Import the tool so side-effects fire (registry registration). ── +import tools.image_edit_tool as edit_tool # noqa: E402 + + +# --------------------------------------------------------------------------- +# Schema tests +# --------------------------------------------------------------------------- + + +class TestImageEditSchema: + def test_name_is_image_edit(self): + assert edit_tool.IMAGE_EDIT_SCHEMA["name"] == "image_edit" + + def test_required_fields_are_prompt_and_image(self): + params = cast(dict, edit_tool.IMAGE_EDIT_SCHEMA["parameters"]) + assert set(params["required"]) == {"prompt", "image"} + + def test_prompt_is_string(self): + params = cast(dict, edit_tool.IMAGE_EDIT_SCHEMA["parameters"]) + assert params["properties"]["prompt"]["type"] == "string" + + def test_image_is_string(self): + params = cast(dict, edit_tool.IMAGE_EDIT_SCHEMA["parameters"]) + assert params["properties"]["image"]["type"] == "string" + + def test_aspect_ratio_has_valid_default(self): + params = cast(dict, edit_tool.IMAGE_EDIT_SCHEMA["parameters"]) + ar = params["properties"]["aspect_ratio"] + assert ar["default"] == "landscape" + assert "landscape" in ar["enum"] + assert "square" in ar["enum"] + assert "portrait" in ar["enum"] + + +# --------------------------------------------------------------------------- +# Handler tests +# --------------------------------------------------------------------------- + + +class TestImageEditHandler: + def test_missing_prompt_returns_error(self): + result = json.loads(edit_tool._handle_image_edit( + {"image": "/tmp/test.png"} + )) + assert result["success"] is False + assert "prompt" in result["error"].lower() + + def test_missing_image_returns_error(self): + result = json.loads(edit_tool._handle_image_edit( + {"prompt": "restore"} + )) + assert result["success"] is False + assert "image" in result["error"].lower() + + def test_both_missing_returns_error(self): + result = json.loads(edit_tool._handle_image_edit({})) + assert result["success"] is False + + def test_falls_back_to_unsupported_when_no_plugin_provider(self, monkeypatch): + monkeypatch.setattr(edit_tool, "_read_configured_image_provider", lambda: None) + result = json.loads(edit_tool._handle_image_edit( + {"prompt": "restore", "image": "/tmp/test.png"} + )) + assert result["success"] is False + assert result["error_type"] == "unsupported_operation" + + +# --------------------------------------------------------------------------- +# Registry tests +# --------------------------------------------------------------------------- + + +class TestRegistry: + def test_image_edit_registered(self): + from tools.registry import registry + entry = registry._tools.get("image_edit") + assert entry is not None, "image_edit should be in the tool registry" + assert entry.schema["name"] == "image_edit" + + def test_image_edit_is_in_image_gen_toolset(self): + from tools.registry import registry + entry = registry._tools.get("image_edit") + assert entry is not None + assert entry.toolset == "image_gen" + + def test_image_generate_still_registered(self): + """Sanity check — image_generate must not be broken.""" + import tools.image_generation_tool # noqa: F401 — ensure it registers + from tools.registry import registry + entry = registry._tools.get("image_generate") + assert entry is not None + + +# --------------------------------------------------------------------------- +# Dispatch tests (with fake provider) +# --------------------------------------------------------------------------- + + +class _FakeEditProvider: + """Minimal provider stub that supports editing.""" + name = "fake-edit" + last_edit: dict | None = None + + def supports_edit(self) -> bool: + return True + + def edit(self, prompt, image, aspect_ratio="landscape", **kwargs): + self.last_edit = { + "prompt": prompt, + "image": image, + "aspect_ratio": aspect_ratio, + } + return { + "success": True, + "image": "/tmp/edit.png", + "model": "test-model", + "prompt": prompt, + "aspect_ratio": aspect_ratio, + "provider": "fake-edit", + } + + +class _FakeNoEditProvider: + """Provider that does not support editing.""" + name = "fake-noedit" + + def supports_edit(self) -> bool: + return False + + def generate(self, prompt, **kwargs): + return { + "success": True, "image": "/tmp/gen.png", "model": "t", + "prompt": prompt, "aspect_ratio": "landscape", "provider": "fake-noedit", + } + + +class TestDispatch: + def test_calls_edit_on_supporting_provider(self, monkeypatch): + fake = _FakeEditProvider() + monkeypatch.setattr( + edit_tool, "_read_configured_image_provider", lambda: "fake-edit" + ) + # _dispatch_edit imports from hermes_cli.plugins — patch there + try: + import hermes_cli.plugins as plugin_mod + monkeypatch.setattr(plugin_mod, "_ensure_plugins_discovered", lambda force=None: None) + except ImportError: + pass + + import agent.image_gen_registry as reg + monkeypatch.setattr( + reg, "get_provider", + lambda name: fake if name == "fake-edit" else None, + ) + + result = json.loads(edit_tool._handle_image_edit( + {"prompt": "restore", "image": "/tmp/source.png"} + )) + assert result["success"] is True + assert result["image"] == "/tmp/edit.png" + assert fake.last_edit["prompt"] == "restore" + assert fake.last_edit["image"] == "/tmp/source.png" + + def test_returns_unsupported_for_noedit_provider(self, monkeypatch): + fake = _FakeNoEditProvider() + monkeypatch.setattr( + edit_tool, "_read_configured_image_provider", lambda: "fake-noedit" + ) + try: + import hermes_cli.plugins as plugin_mod + monkeypatch.setattr(plugin_mod, "_ensure_plugins_discovered", lambda force=None: None) + except ImportError: + pass + + import agent.image_gen_registry as reg + monkeypatch.setattr( + reg, "get_provider", + lambda name: fake if name == "fake-noedit" else None, + ) + + result = json.loads(edit_tool._handle_image_edit( + {"prompt": "restore", "image": "/tmp/source.png"} + )) + assert result["success"] is False + assert result["error_type"] == "unsupported_operation" diff --git a/tools/image_edit_tool.py b/tools/image_edit_tool.py new file mode 100644 index 0000000000000..3b9cb19686718 --- /dev/null +++ b/tools/image_edit_tool.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""Image editing tool. + +Provides a provider-dispatched tool for prompt-guided image-to-image +editing. The first local implementation targets the openai-codex image_gen +backend, which passes reference images through the Codex Responses API. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any, Dict + +from agent.image_gen_provider import DEFAULT_ASPECT_RATIO, VALID_ASPECT_RATIOS + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + +IMAGE_EDIT_SCHEMA = { + "name": "image_edit", + "description": ( + "Edit an existing image using a text instruction and a reference image. " + "Mandatory: when the user provides, uploads, links, or names any " + "reference/source/product/person image, use this image-to-image tool " + "rather than image_generate/text-to-image. " + "The active image backend is user-configured. Currently this is intended " + "for backends that support image-to-image editing, such as OpenAI Codex " + "auth with GPT Image 2. Returns either a URL or an absolute file path in " + "the ``image`` field; display it with markdown " + "![description](url-or-path)." + ), + "parameters": { + "type": "object", + "properties": { + "prompt": { + "type": "string", + "description": ( + "Instruction describing the edit to apply while preserving " + "unchanged parts of the source image." + ), + }, + "image": { + "type": "string", + "description": ( + "Reference image as an HTTP(S) URL, data URL, or absolute " + "local file path." + ), + }, + "aspect_ratio": { + "type": "string", + "enum": list(VALID_ASPECT_RATIOS), + "description": ( + "Desired output aspect ratio. 'landscape' is 16:9 wide, " + "'portrait' is 16:9 tall, 'square' is 1:1." + ), + "default": DEFAULT_ASPECT_RATIO, + }, + }, + "required": ["prompt", "image"], + }, +} + + +# --------------------------------------------------------------------------- +# Plugin dispatch +# --------------------------------------------------------------------------- + + +def _read_configured_image_provider() -> str | None: + """Return the value of ``image_gen.provider`` from config.yaml, or None.""" + try: + from hermes_cli.config import load_config + + cfg = load_config() + section = cfg.get("image_gen") if isinstance(cfg, dict) else None + if isinstance(section, dict): + value = section.get("provider") + if isinstance(value, str) and value.strip(): + return value.strip() + except Exception as exc: + logger.debug("Could not read image_gen.provider: %s", exc) + return None + + +def _dispatch_edit(prompt: str, image: str, aspect_ratio: str) -> str | None: + """Route the edit call to a plugin-registered provider. + + Returns a JSON string on dispatch, or ``None`` to fall through (which + yields an unsupported error for editing since FAL can't edit). + """ + configured = _read_configured_image_provider() + if not configured or configured == "fal": + return None + + try: + from agent.image_gen_registry import get_provider + from hermes_cli.plugins import _ensure_plugins_discovered + + _ensure_plugins_discovered() + provider = get_provider(configured) + except Exception as exc: + logger.debug("image_edit plugin dispatch skipped: %s", exc) + return None + + if provider is None: + try: + _ensure_plugins_discovered(force=True) + provider = get_provider(configured) + except Exception as exc: + logger.debug("image_edit plugin force-refresh skipped: %s", exc) + + if provider is None: + return json.dumps({ + "success": False, + "image": None, + "error": ( + f"image_gen.provider='{configured}' is set but no plugin " + f"registered that name. Run `hermes plugins list` to see " + f"available image gen backends." + ), + "error_type": "provider_not_registered", + }) + + if not getattr(provider, "supports_edit", lambda: False)(): + return json.dumps({ + "success": False, + "image": None, + "error": ( + f"Image editing is not supported by provider " + f"'{getattr(provider, 'name', configured)}'." + ), + "error_type": "unsupported_operation", + "provider": getattr(provider, "name", configured), + }) + + try: + result = provider.edit(prompt=prompt, image=image, aspect_ratio=aspect_ratio) + except Exception as exc: + logger.warning( + "Image edit provider '%s' raised: %s", + getattr(provider, "name", "?"), exc, + ) + return json.dumps({ + "success": False, + "image": None, + "error": f"Provider '{getattr(provider, 'name', '?')}' error: {exc}", + "error_type": "provider_exception", + }) + if not isinstance(result, dict): + return json.dumps({ + "success": False, + "image": None, + "error": "Provider returned a non-dict result", + "error_type": "provider_contract", + }) + return json.dumps(result) + + +# --------------------------------------------------------------------------- +# Handler +# --------------------------------------------------------------------------- + + +def _handle_image_edit(args: Dict[str, Any], **kw: Any) -> str: + prompt = (args.get("prompt") or "").strip() + image = (args.get("image") or "").strip() + + if not prompt: + from tools.registry import tool_error + return tool_error("prompt is required for image editing", success=False) + if not image: + from tools.registry import tool_error + return tool_error("image is required for image editing", success=False) + + aspect_ratio = args.get("aspect_ratio", DEFAULT_ASPECT_RATIO) + + dispatched = _dispatch_edit(prompt, image, aspect_ratio) + if dispatched is not None: + return dispatched + + # No plugin provider available that supports editing. + return json.dumps({ + "success": False, + "image": None, + "error": ( + "No image editing provider is configured. Set " + "``image_gen.provider`` in config.yaml to a backend that supports " + "image-to-image editing (e.g. ``openai-codex``), then restart or " + "/reset." + ), + "error_type": "unsupported_operation", + }) + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- +from tools.registry import registry + + +def _edit_check_fn() -> bool: + """Gate registration: require a plugin-capable Hermes version.""" + try: + from agent.image_gen_registry import get_provider # noqa: F401 + return True + except ImportError: + return False + + +registry.register( + name="image_edit", + toolset="image_gen", + schema=IMAGE_EDIT_SCHEMA, + handler=_handle_image_edit, + check_fn=_edit_check_fn, + requires_env=[], + is_async=False, + emoji="🖌️", +) diff --git a/toolsets.py b/toolsets.py index bab7677887a82..449cd809bd3c3 100644 --- a/toolsets.py +++ b/toolsets.py @@ -124,7 +124,7 @@ "image_gen": { "description": "Creative generation tools (images)", - "tools": ["image_generate"], + "tools": ["image_generate", "image_edit"], "includes": [] },