feat: add configurable image generation tool

ohmo/gateway/bridge.py

@@ -276,6 +276,7 @@ class OhmoGatewayBridge:
                         channel=message.channel,
                         chat_id=message.chat_id,
                         content=update.text,
+                        media=list(getattr(update, "media", None) or (update.metadata or {}).get("_media") or []),
                         metadata={**inbound_meta, **(update.metadata or {})},
                     )
                 )

ohmo/gateway/runtime.py

@@ -86,6 +86,7 @@ class GatewayStreamUpdate:
     kind: str
     text: str
     metadata: dict[str, object]
+    media: list[str] | None = None
 
 
 class OhmoSessionRuntimePool:
@@ -575,6 +576,14 @@ class OhmoSessionRuntimePool:
                 bundle.session_id,
                 event.tool_name,
             )
+            media = _extract_tool_media(event)
+            if media:
+                yield GatewayStreamUpdate(
+                    kind="media",
+                    text=_format_tool_media_caption(event, media),
+                    metadata={"_session_key": session_key, "_media": media, "_tool_media": True},
+                    media=media,
+                )
             return
         if isinstance(event, ErrorEvent):
             logger.error(
@@ -773,6 +782,45 @@ def _sanitize_snapshot_messages(raw_messages: object) -> list[dict[str, object]]
     return [message.model_dump(mode="json") for message in _sanitize_group_command_prompts(messages)]
 
 
+def _extract_tool_media(event: ToolExecutionCompleted) -> list[str]:
+    """Return local media paths produced by a tool completion event."""
+    if event.is_error or not isinstance(event.metadata, dict):
+        return []
+    raw_paths = event.metadata.get("paths") or event.metadata.get("media")
+    if isinstance(raw_paths, str):
+        candidates = [raw_paths]
+    elif isinstance(raw_paths, list):
+        candidates = [str(item) for item in raw_paths if isinstance(item, str) and item.strip()]
+    else:
+        candidates = []
+    media: list[str] = []
+    seen: set[str] = set()
+    for raw in candidates:
+        path = Path(raw).expanduser()
+        if not path.is_absolute():
+            path = path.resolve()
+        if not path.is_file():
+            continue
+        resolved = str(path)
+        if resolved not in seen:
+            seen.add(resolved)
+            media.append(resolved)
+    return media
+
+
+def _format_tool_media_caption(event: ToolExecutionCompleted, media: list[str]) -> str:
+    """Return a short caption for media generated by tools."""
+    if event.tool_name == "image_generation":
+        provider = ""
+        if isinstance(event.metadata, dict):
+            provider = str(event.metadata.get("provider") or "").strip()
+        suffix = f" via {provider}" if provider else ""
+        names = ", ".join(Path(path).name for path in media)
+        return f"已生成图片{suffix}：{names}"
+    names = ", ".join(Path(path).name for path in media)
+    return f"已生成文件：{names}"
+
+
 def _sanitize_group_command_prompts(messages: list[ConversationMessage]) -> list[ConversationMessage]:
     """Replace internal /group tool-driving prompts with durable user-facing history."""
     return [_sanitize_group_command_prompt(message) for message in messages]

src/openharness/channels/impl/feishu.py

@@ -432,6 +432,25 @@ class FeishuChannel(BaseChannel):
         self._sender_cache: OrderedDict[str, _FeishuSenderInfo] = OrderedDict()
         self._loop: asyncio.AbstractEventLoop | None = None
 
+    def _ensure_rest_client(self) -> bool:
+        """Initialize the Feishu REST client without starting the WebSocket receiver."""
+        if self._client is not None:
+            return True
+        if not FEISHU_AVAILABLE:
+            logger.error("Feishu SDK not installed. Run: pip install lark-oapi")
+            return False
+        if not self.config.app_id or not self.config.app_secret:
+            logger.error("Feishu app_id and app_secret not configured")
+            return False
+        import lark_oapi as lark
+
+        self._client = lark.Client.builder() \
+            .app_id(self.config.app_id) \
+            .app_secret(self.config.app_secret) \
+            .log_level(lark.LogLevel.INFO) \
+            .build()
+        return True
+
     async def start(self) -> None:
         """Start the Feishu bot with WebSocket long connection."""
         if not FEISHU_AVAILABLE:
@@ -446,12 +465,8 @@ class FeishuChannel(BaseChannel):
         self._running = True
         self._loop = asyncio.get_running_loop()
 
-        # Create Lark client for sending messages
-        self._client = lark.Client.builder() \
-            .app_id(self.config.app_id) \
-            .app_secret(self.config.app_secret) \
-            .log_level(lark.LogLevel.INFO) \
-            .build()
+        if not self._ensure_rest_client():
+            return
 
         # Create event handler (only register message receive, ignore other events)
         event_handler = lark.EventDispatcherHandler.builder(
@@ -1046,7 +1061,7 @@ class FeishuChannel(BaseChannel):
 
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Feishu, including media (images/files) if present."""
-        if not self._client:
+        if not self._ensure_rest_client():
             logger.warning("Feishu client not initialized")
             return
 
@@ -1060,19 +1075,28 @@ class FeishuChannel(BaseChannel):
                 else None
             )
 
+            failed_media: list[str] = []
+            sent_media: list[str] = []
             for file_path in msg.media:
                 if not os.path.isfile(file_path):
                     logger.warning("Media file not found: %s", file_path)
+                    failed_media.append(file_path)
                     continue
                 ext = os.path.splitext(file_path)[1].lower()
                 if ext in self._IMAGE_EXTS:
                     key = await loop.run_in_executor(None, self._upload_image_sync, file_path)
                     if key:
-                        await loop.run_in_executor(
+                        ok = await loop.run_in_executor(
                             None, self._send_message_sync,
                             receive_id_type, msg.chat_id, "image", json.dumps({"image_key": key}, ensure_ascii=False),
                             reply_mid,
                         )
+                        if ok:
+                            sent_media.append(file_path)
+                        else:
+                            failed_media.append(file_path)
+                    else:
+                        failed_media.append(file_path)
                 else:
                     key = await loop.run_in_executor(None, self._upload_file_sync, file_path)
                     if key:
@@ -1082,11 +1106,26 @@ class FeishuChannel(BaseChannel):
                             media_type = "media"
                         else:
                             media_type = "file"
-                        await loop.run_in_executor(
+                        ok = await loop.run_in_executor(
                             None, self._send_message_sync,
                             receive_id_type, msg.chat_id, media_type, json.dumps({"file_key": key}, ensure_ascii=False),
                             reply_mid,
                         )
+                        if ok:
+                            sent_media.append(file_path)
+                        else:
+                            failed_media.append(file_path)
+                    else:
+                        failed_media.append(file_path)
+
+            if failed_media:
+                names = ", ".join(os.path.basename(path) for path in failed_media)
+                failure_body = json.dumps({"text": f"文件发送失败：{names}"}, ensure_ascii=False)
+                await loop.run_in_executor(
+                    None, self._send_message_sync,
+                    receive_id_type, msg.chat_id, "text", failure_body,
+                    reply_mid,
+                )
 
             if msg.content and msg.content.strip():
                 fmt = self._detect_msg_format(msg.content)

src/openharness/config/settings.py

@@ -467,6 +467,37 @@ def _profile_from_flat_settings(settings: "Settings") -> tuple[str, ProviderProf
     return name, profile
 
 
+class ImageGenerationConfig(BaseModel):
+    """Configuration for the image_generation tool."""
+
+    provider: str = "auto"
+    model: str = "gpt-image-2"
+    api_key: str = ""
+    base_url: str = ""
+    codex_model: str = "gpt-5.4"
+    codex_base_url: str = ""
+
+    @classmethod
+    def from_env(cls) -> "ImageGenerationConfig":
+        """Load image generation config from environment variables."""
+        return cls(
+            provider=os.environ.get("OPENHARNESS_IMAGE_GENERATION_PROVIDER", "auto").strip()
+            or "auto",
+            model=os.environ.get("OPENHARNESS_IMAGE_GENERATION_MODEL", "gpt-image-2").strip()
+            or "gpt-image-2",
+            api_key=os.environ.get("OPENHARNESS_IMAGE_GENERATION_API_KEY", "").strip(),
+            base_url=os.environ.get("OPENHARNESS_IMAGE_GENERATION_BASE_URL", "").strip(),
+            codex_model=os.environ.get("OPENHARNESS_IMAGE_GENERATION_CODEX_MODEL", "gpt-5.4").strip()
+            or "gpt-5.4",
+            codex_base_url=os.environ.get("OPENHARNESS_IMAGE_GENERATION_CODEX_BASE_URL", "").strip(),
+        )
+
+    @property
+    def is_configured(self) -> bool:
+        """Return True when either a key provider or Codex provider is selected."""
+        return bool(self.api_key or self.provider in {"auto", "codex"})
+
+
 class VisionModelConfig(BaseModel):
     """Configuration for the vision model used by the image_to_text tool.
 
@@ -537,6 +568,9 @@ class Settings(BaseModel):
     # Vision model (image-to-text fallback)
     vision: VisionModelConfig = Field(default_factory=VisionModelConfig)
 
+    # Image generation model
+    image_generation: ImageGenerationConfig = Field(default_factory=ImageGenerationConfig)
+
     def merged_profiles(self) -> dict[str, ProviderProfile]:
         """Return the saved profiles merged over the built-in catalog."""
         merged = default_provider_profiles()

src/openharness/engine/messages.py

@@ -53,6 +53,7 @@ class ToolResultBlock(BaseModel):
     tool_use_id: str
     content: str
     is_error: bool = False
+    result_metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 ContentBlock = Annotated[

src/openharness/engine/query.py

@@ -825,6 +825,7 @@ async def run_query(
                 tool_name=tc.name,
                 output=result.content,
                 is_error=result.is_error,
+                metadata=result.result_metadata,
             ), None
             tool_results = [result]
         else:
@@ -863,6 +864,7 @@ async def run_query(
                     tool_name=tc.name,
                     output=result.content,
                     is_error=result.is_error,
+                    metadata=result.result_metadata,
                 ), None
 
         messages.append(ConversationMessage(role="user", content=tool_results))
@@ -981,6 +983,7 @@ async def _execute_tool_call(
         tool_use_id=tool_use_id,
         content=inline_output,
         is_error=result.is_error,
+        result_metadata=dict(result.metadata or {}),
     )
     _record_tool_carryover(
         context,

src/openharness/engine/stream_events.py

@@ -39,6 +39,7 @@ class ToolExecutionCompleted:
     tool_name: str
     output: str
     is_error: bool = False
+    metadata: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True)

src/openharness/tools/__init__.py

@@ -19,6 +19,7 @@ from openharness.tools.file_read_tool import FileReadTool
 from openharness.tools.file_write_tool import FileWriteTool
 from openharness.tools.glob_tool import GlobTool
 from openharness.tools.grep_tool import GrepTool
+from openharness.tools.image_generation_tool import ImageGenerationTool
 from openharness.tools.image_to_text_tool import ImageToTextTool
 from openharness.tools.list_mcp_resources_tool import ListMcpResourcesTool
 from openharness.tools.lsp_tool import LspTool
@@ -59,6 +60,7 @@ def create_default_tool_registry(mcp_manager=None) -> ToolRegistry:
         GlobTool(),
         GrepTool(),
         ImageToTextTool(),
+        ImageGenerationTool(),
         SkillTool(),
         ToolSearchTool(),
         WebFetchTool(),

src/openharness/tools/image_generation_tool.py

@@ -0,0 +1,356 @@
+"""Generate or edit raster images with configurable image generation providers."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from pathlib import Path
+from typing import Any, Literal
+
+import httpx
+from openai import AsyncOpenAI
+from pydantic import BaseModel, Field
+
+from openharness.api.codex_client import _build_codex_headers, _resolve_codex_url
+from openharness.api.openai_client import _normalize_openai_base_url
+from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult
+
+log = logging.getLogger(__name__)
+
+_DEFAULT_PROMPT = (
+    "Create a high-quality raster image that satisfies the user's request. "
+    "Avoid watermarks, unintended text, and unrelated logos."
+)
+_DEFAULT_MODEL = "gpt-image-2"
+_DEFAULT_OUTPUT_DIR = "generated_images"
+ImageGenerationProvider = Literal["auto", "openai", "codex"]
+
+
+class ImageGenerationToolInput(BaseModel):
+    """Arguments for image generation or editing."""
+
+    prompt: str = Field(default=_DEFAULT_PROMPT, description="Image generation or edit prompt.")
+    provider: ImageGenerationProvider = Field(
+        default="auto",
+        description="Image generation provider: auto, openai, or codex.",
+    )
+    image_paths: list[str] = Field(
+        default_factory=list,
+        description="Local image paths to edit or use as references. OpenAI provider uses image edit mode. Codex hosted generation currently treats these as visual context.",
+    )
+    mask_path: str | None = Field(default=None, description="Optional PNG mask path for OpenAI edit mode.")
+    output_path: str | None = Field(
+        default=None,
+        description="Optional output path. For multiple images, numeric suffixes are added.",
+    )
+    output_dir: str = Field(
+        default=_DEFAULT_OUTPUT_DIR,
+        description="Output directory used when output_path is not provided.",
+    )
+    model: str | None = Field(default=None, description="OpenAI image model override.")
+    n: int = Field(default=1, ge=1, le=10, description="Number of images to generate.")
+    size: str = Field(default="auto", description="OpenAI image size, e.g. auto, 1024x1024, 1536x1024.")
+    quality: str = Field(default="medium", description="OpenAI image quality, e.g. low, medium, high, auto.")
+    background: Literal["transparent", "opaque", "auto"] | None = Field(
+        default=None,
+        description="Optional OpenAI background mode when supported by the provider.",
+    )
+    output_format: Literal["png", "jpeg", "webp"] = Field(default="png", description="Output image format.")
+    output_compression: int | None = Field(
+        default=None,
+        ge=0,
+        le=100,
+        description="Optional OpenAI compression level for lossy output formats when supported.",
+    )
+    input_fidelity: Literal["low", "high"] | None = Field(
+        default=None,
+        description="Optional OpenAI edit input fidelity when supported by the provider.",
+    )
+    moderation: str | None = Field(default=None, description="Optional OpenAI moderation setting.")
+    overwrite: bool = Field(default=False, description="Whether to overwrite existing output files.")
+
+
+class ImageGenerationTool(BaseTool):
+    """Generate or edit raster images and save them to local files."""
+
+    name = "image_generation"
+    description = (
+        "Generate or edit raster images using a configurable image generation provider. "
+        "Use this for bitmap assets such as photos, illustrations, sprites, mockups, "
+        "transparent cutouts, or edited local images. Supports provider='codex' for "
+        "Codex hosted image_generation with Codex subscription auth, and provider='openai' "
+        "for OpenAI-compatible key/base_url image APIs."
+    )
+    input_model = ImageGenerationToolInput
+
+    async def execute(self, arguments: ImageGenerationToolInput, context: ToolExecutionContext) -> ToolResult:
+        config = context.metadata.get("image_generation_config", {})
+        if not isinstance(config, dict):
+            config = {}
+        provider = _resolve_provider(arguments.provider, config)
+
+        try:
+            output_paths = self._resolve_output_paths(arguments, context.cwd)
+            if provider == "codex":
+                image_b64, revised_prompt = await self._generate_with_codex(arguments, config)
+                written = self._write_images(image_b64, output_paths, overwrite=arguments.overwrite)
+                extra = f"\nRevised prompt: {revised_prompt}" if revised_prompt else ""
+                return ToolResult(
+                    output=(
+                        "[Image generation via Codex hosted image_generation]\n"
+                        + "\n".join(f"Wrote {path}" for path in written)
+                        + extra
+                    ),
+                    metadata={
+                        "paths": [str(path) for path in written],
+                        "provider": "codex",
+                        "revised_prompt": revised_prompt,
+                    },
+                )
+
+            image_b64 = await self._generate_with_openai(arguments, config)
+            written = self._write_images(image_b64, output_paths, overwrite=arguments.overwrite)
+        except Exception as exc:
+            log.exception("image_generation failed")
+            return ToolResult(output=f"image_generation failed: {exc}", is_error=True)
+
+        mode = "edit" if arguments.image_paths else "generate"
+        model = (arguments.model or str(config.get("model") or _DEFAULT_MODEL)).strip()
+        return ToolResult(
+            output=(
+                f"[Image generation via {model} ({mode}, openai)]\n"
+                + "\n".join(f"Wrote {path}" for path in written)
+            ),
+            metadata={"paths": [str(path) for path in written], "model": model, "mode": mode, "provider": "openai"},
+        )
+
+    async def _generate_with_openai(self, arguments: ImageGenerationToolInput, config: dict[str, object]) -> list[str]:
+        model = (arguments.model or str(config.get("model") or _DEFAULT_MODEL)).strip()
+        api_key = str(config.get("api_key") or "").strip()
+        base_url = str(config.get("base_url") or "").strip()
+        if not api_key:
+            raise RuntimeError(
+                "OpenAI image generation API key is not configured. Set image_generation.api_key "
+                "or OPENHARNESS_IMAGE_GENERATION_API_KEY, or choose provider='codex'."
+            )
+        if arguments.image_paths:
+            return await self._edit_images(arguments, model, api_key, base_url)
+        return await self._generate_images(arguments, model, api_key, base_url)
+
+    async def _generate_with_codex(
+        self,
+        arguments: ImageGenerationToolInput,
+        config: dict[str, object],
+    ) -> tuple[list[str], str | None]:
+        auth_token = str(config.get("codex_auth_token") or "").strip()
+        if not auth_token:
+            raise RuntimeError(
+                "Codex image generation auth is not configured. Run 'oh auth codex-login' "
+                "or use provider='openai' with OPENHARNESS_IMAGE_GENERATION_API_KEY."
+            )
+        model = str(config.get("codex_model") or "gpt-5.4").strip() or "gpt-5.4"
+        base_url = str(config.get("codex_base_url") or "").strip()
+        prompt = _codex_prompt(arguments)
+        body: dict[str, Any] = {
+            "model": model,
+            "store": False,
+            "stream": True,
+            "instructions": "Generate the requested image using the hosted image_generation tool.",
+            "input": [{"role": "user", "content": _codex_user_content(arguments, prompt)}],
+            "text": {"verbosity": "medium"},
+            "tools": [{"type": "image_generation", "output_format": arguments.output_format}],
+            "tool_choice": "auto",
+            "parallel_tool_calls": False,
+        }
+        headers = _build_codex_headers(auth_token)
+        url = _resolve_codex_url(base_url)
+
+        image_results: list[str] = []
+        revised_prompt: str | None = None
+        async with httpx.AsyncClient(timeout=180.0, follow_redirects=True) as client:
+            async with client.stream("POST", url, headers=headers, json=body) as response:
+                if response.status_code >= 400:
+                    payload = await response.aread()
+                    raise RuntimeError(payload.decode("utf-8", "replace") or f"Codex request failed: {response.status_code}")
+                async for event in _iter_sse_events(response):
+                    if event.get("type") != "response.output_item.done":
+                        if event.get("type") == "response.failed":
+                            raise RuntimeError(json.dumps(event.get("response") or event, ensure_ascii=False))
+                        if event.get("type") == "error":
+                            raise RuntimeError(json.dumps(event, ensure_ascii=False))
+                        continue
+                    item = event.get("item")
+                    if not isinstance(item, dict) or item.get("type") != "image_generation_call":
+                        continue
+                    result = item.get("result")
+                    if isinstance(result, str) and result:
+                        image_results.append(result)
+                    candidate = item.get("revised_prompt")
+                    if isinstance(candidate, str) and candidate:
+                        revised_prompt = candidate
+        if not image_results:
+            raise RuntimeError("Codex hosted image_generation returned no image result")
+        return image_results, revised_prompt
+
+    @staticmethod
+    async def _generate_images(arguments: ImageGenerationToolInput, model: str, api_key: str, base_url: str) -> list[str]:
+        client = AsyncOpenAI(
+            api_key=api_key,
+            base_url=_normalize_openai_base_url(base_url),
+            default_headers={"Authorization": f"Bearer {api_key}"},
+        )
+        result = await client.images.generate(**_image_payload(arguments, model))
+        return _extract_b64_images(result)
+
+    @staticmethod
+    async def _edit_images(arguments: ImageGenerationToolInput, model: str, api_key: str, base_url: str) -> list[str]:
+        client = AsyncOpenAI(
+            api_key=api_key,
+            base_url=_normalize_openai_base_url(base_url),
+            default_headers={"Authorization": f"Bearer {api_key}"},
+        )
+        image_handles = [Path(path).expanduser().resolve().open("rb") for path in arguments.image_paths]
+        mask_handle = Path(arguments.mask_path).expanduser().resolve().open("rb") if arguments.mask_path else None
+        try:
+            payload = _image_payload(arguments, model)
+            payload["image"] = image_handles if len(image_handles) > 1 else image_handles[0]
+            if mask_handle is not None:
+                payload["mask"] = mask_handle
+            result = await client.images.edit(**payload)
+        finally:
+            for handle in image_handles:
+                handle.close()
+            if mask_handle is not None:
+                mask_handle.close()
+        return _extract_b64_images(result)
+
+    @staticmethod
+    def _resolve_output_paths(arguments: ImageGenerationToolInput, cwd: Path) -> list[Path]:
+        suffix = f".{arguments.output_format}"
+        if arguments.output_path:
+            base = Path(arguments.output_path)
+            if not base.is_absolute():
+                base = cwd / base
+            base = base.expanduser().resolve()
+        else:
+            out_dir = Path(arguments.output_dir)
+            if not out_dir.is_absolute():
+                out_dir = cwd / out_dir
+            out_dir = out_dir.expanduser().resolve()
+            base = out_dir / f"image{suffix}"
+        if base.suffix.lower() != suffix:
+            base = base.with_suffix(suffix)
+        if arguments.n == 1:
+            return [base]
+        return [base.with_name(f"{base.stem}-{idx}{base.suffix}") for idx in range(1, arguments.n + 1)]
+
+    @staticmethod
+    def _write_images(images: list[str], output_paths: list[Path], *, overwrite: bool) -> list[Path]:
+        written: list[Path] = []
+        for image_b64, output_path in zip(images, output_paths, strict=False):
+            if output_path.exists() and not overwrite:
+                raise FileExistsError(f"output already exists: {output_path} (set overwrite=true)")
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            output_path.write_bytes(base64.b64decode(image_b64))
+            written.append(output_path)
+        if not written:
+            raise RuntimeError("provider returned no image data")
+        return written
+
+
+def _resolve_provider(requested: str, config: dict[str, object]) -> Literal["openai", "codex"]:
+    if requested in {"openai", "codex"}:
+        return requested  # type: ignore[return-value]
+    configured = str(config.get("provider") or "auto").strip().lower()
+    if configured in {"openai", "codex"}:
+        return configured  # type: ignore[return-value]
+    if str(config.get("codex_auth_token") or "").strip():
+        return "codex"
+    return "openai"
+
+
+def _image_payload(arguments: ImageGenerationToolInput, model: str) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "model": model,
+        "prompt": arguments.prompt,
+        "n": arguments.n,
+        "size": arguments.size,
+        "quality": arguments.quality,
+        "background": arguments.background,
+        "output_format": arguments.output_format,
+        "output_compression": arguments.output_compression,
+        "input_fidelity": arguments.input_fidelity,
+        "moderation": arguments.moderation,
+    }
+    return {key: value for key, value in payload.items() if value is not None}
+
+
+def _codex_prompt(arguments: ImageGenerationToolInput) -> str:
+    lines = [arguments.prompt]
+    if arguments.n > 1:
+        lines.append(f"Generate {arguments.n} distinct variants.")
+    if arguments.image_paths:
+        lines.append("Use the attached image(s) as visual context/reference for the generation or edit.")
+    return "\n".join(line for line in lines if line.strip())
+
+
+def _codex_user_content(arguments: ImageGenerationToolInput, prompt: str) -> list[dict[str, str]]:
+    content = [{"type": "input_text", "text": prompt}]
+    for path_str in arguments.image_paths:
+        path = Path(path_str).expanduser().resolve()
+        media_type = _media_type_for_path(path)
+        data = base64.b64encode(path.read_bytes()).decode("ascii")
+        content.append({"type": "input_image", "image_url": f"data:{media_type};base64,{data}"})
+    return content
+
+
+def _media_type_for_path(path: Path) -> str:
+    return {
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".webp": "image/webp",
+        ".gif": "image/gif",
+    }.get(path.suffix.lower(), "image/png")
+
+
+async def _iter_sse_events(response: httpx.Response):
+    data_lines: list[str] = []
+    async for line in response.aiter_lines():
+        if line == "":
+            if data_lines:
+                payload = "\n".join(data_lines).strip()
+                data_lines = []
+                if payload and payload != "[DONE]":
+                    try:
+                        event = json.loads(payload)
+                    except json.JSONDecodeError:
+                        continue
+                    if isinstance(event, dict):
+                        yield event
+            continue
+        if line.startswith("data:"):
+            data_lines.append(line[5:].strip())
+    if data_lines:
+        payload = "\n".join(data_lines).strip()
+        if payload and payload != "[DONE]":
+            try:
+                event = json.loads(payload)
+            except json.JSONDecodeError:
+                return
+            if isinstance(event, dict):
+                yield event
+
+
+def _extract_b64_images(result: Any) -> list[str]:
+    images: list[str] = []
+    for item in getattr(result, "data", []) or []:
+        b64 = getattr(item, "b64_json", None)
+        if isinstance(b64, str) and b64:
+            images.append(b64)
+            continue
+        url = getattr(item, "url", None)
+        if isinstance(url, str) and url.startswith("data:image/") and ";base64," in url:
+            images.append(url.split(";base64,", 1)[1])
+    return images

src/openharness/ui/runtime.py

@@ -50,6 +50,45 @@ StreamRenderer = Callable[[StreamEvent], Awaitable[None]]
 ClearHandler = Callable[[], Awaitable[None]]
 
 
+def _resolve_image_generation_config(settings) -> dict[str, str]:
+    """Resolve image generation configuration from settings, environment, and Codex auth."""
+    from openharness.config.settings import ImageGenerationConfig, ProviderProfile
+
+    cfg = settings.image_generation
+    env_cfg = ImageGenerationConfig.from_env()
+    resolved = {
+        "provider": cfg.provider or env_cfg.provider,
+        "model": cfg.model or env_cfg.model,
+        "api_key": cfg.api_key or env_cfg.api_key,
+        "base_url": cfg.base_url or env_cfg.base_url,
+        "codex_model": cfg.codex_model or env_cfg.codex_model,
+        "codex_base_url": cfg.codex_base_url or env_cfg.codex_base_url,
+    }
+
+    try:
+        codex_profile = settings.merged_profiles().get("codex") or ProviderProfile(
+            label="Codex Subscription",
+            provider="openai_codex",
+            api_format="openai",
+            auth_source="codex_subscription",
+            default_model="gpt-5.4",
+        )
+        codex_settings = settings.model_copy(
+            update={
+                "active_profile": "codex",
+                "profiles": {**settings.profiles, "codex": codex_profile},
+            }
+        ).materialize_active_profile()
+        codex_auth = codex_settings.resolve_auth()
+        resolved["codex_auth_token"] = codex_auth.value
+        resolved["codex_base_url"] = resolved["codex_base_url"] or (codex_settings.base_url or "")
+        resolved["codex_model"] = resolved["codex_model"] or codex_settings.model
+    except Exception:
+        pass
+
+    return resolved
+
+
 def _resolve_vision_config(settings) -> dict[str, str]:
     """Resolve the vision model configuration from settings or environment.
 
@@ -348,6 +387,7 @@ async def build_runtime(
             "extra_plugin_roots": normalized_plugin_roots,
             "session_id": session_id,
             "vision_model_config": _resolve_vision_config(settings),
+            "image_generation_config": _resolve_image_generation_config(settings),
             **restored_metadata,
         },
     )

tests/test_engine/test_query_engine.py

@@ -1182,7 +1182,7 @@ class _OkTool(BaseTool):
 
     async def execute(self, arguments: BaseModel, context: ToolExecutionContext) -> ToolResult:
         del arguments, context
-        return ToolResult(output="ok")
+        return ToolResult(output="ok", metadata={"sentinel": "metadata"})
 
 
 class _BoomTool(BaseTool):
@@ -1340,6 +1340,7 @@ async def test_query_engine_synthesizes_tool_result_when_parallel_tool_raises(tm
     assert set(completed_by_name) == {"ok_tool", "boom_tool"}
     assert completed_by_name["ok_tool"].is_error is False
     assert completed_by_name["ok_tool"].output == "ok"
+    assert completed_by_name["ok_tool"].metadata == {"sentinel": "metadata"}
     assert completed_by_name["boom_tool"].is_error is True
     assert "RuntimeError" in completed_by_name["boom_tool"].output
     assert "boom" in completed_by_name["boom_tool"].output

tests/test_ohmo/test_gateway.py

@@ -404,6 +404,55 @@ async def test_runtime_pool_stream_message_emits_progress_and_tool_hint(tmp_path
     assert updates[-1].text == "done"
 
 
+@pytest.mark.asyncio
+@pytest.mark.asyncio
+async def test_runtime_pool_stream_message_emits_media_for_generated_tool_paths(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    image_path = tmp_path / "generated.png"
+    image_path.write_bytes(b"png")
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+            async def submit_message(self, content):
+                yield ToolExecutionCompleted(
+                    tool_name="image_generation",
+                    output=f"Wrote {image_path}",
+                    metadata={"paths": [str(image_path)], "provider": "codex"},
+                )
+                yield AssistantTextDelta(text="done")
+
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            cwd=str(tmp_path),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: None),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="draw")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    media_updates = [u for u in updates if u.kind == "media"]
+    assert len(media_updates) == 1
+    assert media_updates[0].media == [str(image_path)]
+    assert media_updates[0].metadata["_media"] == [str(image_path)]
+    assert "已生成图片 via codex" in media_updates[0].text
+
+
 @pytest.mark.asyncio
 async def test_runtime_pool_stream_message_formats_auto_compact_status_for_feishu(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"
@@ -1091,6 +1140,36 @@ def test_runtime_pool_includes_group_speaker_context():
     assert "请帮我看一下" in text
 
 
+@pytest.mark.asyncio
+async def test_gateway_bridge_publishes_media_updates():
+    bus = MessageBus()
+
+    class FakeRuntimePool:
+        async def stream_message(self, message, session_key):
+            yield SimpleNamespace(
+                kind="media",
+                text="已生成图片：generated.png",
+                media=["/tmp/generated.png"],
+                metadata={"_session_key": session_key, "_media": ["/tmp/generated.png"]},
+            )
+
+    bridge = OhmoGatewayBridge(bus=bus, runtime_pool=FakeRuntimePool())
+    task = asyncio.create_task(bridge.run())
+    try:
+        await bus.publish_inbound(InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="draw"))
+        outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+    finally:
+        bridge.stop()
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    assert outbound.content == "已生成图片：generated.png"
+    assert outbound.media == ["/tmp/generated.png"]
+
+
 @pytest.mark.asyncio
 async def test_gateway_bridge_publishes_progress_updates():
     bus = MessageBus()

tests/test_tools/test_image_generation_tool.py

@@ -0,0 +1,215 @@
+"""Tests for the image_generation tool."""
+
+from __future__ import annotations
+
+import base64
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from openharness.config.settings import ImageGenerationConfig
+from openharness.tools.base import ToolExecutionContext
+from openharness.tools.image_generation_tool import ImageGenerationTool, ImageGenerationToolInput
+
+
+class _FakeStreamResponse:
+    def __init__(self, *, status_code: int = 200, lines: list[str] | None = None, body: str = "") -> None:
+        self.status_code = status_code
+        self._lines = lines or []
+        self._body = body.encode("utf-8")
+
+    async def __aenter__(self) -> "_FakeStreamResponse":
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        return None
+
+    async def aread(self) -> bytes:
+        return self._body
+
+    async def aiter_lines(self):
+        for line in self._lines:
+            yield line
+
+
+class _FakeAsyncClient:
+    def __init__(self, response: _FakeStreamResponse, sink: dict[str, Any]) -> None:
+        self._response = response
+        self._sink = sink
+
+    async def __aenter__(self) -> "_FakeAsyncClient":
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        return None
+
+    def stream(self, method: str, url: str, *, headers: dict[str, str], json: dict[str, Any]):
+        self._sink["method"] = method
+        self._sink["url"] = url
+        self._sink["headers"] = headers
+        self._sink["json"] = json
+        return self._response
+
+
+def _b64url(data: dict[str, object]) -> str:
+    raw = json.dumps(data, separators=(",", ":")).encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _fake_codex_token() -> str:
+    payload = {"https://api.openai.com/auth": {"chatgpt_account_id": "acct_test"}}
+    return f"{_b64url({'alg': 'none', 'typ': 'JWT'})}.{_b64url(payload)}.sig"
+
+
+@pytest.mark.asyncio
+async def test_execute_requires_api_key_for_openai(tmp_path: Path) -> None:
+    tool = ImageGenerationTool()
+    result = await tool.execute(
+        ImageGenerationToolInput(prompt="a cat", provider="openai"),
+        ToolExecutionContext(cwd=tmp_path, metadata={"image_generation_config": {}}),
+    )
+    assert result.is_error
+    assert "API key is not configured" in result.output
+
+
+@pytest.mark.asyncio
+async def test_execute_generate_writes_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    image_bytes = b"fake-png"
+    image_b64 = base64.b64encode(image_bytes).decode("ascii")
+
+    async def fake_generate_images(arguments, model, api_key, base_url):
+        assert arguments.prompt == "a cat"
+        assert model == "gpt-image-2"
+        assert api_key == "test-key"
+        assert base_url == ""
+        return [image_b64]
+
+    monkeypatch.setattr(ImageGenerationTool, "_generate_images", staticmethod(fake_generate_images))
+
+    tool = ImageGenerationTool()
+    result = await tool.execute(
+        ImageGenerationToolInput(prompt="a cat", output_path="assets/cat.png", provider="openai"),
+        ToolExecutionContext(
+            cwd=tmp_path,
+            metadata={"image_generation_config": {"api_key": "test-key", "model": "gpt-image-2"}},
+        ),
+    )
+
+    out = tmp_path / "assets" / "cat.png"
+    assert not result.is_error
+    assert out.read_bytes() == image_bytes
+    assert result.metadata["paths"] == [str(out)]
+    assert result.metadata["provider"] == "openai"
+
+
+@pytest.mark.asyncio
+async def test_execute_codex_hosted_generation(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    image_bytes = b"codex-png"
+    image_b64 = base64.b64encode(image_bytes).decode("ascii")
+    sink: dict[str, Any] = {}
+    response = _FakeStreamResponse(
+        lines=[
+            'data: {"type":"response.output_item.done","item":{"id":"ig_1","type":"image_generation_call","status":"completed","revised_prompt":"blue icon","result":"%s"}}' % image_b64,
+            "",
+            'data: {"type":"response.completed","response":{"status":"completed"}}',
+            "",
+        ]
+    )
+    monkeypatch.setattr(
+        "openharness.tools.image_generation_tool.httpx.AsyncClient",
+        lambda *args, **kwargs: _FakeAsyncClient(response, sink),
+    )
+
+    tool = ImageGenerationTool()
+    result = await tool.execute(
+        ImageGenerationToolInput(prompt="a blue icon", output_path="icon.png", provider="codex"),
+        ToolExecutionContext(
+            cwd=tmp_path,
+            metadata={
+                "image_generation_config": {
+                    "codex_auth_token": _fake_codex_token(),
+                    "codex_model": "gpt-5.4",
+                }
+            },
+        ),
+    )
+
+    out = tmp_path / "icon.png"
+    assert not result.is_error
+    assert out.read_bytes() == image_bytes
+    assert result.metadata["provider"] == "codex"
+    assert result.metadata["revised_prompt"] == "blue icon"
+    assert sink["url"].endswith("/codex/responses")
+    assert sink["json"]["tools"] == [{"type": "image_generation", "output_format": "png"}]
+    assert sink["json"]["tool_choice"] == "auto"
+
+
+@pytest.mark.asyncio
+async def test_auto_provider_prefers_codex_when_token_available(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    image_b64 = base64.b64encode(b"codex").decode("ascii")
+
+    async def fake_codex(self, arguments, config):
+        return [image_b64], None
+
+    monkeypatch.setattr(ImageGenerationTool, "_generate_with_codex", fake_codex)
+
+    tool = ImageGenerationTool()
+    result = await tool.execute(
+        ImageGenerationToolInput(prompt="a cat", output_path="cat.png"),
+        ToolExecutionContext(
+            cwd=tmp_path,
+            metadata={"image_generation_config": {"codex_auth_token": _fake_codex_token()}},
+        ),
+    )
+
+    assert not result.is_error
+    assert result.metadata["provider"] == "codex"
+
+
+@pytest.mark.asyncio
+async def test_execute_refuses_overwrite_by_default(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    out = tmp_path / "image.png"
+    out.write_bytes(b"existing")
+    image_b64 = base64.b64encode(b"new").decode("ascii")
+
+    async def fake_generate_images(arguments, model, api_key, base_url):
+        return [image_b64]
+
+    monkeypatch.setattr(ImageGenerationTool, "_generate_images", staticmethod(fake_generate_images))
+
+    tool = ImageGenerationTool()
+    result = await tool.execute(
+        ImageGenerationToolInput(prompt="a cat", output_path=str(out), provider="openai"),
+        ToolExecutionContext(cwd=tmp_path, metadata={"image_generation_config": {"api_key": "test"}}),
+    )
+
+    assert result.is_error
+    assert "output already exists" in result.output
+    assert out.read_bytes() == b"existing"
+
+
+def test_resolve_output_paths_multiple(tmp_path: Path) -> None:
+    paths = ImageGenerationTool._resolve_output_paths(
+        ImageGenerationToolInput(output_path="hero.png", n=2),
+        tmp_path,
+    )
+    assert paths == [tmp_path / "hero-1.png", tmp_path / "hero-2.png"]
+
+
+def test_image_generation_config_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_IMAGE_GENERATION_PROVIDER", "openai")
+    monkeypatch.setenv("OPENHARNESS_IMAGE_GENERATION_MODEL", "gpt-image-1")
+    monkeypatch.setenv("OPENHARNESS_IMAGE_GENERATION_API_KEY", "sk-test")
+    monkeypatch.setenv("OPENHARNESS_IMAGE_GENERATION_BASE_URL", "https://example.test/v1")
+    monkeypatch.setenv("OPENHARNESS_IMAGE_GENERATION_CODEX_MODEL", "gpt-5.4")
+
+    cfg = ImageGenerationConfig.from_env()
+
+    assert cfg.provider == "openai"
+    assert cfg.model == "gpt-image-1"
+    assert cfg.api_key == "sk-test"
+    assert cfg.base_url == "https://example.test/v1"
+    assert cfg.codex_model == "gpt-5.4"
+    assert cfg.is_configured