#!/usr/bin/env python3
"""Local MCP server for Codex-orchestrated, profile-based image work."""

from __future__ import annotations

import base64
import datetime as dt
import glob
import json
import mimetypes
import os
import re
import socket
import subprocess
import sys
import tempfile
import time
import traceback
import urllib.error
import urllib.parse
import urllib.request
import uuid
import webbrowser
from pathlib import Path
from typing import Any

from yoho_credentials import CredentialError, load_api_key


PROFILE_PATH = Path.home() / '.codex' / 'yohohoho-image' / 'legacy-profiles-disabled.json'
CONFIG_API_ROOT = os.environ.get('YOHO_IMAGE_CONFIG_API', 'https://yohohoho.asia/api/v1').rstrip('/')
CONFIG_WEB_URL = 'https://yohohoho.asia/'
YOHO_GATEWAY_ORIGIN = os.environ.get('YOHO_IMAGE_GATEWAY_ORIGIN', 'https://yohohoho.asia').rstrip('/')
PUBLIC_PROFILE_ID = 'yohohoho'
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp'}
MAX_REFERENCES = 8
MAX_BATCH = 4
MAX_TIMEOUT_SECONDS = 600
MIN_PIXELS = 655_360
MAX_PIXELS = 8_294_400
MAX_EDGE = 3840
PROFILE_ID_PATTERN = r'[a-z0-9][a-z0-9_-]{0,63}'
# The protocol is a user-owned service setting.  Keep routing deterministic:
# the request path must come from the saved profile, never from model-name heuristics.
PROFILE_PROTOCOLS = {'openai_images', 'chat_completions', 'gemini_native'}
IMAGE_PLATFORMS = {
    'openai': 'OpenAI',
    'gemini_nanobanana': 'Gemini-Nanobanana',
    'gemini': 'Gemini',
    'grok': 'Grok',
    'other': '其他',
}
PLATFORM_PROTOCOLS = {
    'openai': 'openai_images',
    'gemini_nanobanana': 'gemini_native',
    'gemini': 'chat_completions',
    'grok': 'openai_images',
    'other': 'openai_images',
}
PLATFORM_MODEL_PATTERNS = {
    'openai': ('gpt-image', 'dall-e', 'openai-image'),
    'gemini_nanobanana': ('nano-banana', 'nanobanana'),
    'gemini': ('gemini', 'imagen'),
    'grok': ('grok',),
}
CONFIG_UI_SCRIPT = Path(__file__).resolve().parent / 'gpt_image2_config_ui.py'
CONFIG_UI_LAUNCHER = CONFIG_UI_SCRIPT.parent.parent / 'open-config-ui.cmd'
CONFIG_UI_HOST = '127.0.0.1'
CONFIG_UI_PORT = 8765
CONFIG_UI_PORT_RANGE = range(8765, 8776)
WIRE_MODE = 'line'
UI_SOURCE = 'config_ui'


class ToolError(Exception):
    pass


def send(payload: dict[str, Any]) -> None:
    body = json.dumps(payload, ensure_ascii=False, separators=(',', ':')).encode('utf-8')
    if WIRE_MODE == 'framed':
        sys.stdout.buffer.write(f'Content-Length: {len(body)}\r\n\r\n'.encode('ascii'))
        sys.stdout.buffer.write(body)
    else:
        # MCP's stdio transport is normally newline-delimited JSON.
        sys.stdout.buffer.write(body + b'\n')
    sys.stdout.buffer.flush()


def receive() -> dict[str, Any] | None:
    global WIRE_MODE
    while True:
        line = sys.stdin.buffer.readline()
        if not line:
            return None
        raw = line.strip()
        if not raw:
            continue
        if raw.lower().startswith(b'content-length:'):
            try:
                length = int(raw.split(b':', 1)[1].strip())
            except (IndexError, ValueError) as exc:
                raise ToolError('Invalid Content-Length header') from exc
            while True:
                header = sys.stdin.buffer.readline()
                if not header:
                    return None
                if header in {b'\r\n', b'\n'}:
                    break
            body = sys.stdin.buffer.read(length)
            if len(body) != length:
                raise ToolError('MCP stdio frame ended before Content-Length bytes were read')
            WIRE_MODE = 'framed'
            try:
                value = json.loads(body.decode('utf-8-sig'))
            except (UnicodeDecodeError, json.JSONDecodeError) as exc:
                raise ToolError('MCP stdio frame body must be a JSON object') from exc
            if not isinstance(value, dict):
                raise ToolError('MCP stdio frame body must be a JSON object')
            return value
        try:
            value = json.loads(raw.decode('utf-8-sig'))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ToolError('MCP stdio input must be a JSON object on one line') from exc
        if not isinstance(value, dict):
            raise ToolError('MCP stdio input must be a JSON object')
        return value


def tool_response(data: dict[str, Any], error: bool = False) -> dict[str, Any]:
    value: dict[str, Any] = {'content': [{'type': 'text', 'text': json.dumps(data, ensure_ascii=False, indent=2)}]}
    if error:
        value['isError'] = True
    return value


def schemas() -> list[dict[str, Any]]:
    common = {
        'workspace_dir': {'type': 'string'},
        'profile': {'type': 'string', 'description': 'Optional compatibility field. Omit it to use the latest YOHOHOHO website configuration.'},
        'prompt': {'type': 'string'},
        'size': {'type': 'string', 'default': 'auto'},
        'quality': {'type': 'string', 'enum': ['low', 'medium', 'high', 'auto'], 'default': 'medium'},
        'output_format': {'type': 'string', 'enum': ['png', 'jpeg', 'webp'], 'default': 'png'},
        'moderation': {'type': 'string', 'enum': ['auto', 'low'], 'default': 'auto'},
        'output_compression': {'type': 'integer', 'minimum': 0, 'maximum': 100},
        'count': {'type': 'integer', 'minimum': 1, 'maximum': MAX_BATCH, 'default': 1},
        'file_stem': {'type': 'string'},
        'execute': {'type': 'boolean', 'default': True},
    }
    edit = {**common, 'reference_dir': {'type': 'string'}, 'reference_glob': {'type': 'string', 'default': '*'}}
    return [
        {'name': 'gpt_image2_list_profiles', 'description': 'Read the current YOHOHOHO image platform and model from the website without exposing credentials.', 'inputSchema': {'type': 'object', 'properties': {}, 'additionalProperties': False}},
        {'name': 'gpt_image2_list_models', 'description': 'Read image models available to the current YOHOHOHO platform without generating an image.', 'inputSchema': {'type': 'object', 'properties': {'profile': {'type': 'string', 'description': 'Optional compatibility field; only the YOHOHOHO profile is available.'}}, 'additionalProperties': False}},
        {'name': 'gpt_image2_open_config_ui', 'description': 'Open the YOHOHOHO image configuration website.', 'inputSchema': {'type': 'object', 'properties': {}, 'additionalProperties': False}},
        {'name': 'gpt_image2_generate', 'description': 'Generate one to four images. Explicit image requests execute immediately unless execute=false is supplied.', 'inputSchema': {'type': 'object', 'properties': common, 'required': ['workspace_dir', 'prompt'], 'additionalProperties': False}},
        {'name': 'gpt_image2_edit_folder', 'description': 'Generate or edit from up to eight images in one reference folder. Only selected files are sent to the API.', 'inputSchema': {'type': 'object', 'properties': edit, 'required': ['workspace_dir', 'reference_dir', 'prompt'], 'additionalProperties': False}},
        {'name': 'gpt_image2_recent_outputs', 'description': 'List local generated images and manifests for explicit reuse.', 'inputSchema': {'type': 'object', 'properties': {'workspace_dir': {'type': 'string'}, 'profile': {'type': 'string'}, 'limit': {'type': 'integer', 'minimum': 1, 'maximum': 50, 'default': 10}}, 'required': ['workspace_dir'], 'additionalProperties': False}},
    ]


def text(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ToolError(f'{field} must be a non-empty string')
    return value.strip()


def integer(value: Any, minimum: int, maximum: int, field: str) -> int:
    try:
        result = int(value)
    except (TypeError, ValueError) as exc:
        raise ToolError(f'{field} must be an integer') from exc
    if not minimum <= result <= maximum:
        raise ToolError(f'{field} must be between {minimum} and {maximum}')
    return result


def choice(value: Any, choices: set[str], field: str) -> str:
    result = str(value or '').strip().lower()
    if result not in choices:
        raise ToolError(f'{field} must be one of: {", ".join(sorted(choices))}')
    return result


def profile_protocol(value: Any, field: str = 'protocol') -> str:
    return choice(value, PROFILE_PROTOCOLS, field)


def optional_profile_protocol(value: Any, field: str = 'protocol') -> str | None:
    """Read legacy profiles without choosing an interface on their behalf."""
    raw = str(value or '').strip().lower()
    if not raw or raw == 'auto':
        return None
    return profile_protocol(raw, field)


def infer_platform(item: dict[str, Any]) -> str:
    """Keep older profiles usable while making the platform explicit for new saves."""
    raw = str(item.get('platform') or '').strip().lower()
    if raw in IMAGE_PLATFORMS:
        return raw
    model = str(item.get('model') or '').strip().lower()
    if 'grok' in model:
        return 'grok'
    if 'nana-banana' in model or 'nanobanana' in model:
        return 'gemini_nanobanana'
    if 'gemini' in model or 'imagen' in model:
        return 'gemini'
    if 'gpt-image' in model or 'dall-e' in model:
        return 'openai'
    if str(item.get('protocol') or '').strip().lower() == 'gemini_native':
        return 'gemini_nanobanana'
    if str(item.get('protocol') or '').strip().lower() == 'chat_completions':
        return 'gemini'
    return 'other'


def profile_platform(value: Any, item: dict[str, Any] | None = None, field: str = 'platform') -> str:
    if value is None or not str(value).strip():
        return infer_platform(item or {})
    return choice(value, set(IMAGE_PLATFORMS), field)


def platform_protocol(platform: Any, legacy_protocol: Any = None) -> str:
    selected = profile_platform(platform)
    if selected in PLATFORM_PROTOCOLS:
        return PLATFORM_PROTOCOLS[selected]
    return profile_protocol(legacy_protocol)


def profile_id(value: Any) -> str:
    result = text(value, 'profile_id')
    if not re.fullmatch(PROFILE_ID_PATTERN, result):
        raise ToolError('profile_id must start with a-z or 0-9 and contain only lowercase letters, numbers, _ or -')
    return result


def normalize_base_url(value: Any, field: str = 'base_url') -> str:
    raw_base_url = text(value, field).rstrip('/')
    parsed_base_url = urllib.parse.urlsplit(raw_base_url)
    if parsed_base_url.scheme not in {'http', 'https'} or not parsed_base_url.netloc or parsed_base_url.query or parsed_base_url.fragment:
        raise ToolError(f'{field} must be an http(s) URL without query or fragment')
    path_parts = [part for part in parsed_base_url.path.split('/') if part]
    if 'v1' in path_parts:
        normalized_path = '/' + '/'.join(path_parts[:path_parts.index('v1') + 1])
    else:
        normalized_path = ('/' + '/'.join(path_parts) if path_parts else '') + '/v1'
    return urllib.parse.urlunsplit((parsed_base_url.scheme, parsed_base_url.netloc, normalized_path, '', '')).rstrip('/')


def size(value: Any) -> str:
    raw = str(value or 'auto').lower().replace('×', 'x')
    if raw == 'auto':
        return raw
    match = re.fullmatch(r'(\d{2,5})x(\d{2,5})', raw)
    if not match:
        raise ToolError('size must be auto or WIDTHxHEIGHT')
    width, height = int(match.group(1)), int(match.group(2))
    if width > MAX_EDGE or height > MAX_EDGE or width % 16 or height % 16:
        raise ToolError(f'size edges must be multiples of 16 and no larger than {MAX_EDGE}')
    if not MIN_PIXELS <= width * height <= MAX_PIXELS or max(width, height) / min(width, height) > 3:
        raise ToolError('size pixels or aspect ratio are outside GPT Image 2 limits')
    return f'{width}x{height}'


def workspace(value: Any) -> Path:
    path = Path(text(value, 'workspace_dir')).expanduser().resolve()
    if not path.is_dir():
        raise ToolError(f'workspace_dir is not an existing directory: {path}')
    return path


def remote_json(path: str) -> dict[str, Any]:
    try:
        api_key = load_api_key()
    except CredentialError as exc:
        raise ToolError(str(exc)) from exc
    request = urllib.request.Request(
        f'{CONFIG_API_ROOT}{path}',
        headers={
            'Authorization': f'Bearer {api_key}',
            'Accept': 'application/json',
            'User-Agent': 'YOHOHOHO-Image-Plugin/1.0',
        },
        method='GET',
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:  # noqa: S310 - fixed YOHOHOHO endpoint
            raw = response.read()
    except urllib.error.HTTPError as exc:
        try:
            payload = json.loads(exc.read().decode('utf-8', errors='replace'))
            message = (payload.get('error') or {}).get('message') if isinstance(payload, dict) else None
            code = (payload.get('error') or {}).get('code') if isinstance(payload, dict) else None
        except json.JSONDecodeError:
            message = None
            code = None
        if code == 'CONFIG_NOT_FOUND':
            raise ToolError(f'尚未保存生图配置，请打开 {CONFIG_WEB_URL} 选择平台和模型。') from exc
        raise ToolError(message or f'配置同步失败（HTTP {exc.code}），本次生图已停止。') from exc
    except (TimeoutError, socket.timeout) as exc:
        raise ToolError('YOHOHOHO 配置同步超时，本次生图已停止；未使用旧配置，也未自动重试。') from exc
    except urllib.error.URLError as exc:
        raise ToolError(f'无法连接 YOHOHOHO 配置服务：{exc.reason}。本次生图已停止。') from exc
    try:
        payload = json.loads(raw.decode('utf-8'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ToolError('YOHOHOHO 配置服务返回了无效数据，本次生图已停止。') from exc
    if not isinstance(payload, dict) or payload.get('ok') is not True:
        raise ToolError('YOHOHOHO 配置同步失败，本次生图已停止。')
    return payload


def profile_config() -> tuple[str, dict[str, dict[str, Any]]]:
    payload = remote_json('/config')
    config = payload.get('config')
    if not isinstance(config, dict):
        raise ToolError('YOHOHOHO 配置服务没有返回有效配置，本次生图已停止。')
    platform = profile_platform(config.get('platform'), config)
    model = text(config.get('model'), 'config.model')
    expected_protocol = PLATFORM_PROTOCOLS[platform]
    if config.get('protocol') not in (None, expected_protocol):
        raise ToolError('YOHOHOHO 平台协议配置不一致，本次生图已停止。')
    try:
        api_key = load_api_key()
    except CredentialError as exc:
        raise ToolError(str(exc)) from exc
    api_path = '/v1beta' if platform == 'gemini_nanobanana' else '/v1'
    profile = {
        'label': str(config.get('platform_label') or IMAGE_PLATFORMS[platform]),
        'base_url': f'{YOHO_GATEWAY_ORIGIN}{api_path}',
        'api_key': api_key,
        'platform': platform,
        'model': model,
        'protocol': expected_protocol,
        'timeout_seconds': 600,
        'response_format_b64_json': False,
        'stream_images': False,
        'stream_partial_images': 0,
    }
    return PUBLIC_PROFILE_ID, {PUBLIC_PROFILE_ID: profile}


def selected_profile(value: Any) -> tuple[str, dict[str, Any]]:
    active, profiles = profile_config()
    profile_id = text(value, 'profile') if value else active
    if profile_id not in profiles:
        raise ToolError('YOHOHOHO 插件只使用网站当前保存的生图配置，请省略 profile 参数后重试。')
    if value and profile_id != active:
        raise ToolError('拒绝使用非当前网站配置；请在 YOHOHOHO 配置页切换平台或模型。')
    return profile_id, profiles[profile_id]


def explicit_confirmation(value: Any) -> bool:
    """Require a real JSON boolean for an MCP-initiated state change."""
    if value is None:
        return False
    if not isinstance(value, bool):
        raise ToolError('confirm must be a boolean')
    return value


def pending_profile_switch(current: str, requested: str, operation: str = 'switch') -> dict[str, Any]:
    """Return a non-mutating approval request for a possible fallback switch."""
    return {
        'ok': False,
        'requires_confirmation': True,
        'operation': operation,
        'current_active_profile': current,
        'requested_profile': requested,
        'manual_action': 'configuration_ui',
        'message': (
            f'The manually selected active image service is {current!r}. '
            f'MCP automatic switching is blocked; the requested fallback is {requested!r}. '
            'No profile was changed and no paid retry was attempted. '
            'After the user explicitly approves, switch the service in the configuration UI. '
            'A confirm flag from the model is not treated as user approval.'
        ),
    }


def resolved_protocol(profile: dict[str, Any]) -> str:
    """Resolve the request format from the selected platform."""
    platform = profile_platform(profile.get('platform'), profile)
    if platform in PLATFORM_PROTOCOLS:
        return PLATFORM_PROTOCOLS[platform]
    configured = optional_profile_protocol(profile.get('protocol'))
    if configured is None:
        raise ToolError('This image service has no image platform configured. Open the configuration UI and choose an image platform.')
    return configured


def public_profile(profile_id: str, profile: dict[str, Any]) -> dict[str, Any]:
    return {
        'id': profile_id,
        'label': profile['label'],
        'base_url': profile['base_url'],
        'platform': profile_platform(profile.get('platform'), profile),
        'model': profile['model'],
        'protocol': profile.get('protocol'),
        'timeout_seconds': profile['timeout_seconds'],
        'stream_images': profile['stream_images'],
        'has_api_key': bool(profile.get('api_key')),
    }


def list_models(args: dict[str, Any]) -> dict[str, Any]:
    extra = set(args) - {'profile'}
    if extra:
        raise ToolError(f'Unsupported argument(s) for gpt_image2_list_models: {", ".join(sorted(extra))}')
    identifier, profile = selected_profile(args.get('profile'))
    selected_platform = profile['platform']
    payload = remote_json(f'/models?platform={urllib.parse.quote(selected_platform, safe="")}')
    image_models = payload.get('models')
    if not isinstance(image_models, list) or any(not isinstance(item, str) for item in image_models):
        raise ToolError('YOHOHOHO 配置服务返回了无效模型列表。')
    return {
        'ok': True,
        'profile_id': identifier,
        'base_url': profile['base_url'],
        'platform': selected_platform,
        'current_model': profile['model'],
        'models': image_models,
        'image_models': image_models,
        'current_model_available': profile['model'] in set(image_models),
        'message': f'已读取当前平台的 {len(image_models)} 个可用生图模型。',
    }


def open_config_ui(args: dict[str, Any]) -> dict[str, Any]:
    if args:
        raise ToolError('gpt_image2_open_config_ui accepts no arguments')
    opened = webbrowser.open(CONFIG_WEB_URL)
    return {
        'ok': True,
        'url': CONFIG_WEB_URL,
        'opened': bool(opened),
        'message': '请在 YOHOHOHO 图片生成配置服务中切换平台和模型。',
    }


def configure_profile(args: dict[str, Any]) -> dict[str, Any]:
    allowed = {'profile_id', 'label', 'base_url', 'api_key', 'platform', 'model', 'protocol', 'timeout_seconds', 'active', 'clear_api_key', 'confirm', 'execute', '_source'}
    extra = set(args) - allowed
    if extra:
        raise ToolError(f'Unsupported argument(s) for gpt_image2_configure_profile: {", ".join(sorted(extra))}')
    if False and args.get('active') and args.get('_source') != UI_SOURCE:  # legacy guard; confirmation flow below returns structured approval requests
        raise ToolError('禁止由 MCP 自动切换当前服务，请在配置界面中手动切换。')
    identifier = profile_id(args.get('profile_id'))
    data = read_profile_document()
    profiles = data['profiles']
    explicit_confirmation(args.get('confirm'))
    current_active = data.get('active_profile') if data.get('active_profile') in profiles else next(iter(profiles))
    if bool(args.get('active')) and args.get('_source') != UI_SOURCE:
        return pending_profile_switch(current_active, identifier, 'configure_profile')
    existing = profiles.get(identifier)
    if existing is not None and not isinstance(existing, dict):
        raise ToolError(f'Profile {identifier!r} must be an object')
    if existing is None and not args.get('base_url'):
        raise ToolError('A new profile requires base_url')
    if existing is None and not args.get('api_key'):
        raise ToolError('A new profile requires api_key')
    item = dict(existing or {})
    changed: list[str] = []
    if 'label' in args:
        item['label'] = text(args['label'], 'label')
        changed.append('label')
    if 'base_url' in args:
        item['base_url'] = normalize_base_url(args['base_url'])
        changed.append('base_url')
    elif 'base_url' not in item:
        raise ToolError('Profile requires base_url')
    if args.get('clear_api_key'):
        item.pop('api_key', None)
        changed.append('api_key_cleared')
    elif 'api_key' in args:
        item['api_key'] = text(args['api_key'], 'api_key')
        changed.append('api_key')
    if 'api_key' not in item or not str(item.get('api_key') or '').strip():
        raise ToolError('Profile requires api_key; pass api_key or keep the existing key')
    platform_selected = 'platform' in args
    if platform_selected:
        item['platform'] = profile_platform(args['platform'], item)
        changed.append('platform')
    elif existing is None:
        raise ToolError('A new profile requires an explicit image platform')
    if 'model' in args:
        item['model'] = text(args['model'], 'model')
        changed.append('model')
    if platform_selected:
        mapped_protocol = platform_protocol(item['platform'], item.get('protocol'))
        if item.get('protocol') != mapped_protocol:
            item['protocol'] = mapped_protocol
            changed.append('protocol')
    elif 'protocol' in args:
        item['protocol'] = profile_protocol(args['protocol'])
        changed.append('protocol')
    elif optional_profile_protocol(item.get('protocol')) is None:
        raise ToolError('请选择接口协议后再保存服务：OpenAI Images 或 Chat Completions。')
    if 'timeout_seconds' in args:
        item['timeout_seconds'] = integer(args['timeout_seconds'], 15, MAX_TIMEOUT_SECONDS, 'timeout_seconds')
        changed.append('timeout_seconds')
    item.setdefault('label', identifier)
    item.setdefault('model', 'gpt-image-2')
    item['platform'] = profile_platform(item.get('platform'), item)
    if item['platform'] in PLATFORM_PROTOCOLS:
        item['protocol'] = PLATFORM_PROTOCOLS[item['platform']]
    if optional_profile_protocol(item.get('protocol')) is None:
        raise ToolError('Profile requires an explicit protocol selected by the user.')
    item.setdefault('timeout_seconds', 600)
    execute = bool(args.get('execute', True))
    active_will_be = data.get('active_profile') if data.get('active_profile') in profiles else next(iter(profiles))
    if bool(args.get('active')):
        active_will_be = identifier
        changed.append('active_profile')
    profiles[identifier] = item
    data['version'] = data.get('version', 1)
    data['active_profile'] = active_will_be
    if execute:
        write_profile_document(data)
        active, profiles_public = public_profiles_from_document(data)
    else:
        # Validate the proposed document without touching disk and redact secrets in the preview.
        preview = dict(data)
        preview['profiles'] = dict(profiles)
        active, profiles_public = public_profiles_from_document(data)
        simulated = dict(next(profile for profile in profiles_public if profile['id'] == identifier))
        profiles_public = [simulated if profile['id'] == identifier else profile for profile in profiles_public]
    return {'ok': True, 'saved': execute, 'profile_id': identifier, 'changed': changed, 'active_profile': active_will_be if not execute else active, 'profile': next(profile for profile in profiles_public if profile['id'] == identifier), 'profiles': profiles_public, 'message': 'Profile updated locally.' if execute else 'Preview only. No file was changed.'}


def set_active_profile(args: dict[str, Any]) -> dict[str, Any]:
    if set(args) - {'profile_id', 'confirm', 'execute', '_source'}:
        raise ToolError('Unsupported argument for gpt_image2_set_active_profile')
    if False and args.get('_source') != UI_SOURCE:  # legacy guard; confirmation flow below returns structured approval requests
        raise ToolError('禁止由 MCP 自动切换当前服务，请在配置界面中手动切换。')
    identifier = profile_id(args.get('profile_id'))
    data = read_profile_document()
    if identifier not in data['profiles']:
        raise ToolError(f'Unknown profile {identifier!r}. Call gpt_image2_list_profiles first.')
    if optional_profile_protocol(data['profiles'][identifier].get('protocol')) is None:
        raise ToolError('请先为该服务选择接口协议，再将它设为当前服务。')
    explicit_confirmation(args.get('confirm'))
    current_active = data.get('active_profile') if data.get('active_profile') in data['profiles'] else next(iter(data['profiles']))
    execute = bool(args.get('execute', True))
    if identifier != current_active and args.get('_source') != UI_SOURCE:
        return pending_profile_switch(current_active, identifier)
    if execute:
        data['active_profile'] = identifier
        write_profile_document(data)
    active, profiles_public = public_profiles_from_document(data) if execute else (identifier, [public_profile(key, value) for key, value in profile_config()[1].items()])
    return {'ok': True, 'saved': execute, 'active_profile': active, 'profiles': profiles_public, 'message': 'Active profile changed.' if execute else 'Preview only. No file was changed.'}


def remove_profile(args: dict[str, Any]) -> dict[str, Any]:
    if set(args) - {'profile_id', 'execute', '_source'}:
        raise ToolError('Unsupported argument for gpt_image2_remove_profile')
    if args.get('_source') != UI_SOURCE:
        raise ToolError('服务删除和活动服务变更必须在配置界面中手动执行。')
    identifier = profile_id(args.get('profile_id'))
    data = read_profile_document()
    profiles = data['profiles']
    if identifier not in profiles:
        raise ToolError(f'Unknown profile {identifier!r}. Call gpt_image2_list_profiles first.')
    if len(profiles) <= 1:
        raise ToolError('Cannot remove the last image API profile')
    remaining = [key for key in profiles if key != identifier]
    new_active = data.get('active_profile')
    if new_active == identifier or new_active not in remaining:
        new_active = remaining[0]
    execute = bool(args.get('execute', True))
    if execute:
        del profiles[identifier]
        data['active_profile'] = new_active
        write_profile_document(data)
        active, profiles_public = public_profiles_from_document(data)
    else:
        current_profiles = profile_config()[1]
        profiles_public = [public_profile(key, value) for key, value in current_profiles.items() if key != identifier]
        active = new_active
    return {'ok': True, 'saved': execute, 'removed_profile': identifier, 'active_profile': active, 'profiles': profiles_public, 'message': 'Profile removed.' if execute else 'Preview only. No file was changed.'}


def request_args(args: dict[str, Any], profile: dict[str, Any]) -> tuple[dict[str, Any], int]:
    count = integer(args.get('count', 1), 1, MAX_BATCH, 'count')
    request = {
        'model': profile['model'],
        'prompt': text(args.get('prompt'), 'prompt'),
        'size': size(args.get('size', 'auto')),
        'quality': choice(args.get('quality', 'medium'), {'low', 'medium', 'high', 'auto'}, 'quality'),
        'output_format': choice(args.get('output_format', 'png'), {'png', 'jpeg', 'webp'}, 'output_format'),
        'moderation': choice(args.get('moderation', 'auto'), {'auto', 'low'}, 'moderation'),
    }
    if args.get('output_compression') is not None:
        if request['output_format'] == 'png':
            raise ToolError('output_compression is only valid for jpeg or webp')
        request['output_compression'] = integer(args['output_compression'], 0, 100, 'output_compression')
    if profile['response_format_b64_json']:
        request['response_format'] = 'b64_json'
    if profile['stream_images']:
        request['stream'] = True
        request['partial_images'] = profile['stream_partial_images']
    return request, count


def chat_request_args(request: dict[str, Any]) -> dict[str, Any]:
    """Build the broadly supported OpenAI-compatible chat request for image models."""
    body: dict[str, Any] = {
        'model': request['model'],
        'messages': [{'role': 'user', 'content': request['prompt']}],
        'modalities': ['text', 'image'],
    }
    image_config = {
        'size': request['size'],
        'quality': request['quality'],
        'output_format': request['output_format'],
    }
    if request.get('output_compression') is not None:
        image_config['output_compression'] = request['output_compression']
    if any(value not in {'auto', 'png'} for value in image_config.values() if isinstance(value, str)) or request['size'] != 'auto':
        body['image_config'] = image_config
    return body


def gemini_aspect_ratio(value: str) -> str | None:
    if value == 'auto':
        return None
    match = re.fullmatch(r'(\d+)x(\d+)', value)
    if not match:
        return None
    width, height = (int(part) for part in match.groups())
    supported = ('1:1', '2:3', '3:2', '3:4', '4:3', '4:5', '5:4', '9:16', '16:9', '21:9')
    return min(supported, key=lambda ratio: abs(width / height - (int(ratio.split(':')[0]) / int(ratio.split(':')[1]))))


def gemini_native_request_args(request: dict[str, Any]) -> dict[str, Any]:
    """Build the native Gemini GenerateContent request for image models."""
    generation_config: dict[str, Any] = {'responseModalities': ['TEXT', 'IMAGE']}
    aspect_ratio = gemini_aspect_ratio(request['size'])
    if aspect_ratio:
        generation_config['imageConfig'] = {'aspectRatio': aspect_ratio}
    return {
        'contents': [{'role': 'user', 'parts': [{'text': request['prompt']}]}],
        'generationConfig': generation_config,
    }


def openai_images_request_args(request: dict[str, Any]) -> dict[str, Any]:
    """Remove local-only automatic values before calling strict image APIs."""
    body = dict(request)
    if body.get('size') == 'auto':
        body.pop('size', None)
    return body


def safe_stem(value: Any, prompt: str) -> str:
    raw = text(value, 'file_stem') if value else prompt
    raw = re.sub(r'\s+', '-', raw.lower())
    raw = re.sub(r'[^a-z0-9\u4e00-\u9fff._-]+', '-', raw).strip('.-_')[:60] or 'gpt-image-2'
    return raw if value else f'{dt.datetime.now().strftime("%H%M%S")}-{raw}'


def references(directory: Any, pattern: Any) -> list[Path]:
    root = Path(text(directory, 'reference_dir')).expanduser().resolve()
    if not root.is_dir():
        raise ToolError(f'reference_dir is not an existing directory: {root}')
    files = sorted((Path(item).resolve() for item in glob.glob(str(root / str(pattern or '*'))) if Path(item).is_file() and Path(item).suffix.lower() in IMAGE_EXTENSIONS), key=lambda item: item.name.lower())
    if not files:
        raise ToolError(f'No reference images matched {pattern or "*"!r} in {root}')
    if len(files) > MAX_REFERENCES:
        raise ToolError(f'{len(files)} reference images exceeds the limit of {MAX_REFERENCES}')
    return files


def decode_image_base64(value: Any, source: str) -> bytes:
    if not isinstance(value, str) or not value.strip():
        raise ToolError(f'{source} did not include image data')
    raw = value.split(',', 1)[1] if value.startswith('data:') and ',' in value else value
    try:
        return base64.b64decode(raw, validate=True)
    except (ValueError, TypeError) as exc:
        raise ToolError(f'{source} returned invalid base64 image data') from exc


def download_image_url(image_url: str, source: str) -> bytes:
    if not re.match(r'^https?://', image_url, flags=re.IGNORECASE):
        if image_url.startswith('data:image/'):
            return decode_image_base64(image_url, source)
        raise ToolError(f'{source} returned an unsupported image URL')
    image_request = urllib.request.Request(image_url, headers={'User-Agent': 'GPT-Image-2-Studio/0.3'}, method='GET')
    try:
        with urllib.request.urlopen(image_request, timeout=60) as response:  # noqa: S310
            return response.read()
    except (TimeoutError, socket.timeout) as exc:
        raise ToolError(f'Unable to download image URL returned by {source}: request timed out') from exc
    except urllib.error.URLError as exc:
        raise ToolError(f'Unable to download image URL returned by {source}: {exc.reason}') from exc


def image_result(payload: Any, headers: dict[str, str]) -> dict[str, Any]:
    if isinstance(payload, list):
        items = payload
    elif isinstance(payload, dict) and isinstance(payload.get('data'), list):
        items = payload['data']
    elif isinstance(payload, dict) and (payload.get('b64_json') or payload.get('url')):
        items = [payload]
    else:
        items = []
    first = next((item for item in items if isinstance(item, dict) and (item.get('b64_json') or item.get('url'))), None)
    if not first:
        raise ToolError('Image API response did not include data[].b64_json or data[].url')
    if isinstance(first.get('b64_json'), str) and first['b64_json'].strip():
        content = decode_image_base64(first['b64_json'], 'Image API')
    elif isinstance(first.get('url'), str) and first['url'].strip():
        content = download_image_url(first['url'].strip(), 'Image API')
    else:
        raise ToolError('Image API response did not include usable image data')
    return {
        'image_bytes': content,
        'request_id': headers.get('x-request-id') or headers.get('openai-request-id'),
        'revised_prompt': first.get('revised_prompt'),
        'usage': payload.get('usage') if isinstance(payload, dict) else None,
    }


def chat_image_result(payload: Any, headers: dict[str, str], source: str = 'Chat image API') -> dict[str, Any]:
    """Extract images from chat or native Gemini candidate responses."""
    references: list[tuple[str, Any]] = []

    def collect(value: Any) -> None:
        if isinstance(value, list):
            for item in value:
                collect(item)
            return
        if isinstance(value, str):
            for match in re.findall(r'data:image/[^;]+;base64,[A-Za-z0-9+/=]+', value):
                references.append(('data_uri', match))
            return
        if not isinstance(value, dict):
            return
        if isinstance(value.get('b64_json'), str):
            references.append(('base64', value['b64_json']))
        if isinstance(value.get('url'), str):
            references.append(('url', value['url']))
        image_url = value.get('image_url')
        if isinstance(image_url, str):
            references.append(('url', image_url))
        elif isinstance(image_url, dict):
            collect(image_url)
        inline = value.get('inlineData') or value.get('inline_data')
        if isinstance(inline, dict) and isinstance(inline.get('data'), str):
            references.append(('base64', inline['data']))
        for key in ('images', 'image', 'content', 'parts', 'candidates', 'choices', 'message'):
            if key in value:
                collect(value[key])

    collect(payload)
    for kind, value in references:
        if kind in {'base64', 'data_uri'}:
            content = decode_image_base64(value, source)
        else:
            content = download_image_url(str(value), source)
        return {
            'image_bytes': content,
            'request_id': headers.get('x-request-id') or headers.get('openai-request-id'),
            'revised_prompt': None,
            'usage': payload.get('usage') if isinstance(payload, dict) else None,
        }
    raise ToolError(f'{source} response did not include an image part, inlineData, b64_json, or image URL')


def stream_image_result(raw: bytes, headers: dict[str, str]) -> dict[str, Any]:
    events = re.split(r'\r?\n\r?\n', raw.decode('utf-8', errors='replace'))
    last_error: str | None = None
    for event in events:
        lines = [line[5:].lstrip() for line in event.splitlines() if line.startswith('data:')]
        value = '\n'.join(lines).strip()
        if not value or value == '[DONE]':
            continue
        try:
            payload = json.loads(value)
        except json.JSONDecodeError as exc:
            raise ToolError('Image API stream included a non-JSON data event') from exc
        if not isinstance(payload, dict):
            continue
        error = payload.get('error')
        if isinstance(error, dict) and isinstance(error.get('message'), str):
            last_error = error['message']
            continue
        event_type = str(payload.get('type') or '')
        if event_type.endswith('.failed') and isinstance(payload.get('message'), str):
            last_error = payload['message']
            continue
        try:
            return image_result(payload, headers)
        except ToolError:
            continue
    if last_error:
        raise ToolError(f'Image API stream failed: {last_error}')
    raise ToolError('Image API stream did not include a completed image result')


def api_call(profile: dict[str, Any], endpoint: str, body: bytes, content_type: str) -> dict[str, Any]:
    request = urllib.request.Request(f'{profile["base_url"]}/{endpoint}', data=body, headers={'Authorization': f'Bearer {profile["api_key"]}', 'Content-Type': content_type, 'User-Agent': 'GPT-Image-2-Studio/0.3'}, method='POST')
    try:
        with urllib.request.urlopen(request, timeout=profile['timeout_seconds']) as response:  # noqa: S310
            raw = response.read()
            headers = {key.lower(): value for key, value in response.headers.items()}
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode('utf-8', errors='replace')
        try:
            error = json.loads(raw).get('error') or {}
            message = error.get('message') or raw
        except json.JSONDecodeError:
            message = raw[:1000]
        request_id = exc.headers.get('x-request-id') or exc.headers.get('openai-request-id')
        raise ToolError(f'Image API HTTP {exc.code}' + (f' request_id={request_id}' if request_id else '') + f': {message}') from exc
    except (TimeoutError, socket.timeout) as exc:
        raise ToolError(f'Image API request timed out after {profile["timeout_seconds"]} seconds') from exc
    except urllib.error.URLError as exc:
        raise ToolError(f'Network error calling image API: {exc.reason}') from exc
    content_type_header = headers.get('content-type', '').lower()
    if content_type_header.startswith('image/'):
        return {'image_bytes': raw, 'request_id': headers.get('x-request-id') or headers.get('openai-request-id'), 'revised_prompt': None, 'usage': None}
    if 'text/event-stream' in content_type_header:
        return stream_image_result(raw, headers)
    try:
        data = json.loads(raw.decode('utf-8'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        preview = raw[:240].decode('utf-8', errors='replace').replace('\n', ' ').strip()
        raise ToolError(f'Image API returned non-JSON content ({content_type_header or "unknown"}): {preview or "empty response"}') from exc
    return image_result(data, headers)


def chat_api_call(profile: dict[str, Any], body: dict[str, Any]) -> dict[str, Any]:
    request = urllib.request.Request(
        f'{profile["base_url"]}/chat/completions',
        data=json.dumps(body, ensure_ascii=False).encode('utf-8'),
        headers={
            'Authorization': f'Bearer {profile["api_key"]}',
            'Content-Type': 'application/json',
            'User-Agent': 'GPT-Image-2-Studio/0.3',
        },
        method='POST',
    )
    try:
        with urllib.request.urlopen(request, timeout=profile['timeout_seconds']) as response:  # noqa: S310
            raw = response.read()
            headers = {key.lower(): value for key, value in response.headers.items()}
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode('utf-8', errors='replace')
        try:
            error = json.loads(raw).get('error') or {}
            message = error.get('message') or raw
        except json.JSONDecodeError:
            message = raw[:1000]
        request_id = exc.headers.get('x-request-id') or exc.headers.get('openai-request-id')
        raise ToolError(f'Chat image API HTTP {exc.code}' + (f' request_id={request_id}' if request_id else '') + f': {message}') from exc
    except (TimeoutError, socket.timeout) as exc:
        raise ToolError(f'Chat image API request timed out after {profile["timeout_seconds"]} seconds') from exc
    except urllib.error.URLError as exc:
        raise ToolError(f'Network error calling Chat image API: {exc.reason}') from exc
    content_type_header = headers.get('content-type', '').lower()
    if content_type_header.startswith('image/'):
        return {'image_bytes': raw, 'request_id': headers.get('x-request-id') or headers.get('openai-request-id'), 'revised_prompt': None, 'usage': None}
    try:
        data = json.loads(raw.decode('utf-8'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        preview = raw[:240].decode('utf-8', errors='replace').replace('\n', ' ').strip()
        raise ToolError(f'Chat image API returned non-JSON content ({content_type_header or "unknown"}): {preview or "empty response"}') from exc
    if isinstance(data, dict) and isinstance(data.get('error'), dict):
        message = data['error'].get('message') or 'unknown error'
        raise ToolError(f'Chat image API error: {message}')
    return chat_image_result(data, headers)


def gemini_native_api_call(profile: dict[str, Any], body: dict[str, Any]) -> dict[str, Any]:
    """Call a native Gemini GenerateContent image endpoint."""
    model = urllib.parse.quote(profile['model'], safe='')
    endpoint = f'{profile["base_url"]}/models/{model}:generateContent'
    request = urllib.request.Request(
        endpoint,
        data=json.dumps(body, ensure_ascii=False).encode('utf-8'),
        headers={
            'Authorization': f'Bearer {profile["api_key"]}',
            'x-goog-api-key': profile['api_key'],
            'Content-Type': 'application/json',
            'User-Agent': 'GPT-Image-2-Studio/0.3',
        },
        method='POST',
    )
    try:
        with urllib.request.urlopen(request, timeout=profile['timeout_seconds']) as response:  # noqa: S310
            raw = response.read()
            headers = {key.lower(): value for key, value in response.headers.items()}
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode('utf-8', errors='replace')
        try:
            error = json.loads(raw).get('error') or {}
            message = error.get('message') or raw
        except json.JSONDecodeError:
            message = raw[:1000]
        request_id = exc.headers.get('x-request-id') or exc.headers.get('openai-request-id')
        raise ToolError(f'Gemini API HTTP {exc.code}' + (f' request_id={request_id}' if request_id else '') + f': {message}') from exc
    except (TimeoutError, socket.timeout) as exc:
        raise ToolError(f'Gemini API request timed out after {profile["timeout_seconds"]} seconds') from exc
    except urllib.error.URLError as exc:
        raise ToolError(f'Network error calling Gemini API: {exc.reason}') from exc
    content_type_header = headers.get('content-type', '').lower()
    if content_type_header.startswith('image/'):
        return {'image_bytes': raw, 'request_id': headers.get('x-request-id') or headers.get('openai-request-id'), 'revised_prompt': None, 'usage': None}
    try:
        data = json.loads(raw.decode('utf-8'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        preview = raw[:240].decode('utf-8', errors='replace').replace('\n', ' ').strip()
        raise ToolError(f'Gemini API returned non-JSON content ({content_type_header or "unknown"}): {preview or "empty response"}') from exc
    if isinstance(data, dict) and isinstance(data.get('error'), dict):
        message = data['error'].get('message') or 'unknown error'
        raise ToolError(f'Gemini API error: {message}')
    return chat_image_result(data, headers, 'Gemini API')


def multipart(fields: dict[str, str], files: list[Path]) -> tuple[bytes, str]:
    boundary = f'----gpt-image-2-studio-{uuid.uuid4().hex}'
    chunks: list[bytes] = []
    for name, value in fields.items():
        chunks += [f'--{boundary}\r\n'.encode(), f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(), value.encode(), b'\r\n']
    for path in files:
        mime = mimetypes.guess_type(path.name)[0] or 'application/octet-stream'
        chunks += [f'--{boundary}\r\n'.encode(), f'Content-Disposition: form-data; name="image[]"; filename="{path.name}"\r\nContent-Type: {mime}\r\n\r\n'.encode(), path.read_bytes(), b'\r\n']
    chunks.append(f'--{boundary}--\r\n'.encode())
    return b''.join(chunks), f'multipart/form-data; boundary={boundary}'


def output_dir(root: Path, create: bool) -> Path:
    path = root / 'output' / 'gpt-image-2' / dt.date.today().isoformat()
    if create:
        path.mkdir(parents=True, exist_ok=True)
    return path


def planned_path(directory: Path, stem: str, output_format: str) -> Path:
    suffix = '.jpg' if output_format == 'jpeg' else f'.{output_format}'
    candidate = directory / f'{stem}{suffix}'
    return candidate if not candidate.exists() else directory / f'{stem}-{uuid.uuid4().hex[:6]}{suffix}'


def generate(args: dict[str, Any], edit_files: list[Path] | None = None) -> dict[str, Any]:
    allowed = {'workspace_dir', 'profile', 'prompt', 'size', 'quality', 'output_format', 'moderation', 'output_compression', 'count', 'file_stem', 'execute'} | ({'reference_dir', 'reference_glob'} if edit_files is not None else set())
    extra = set(args) - allowed
    if extra:
        raise ToolError(f'Unsupported argument(s): {", ".join(sorted(extra))}')
    root = workspace(args.get('workspace_dir'))
    profile_id, profile = selected_profile(args.get('profile'))
    protocol = resolved_protocol(profile)
    if edit_files is not None and protocol in {'chat_completions', 'gemini_native'}:
        raise ToolError('Chat Completions and native Gemini image profiles currently support text-to-image only; use an OpenAI Images profile for reference edits.')
    request, count = request_args(args, profile)
    if protocol == 'chat_completions':
        api_request = chat_request_args(request)
    elif protocol == 'gemini_native':
        api_request = gemini_native_request_args(request)
    else:
        api_request = openai_images_request_args(request)
    execute = bool(args.get('execute', True))
    stem = safe_stem(args.get('file_stem'), request['prompt'])
    folder = output_dir(root, execute)
    planned = []
    for index in range(count):
        item_stem = stem if count == 1 else f'{stem}-{index + 1:02d}'
        image_path = planned_path(folder, item_stem, request['output_format'])
        planned.append((image_path, image_path.with_suffix(image_path.suffix + '.json')))
    result = {'ok': True, 'mode': 'edit_folder' if edit_files is not None else 'generate', 'execute': execute, 'profile': public_profile(profile_id, profile), 'protocol': protocol, 'request': request, 'api_request': api_request, 'count': count, 'outputs': [{'image_path': str(image), 'manifest_path': str(manifest)} for image, manifest in planned]}
    if edit_files is not None:
        result['reference_images'] = [str(path) for path in edit_files]
    if not execute:
        result['message'] = 'Dry run only. No API call was made.'
        return result
    started = time.monotonic()
    done = []
    for batch_index, (image_path, manifest_path) in enumerate(planned, start=1):
        try:
            if edit_files is None:
                if protocol == 'chat_completions':
                    api_result = chat_api_call(profile, api_request)
                elif protocol == 'gemini_native':
                    api_result = gemini_native_api_call(profile, api_request)
                else:
                    api_result = api_call(profile, 'images/generations', json.dumps(api_request).encode('utf-8'), 'application/json')
            else:
                body, content_type = multipart({key: str(value) for key, value in api_request.items()}, edit_files)
                api_result = api_call(profile, 'images/edits', body, content_type)
        except ToolError as exc:
            raise ToolError(
                f'YOHOHOHO 当前生图配置 {profile_id!r} 请求失败：{exc}。'
                '本次请求未自动重试，也未自动切换平台或模型；请在配置网站确认后再试。'
            ) from exc
        image_path.write_bytes(api_result['image_bytes'])
        manifest = {'tool': 'yohohoho-image', 'mode': result['mode'], 'created_at': dt.datetime.now(dt.UTC).isoformat(), 'profile_id': profile_id, 'api_base_url': profile['base_url'], 'protocol': protocol, 'request': request, 'api_request': api_request, 'output_path': str(image_path), 'reference_images': [str(path) for path in edit_files or []], 'batch_index': batch_index, 'batch_count': count, 'elapsed_seconds': round(time.monotonic() - started, 2), 'response': {key: api_result[key] for key in ('request_id', 'revised_prompt', 'usage')}}
        manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8')
        done.append({'image_path': str(image_path), 'manifest_path': str(manifest_path), 'request_id': api_result['request_id'], 'revised_prompt': api_result['revised_prompt']})
    result['outputs'] = done
    result['elapsed_seconds'] = round(time.monotonic() - started, 2)
    return result


def recent(args: dict[str, Any]) -> dict[str, Any]:
    if set(args) - {'workspace_dir', 'profile', 'limit'}:
        raise ToolError('Unsupported argument for gpt_image2_recent_outputs')
    root = workspace(args.get('workspace_dir')) / 'output' / 'gpt-image-2'
    limit = integer(args.get('limit', 10), 1, 50, 'limit')
    profile_id = text(args['profile'], 'profile') if args.get('profile') else None
    if not root.exists():
        return {'ok': True, 'root': str(root), 'items': []}
    items = []
    for image in sorted((path for path in root.rglob('*') if path.suffix.lower() in IMAGE_EXTENSIONS), key=lambda path: path.stat().st_mtime, reverse=True):
        manifest_path = image.with_suffix(image.suffix + '.json')
        try:
            manifest = json.loads(manifest_path.read_text(encoding='utf-8')) if manifest_path.is_file() else {}
        except (OSError, json.JSONDecodeError):
            manifest = {}
        if profile_id and manifest.get('profile_id') != profile_id:
            continue
        items.append({'image_path': str(image), 'manifest_path': str(manifest_path) if manifest_path.is_file() else None, 'profile_id': manifest.get('profile_id'), 'prompt': (manifest.get('request') or {}).get('prompt'), 'modified_at': dt.datetime.fromtimestamp(image.stat().st_mtime).isoformat(timespec='seconds')})
        if len(items) == limit:
            break
    return {'ok': True, 'root': str(root), 'items': items}


def dispatch(name: Any, args: dict[str, Any]) -> dict[str, Any]:
    if name == 'gpt_image2_list_profiles':
        if args:
            raise ToolError('gpt_image2_list_profiles accepts no arguments')
        active, profiles = profile_config()
        return {'ok': True, 'active_profile': active, 'profiles': [public_profile(profile_id, profile) for profile_id, profile in profiles.items()]}
    if name == 'gpt_image2_list_models':
        return list_models(args)
    if name == 'gpt_image2_open_config_ui':
        return open_config_ui(args)
    if name == 'gpt_image2_generate':
        return generate(args)
    if name == 'gpt_image2_edit_folder':
        return generate(args, references(args.get('reference_dir'), args.get('reference_glob', '*')))
    if name == 'gpt_image2_recent_outputs':
        return recent(args)
    raise ToolError(f'Unknown tool: {name}')


def handle_message(message: dict[str, Any]) -> dict[str, Any] | None:
    """Handle one JSON-RPC message without owning stdin/stdout.

    The stable bridge imports this function for every request, which lets an
    already connected Codex task pick up newer plugin code without replacing
    its stdio process.
    """
    request_id = message.get('id')
    method = message.get('method')
    if method == 'initialize':
        version = (message.get('params') or {}).get('protocolVersion') or '2024-11-05'
        return {'jsonrpc': '2.0', 'id': request_id, 'result': {'protocolVersion': version, 'capabilities': {'tools': {'listChanged': False}}, 'serverInfo': {'name': 'yohohoho-image', 'version': '1.0.0'}}}
    if method in {'initialized', 'notifications/initialized', 'notifications/cancelled', 'notifications/progress'}:
        return None
    if method == 'ping':
        return {'jsonrpc': '2.0', 'id': request_id, 'result': {}}
    if method == 'tools/list':
        return {'jsonrpc': '2.0', 'id': request_id, 'result': {'tools': schemas()}}
    if method == 'resources/list':
        return {'jsonrpc': '2.0', 'id': request_id, 'result': {'resources': []}}
    if method == 'resources/templates/list':
        return {'jsonrpc': '2.0', 'id': request_id, 'result': {'resourceTemplates': []}}
    if method == 'tools/call':
        params = message.get('params') or {}
        try:
            result = tool_response(dispatch(params.get('name'), params.get('arguments') or {}))
        except ToolError as exc:
            result = tool_response({'ok': False, 'error': str(exc)}, True)
        except Exception as exc:  # noqa: BLE001 - never drop the MCP transport on a tool bug
            print(traceback.format_exc(), file=sys.stderr, flush=True)
            result = tool_response({'ok': False, 'error': f'Image plugin internal error: {type(exc).__name__}: {exc}'}, True)
        return {'jsonrpc': '2.0', 'id': request_id, 'result': result}
    if request_id is not None:
        return {'jsonrpc': '2.0', 'id': request_id, 'error': {'code': -32601, 'message': f'Method not found: {method}'}}
    return None


def main() -> None:
    while True:
        request_id: Any = None
        try:
            message = receive()
            if message is None:
                return
            request_id = message.get('id')
            response = handle_message(message)
            if response is not None:
                send(response)
        except ToolError as exc:
            # A malformed message should not terminate the long-lived stdio server.
            print(f'MCP input error: {exc}', file=sys.stderr, flush=True)
            if request_id is not None:
                send({'jsonrpc': '2.0', 'id': request_id, 'error': {'code': -32600, 'message': str(exc)}})
        except Exception as exc:  # noqa: BLE001 - keep the connection recoverable
            print(traceback.format_exc(), file=sys.stderr, flush=True)
            if request_id is not None:
                send({'jsonrpc': '2.0', 'id': request_id, 'error': {'code': -32603, 'message': f'Image plugin internal error: {type(exc).__name__}: {exc}'}})


if __name__ == '__main__':
    main()
