auto-added claus
chatStreamSegments.push({ type: "tool", key, title: `${name} ${liveTag}`, body, sqlAudit, images });
};
const appendFinalAssistant = async (message, fallbackText) => {
+ const wsAttachments =
+ message && typeof message === "object" && message.attachments != null ? message.attachments : null;
const normalized = _normalizeAssistantMessage(message, { requireRole: false, requireContentArray: false });
if (normalized) {
const rows = _buildRenderRows(_expandAssistantMessageForRender(normalized));
const last = rows && rows.length ? rows[rows.length - 1] : null;
if (last) {
+ if (wsAttachments != null && (last.attachments == null || last.attachments === "")) {
+ last.attachments = wsAttachments;
+ }
if (streamRow && streamRow.parentNode) streamRow.remove();
await appendMessageRow(messagesEl, last, rowRenderOptions);
scrollMessagesToBottom(true);
@@ -4502,11 +4551,16 @@ ${autoLimit ? `auto-added claus
const t0 = String(fallbackText || "").trim();
if (t0 && !_isSilentReplyStream(t0)) {
if (streamRow && streamRow.parentNode) streamRow.remove();
- await appendMessageRow(messagesEl, {
- role: "assistant",
- content: decodeEscapedNewlines(t0),
- timestamp: new Date().toISOString(),
- }, rowRenderOptions);
+ await appendMessageRow(
+ messagesEl,
+ {
+ role: "assistant",
+ content: decodeEscapedNewlines(t0),
+ timestamp: new Date().toISOString(),
+ ...(wsAttachments != null ? { attachments: wsAttachments } : {}),
+ },
+ rowRenderOptions,
+ );
scrollMessagesToBottom(true);
return true;
}
diff --git a/interfaces/ws/turn_runner.py b/interfaces/ws/turn_runner.py
index 1f7ffe94..23bda0de 100644
--- a/interfaces/ws/turn_runner.py
+++ b/interfaces/ws/turn_runner.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
+import json
import threading
import uuid
from datetime import datetime
@@ -13,6 +14,28 @@ from oclaw.platform.config.paths import db_path
from oclaw.platform.persistence.sqlite_store import SqliteStore
+def _persisted_chat_attachments_nonempty(raw: Any) -> bool:
+ """True when chat_message.attachments has at least one JSON object (list or dict or encoded string)."""
+ if raw is None:
+ return False
+ if isinstance(raw, list):
+ return len(raw) > 0
+ if isinstance(raw, dict):
+ return bool(raw)
+ s = str(raw).strip()
+ if not s or s.lower() == "null":
+ return False
+ try:
+ parsed = json.loads(s)
+ if isinstance(parsed, list):
+ return len(parsed) > 0
+ if isinstance(parsed, dict):
+ return bool(parsed)
+ except Exception:
+ return bool(s)
+ return False
+
+
async def run_agent_turn_via_bridge(
*,
conn: Any,
@@ -533,12 +556,13 @@ async def run_agent_turn_via_bridge(
pass
elif run_status != "failed":
try:
- persisted = store.get_messages(session_id=session_id, limit=12)
+ persisted = store.get_messages(session_id=session_id, limit=64)
for m in reversed(list(persisted or [])):
if str(getattr(m, "role", "") or "").lower() != "assistant":
continue
content = str(getattr(m, "content", "") or "")
- if not content.strip():
+ atraw = getattr(m, "attachments", None)
+ if not content.strip() and not _persisted_chat_attachments_nonempty(atraw):
continue
final_text = content
final_msg = {
@@ -547,7 +571,7 @@ async def run_agent_turn_via_bridge(
"content": content,
"timestamp": str(getattr(m, "timestamp", "") or ""),
"tool_calls": getattr(m, "tool_calls", None),
- "attachments": getattr(m, "attachments", None),
+ "attachments": atraw,
}
break
except Exception:
diff --git a/platform/llm/image_http_common.py b/platform/llm/image_http_common.py
index 1fe936ba..8ba5a1a6 100644
--- a/platform/llm/image_http_common.py
+++ b/platform/llm/image_http_common.py
@@ -4,7 +4,9 @@ from __future__ import annotations
import base64
import io
+import json
import os
+import re
import time
from typing import Any
@@ -18,6 +20,32 @@ def join_url(base: str, path: str) -> str:
return f"{b}/{p}"
+def dashscope_native_multimodal_url_from_compatible_base(base_url: str) -> str | None:
+ """
+ ``qwen-image`` and similar models often return an **empty** OpenAI ``chat/completions`` message on
+ ``/compatible-mode/v1``. Native HTTP uses ``/api/v1/services/aigc/multimodal-generation/generation``
+ and returns ``output.choices[].message.content[].image`` URLs.
+
+ Override: ``AIA_IMAGE_EXPERT_DASHSCOPE_NATIVE_URL`` (full URL).
+ Path suffix: ``AIA_IMAGE_EXPERT_DASHSCOPE_NATIVE_PATH`` (default multimodal-generation path).
+ """
+ explicit = (os.getenv("AIA_IMAGE_EXPERT_DASHSCOPE_NATIVE_URL") or "").strip().rstrip("/")
+ if explicit:
+ return explicit
+ b = (base_url or "").strip().rstrip("/")
+ if not b or "compatible-mode" not in b.lower():
+ return None
+ root = re.sub(r"/compatible-mode/v\d+(?:/.*)?$", "", b, flags=re.IGNORECASE).rstrip("/")
+ if not root or root == b:
+ root = re.sub(r"/compatible-mode/?$", "", b, flags=re.IGNORECASE).rstrip("/")
+ if not root:
+ return None
+ path = (os.getenv("AIA_IMAGE_EXPERT_DASHSCOPE_NATIVE_PATH") or "").strip().lstrip("/")
+ if not path:
+ path = "api/v1/services/aigc/multimodal-generation/generation"
+ return f"{root}/{path}"
+
+
def env_ocr_lane_api_key() -> str:
return (os.getenv("AIA_OCR_API_KEY") or "").strip()
@@ -35,6 +63,25 @@ def env_ocr_lane_chat_endpoint() -> str:
return raw or "/chat/completions"
+def env_image_expert_api_key() -> str:
+ """Bearer key for **image specialist** multimodal/gen HTTP only (never shared with OCR ``AIA_OCR_*``)."""
+
+ return (os.getenv("AIA_IMAGE_EXPERT_API_KEY") or "").strip()
+
+
+def env_image_expert_base_url() -> str:
+ return (os.getenv("AIA_IMAGE_EXPERT_BASE_URL") or "").strip()
+
+
+def env_image_expert_model() -> str:
+ return (os.getenv("AIA_IMAGE_EXPERT_MODEL") or "").strip()
+
+
+def env_image_expert_chat_endpoint() -> str:
+ raw = (os.getenv("AIA_IMAGE_EXPERT_CHAT_ENDPOINT") or "").strip()
+ return raw or "/chat/completions"
+
+
def is_data_url(s: str) -> bool:
return s.startswith("data:") and ";base64," in s
@@ -97,51 +144,406 @@ def post_with_retry(
raise last_exc
+def env_image_expert_download_timeout_sec() -> float:
+ """Read timeout for DashScope/OSS signed result URLs (official samples use ~300s)."""
+ raw = (os.getenv("AIA_IMAGE_EXPERT_DOWNLOAD_TIMEOUT_SEC") or "").strip()
+ if not raw:
+ return 300.0
+ try:
+ v = float(raw)
+ except ValueError:
+ return 300.0
+ return max(15.0, min(v, 900.0))
+
+
+def download_http_url_bytes(
+ url: str,
+ *,
+ timeout_sec: float | None = None,
+ user_agent: str | None = None,
+) -> tuple[bytes, str]:
+ """Streamed GET with ``raise_for_status`` — matches DashScope OSS download guidance (long reads)."""
+ t = float(timeout_sec) if timeout_sec is not None else env_image_expert_download_timeout_sec()
+ ua = (user_agent or "").strip() or (
+ "Mozilla/5.0 (compatible; oclaw-image-expert/1.0; +https://github.com/)"
+ )
+ req_headers = {"User-Agent": ua, "Accept": "*/*"}
+ connect_cap = min(45.0, max(10.0, t / 10.0))
+ timeout = httpx.Timeout(t, connect=connect_cap)
+ with httpx.Client(timeout=timeout, follow_redirects=True) as client:
+ with client.stream("GET", url, headers=req_headers) as r:
+ r.raise_for_status()
+ ctype = str(r.headers.get("content-type") or "").split(";", 1)[0].strip()
+ parts: list[bytes] = []
+ for chunk in r.iter_bytes(chunk_size=65_536):
+ parts.append(chunk)
+ return b"".join(parts), ctype
+
+
+def dashscope_multimodal_http_ok(body: dict[str, Any]) -> tuple[bool, str]:
+ """DashScope HTTP often uses HTTP 200 with a business-level ``code`` for failures."""
+ sc = body.get("status_code")
+ if isinstance(sc, str) and sc.strip().isdigit():
+ sc = int(sc.strip())
+ if isinstance(sc, int) and sc >= 400:
+ return False, str(body.get("message") or body.get("msg") or f"status_code={sc}")
+ code = body.get("code")
+ if code is None:
+ return True, ""
+ if isinstance(code, int):
+ if code == 200:
+ return True, ""
+ msg = str(body.get("message") or body.get("msg") or "").strip()
+ return False, msg or str(code)
+ cs = str(code).strip()
+ if cs == "":
+ return True, ""
+ if cs.lower() in ("success", "ok", "200"):
+ return True, ""
+ msg = str(body.get("message") or body.get("msg") or "").strip()
+ return False, msg or cs
+
+
+def _extract_response_roots(resp_json: dict[str, Any]) -> list[dict[str, Any]]:
+ """Collect dict slices that may carry ``choices`` / ``messages`` (DashScope nests under ``output``)."""
+ roots: list[dict[str, Any]] = []
+ seen: set[int] = set()
+
+ def push(d: dict[str, Any]) -> None:
+ i = id(d)
+ if i in seen:
+ return
+ seen.add(i)
+ roots.append(d)
+
+ push(resp_json)
+ nested = resp_json.get("output")
+ if isinstance(nested, dict):
+ push(nested)
+ deeper = nested.get("output")
+ if isinstance(deeper, dict):
+ push(deeper)
+ result = resp_json.get("result")
+ if isinstance(result, dict):
+ push(result)
+ data_obj = resp_json.get("data")
+ if isinstance(data_obj, dict):
+ push(data_obj)
+ return roots
+
+
+def _append_image_from_any(images: list[str], raw: Any) -> None:
+ if isinstance(raw, str) and raw.strip():
+ s = raw.strip()
+ if s.startswith(("http://", "https://", "data:")):
+ images.append(s)
+ return
+ if isinstance(raw, dict):
+ u = raw.get("url")
+ if isinstance(u, str) and u.strip() and u.strip().startswith(("http://", "https://")):
+ images.append(u.strip())
+
+
+def _parse_content_part(it: Any, *, text_parts: list[str], images: list[str]) -> None:
+ """Parse one multimodal element (DashScope shorthand, OpenAI-style typed blocks, etc.)."""
+ if isinstance(it, str) and it.strip():
+ text_parts.append(it.strip())
+ return
+ if not isinstance(it, dict):
+ return
+ typ = str(it.get("type") or "").strip().lower()
+ if typ == "text":
+ tx = it.get("text")
+ if isinstance(tx, str) and tx.strip():
+ text_parts.append(tx.strip())
+ return
+ if typ == "input_text":
+ tx = it.get("text")
+ if isinstance(tx, str) and tx.strip():
+ text_parts.append(tx.strip())
+ return
+ if typ == "image_url":
+ iu = it.get("image_url")
+ if isinstance(iu, dict):
+ _append_image_from_any(images, iu)
+ elif isinstance(iu, str):
+ _append_image_from_any(images, iu)
+ return
+ if typ == "input_image":
+ iu = it.get("image_url")
+ if isinstance(iu, dict):
+ _append_image_from_any(images, iu)
+ elif isinstance(iu, str):
+ _append_image_from_any(images, iu)
+ return
+ if typ == "image" and it.get("image") is not None:
+ _append_image_from_any(images, it.get("image"))
+ return
+ # Untyped / DashScope shorthand blocks
+ tx = it.get("text")
+ if isinstance(tx, str) and tx.strip():
+ text_parts.append(tx.strip())
+ if it.get("image") is not None:
+ _append_image_from_any(images, it.get("image"))
+ ius = it.get("image_url")
+ if isinstance(ius, str):
+ _append_image_from_any(images, ius)
+ elif isinstance(ius, dict):
+ _append_image_from_any(images, ius)
+ uu = it.get("url")
+ if isinstance(uu, str) and uu.strip().startswith(("http://", "https://")):
+ images.append(uu.strip())
+ if isinstance(it.get("b64_json"), str) and str(it.get("b64_json")).strip():
+ images.append(f"data:image/png;base64,{it['b64_json']}")
+
+
+def _try_parse_content_json_string(raw: str, *, text_parts: list[str], images: list[str]) -> bool:
+ """Some proxies stringify multimodal ``content`` as JSON."""
+ s = raw.strip()
+ if not s or s[0] not in "[{":
+ return False
+ try:
+ parsed = json.loads(s)
+ except Exception:
+ return False
+ if isinstance(parsed, list):
+ for it in parsed:
+ _parse_content_part(it, text_parts=text_parts, images=images)
+ return True
+ if isinstance(parsed, dict):
+ _parse_content_part(parsed, text_parts=text_parts, images=images)
+ return True
+ return False
+
+
+def _harvest_http_urls_from_value(val: Any, images: list[str], *, depth: int, max_depth: int = 6) -> None:
+ """Last resort: collect image URLs nested under ``message`` (unknown multimodal block shapes)."""
+ if depth > max_depth or val is None:
+ return
+ if isinstance(val, str):
+ s = val.strip()
+ if s.startswith(("http://", "https://")) and s not in images:
+ images.append(s)
+ return
+ if isinstance(val, dict):
+ for _k, v in val.items():
+ _harvest_http_urls_from_value(v, images, depth=depth + 1, max_depth=max_depth)
+ return
+ if isinstance(val, list):
+ for it in val:
+ _harvest_http_urls_from_value(it, images, depth=depth + 1, max_depth=max_depth)
+
+
+def _parse_choice_message_content(msg: dict[str, Any], *, text_parts: list[str], images: list[str]) -> None:
+ c = msg.get("content")
+ if c is None:
+ _harvest_http_urls_from_value(msg, images, depth=0)
+ return
+ if isinstance(c, str):
+ if c.strip() and _try_parse_content_json_string(c, text_parts=text_parts, images=images):
+ _harvest_http_urls_from_value(msg, images, depth=0)
+ return
+ if c.strip():
+ text_parts.append(c.strip())
+ _harvest_http_urls_from_value(msg, images, depth=0)
+ return
+ if isinstance(c, dict):
+ _parse_content_part(c, text_parts=text_parts, images=images)
+ _harvest_http_urls_from_value(msg, images, depth=0)
+ return
+ if isinstance(c, list):
+ for it in c:
+ _parse_content_part(it, text_parts=text_parts, images=images)
+ _harvest_http_urls_from_value(msg, images, depth=0)
+ return
+
+
+def _parse_messages_array_for_assistant(
+ messages: list[Any], *, text_parts: list[str], images: list[str]
+) -> None:
+ for m in reversed(messages):
+ if not isinstance(m, dict):
+ continue
+ if str(m.get("role") or "").strip().lower() != "assistant":
+ continue
+ _parse_choice_message_content(m, text_parts=text_parts, images=images)
+ break
+
+
def extract_text_and_images(resp_json: dict[str, Any]) -> tuple[str, list[str]]:
text_parts: list[str] = []
images: list[str] = []
- choices = resp_json.get("choices")
+ roots = _extract_response_roots(resp_json)
+
+ for root in roots:
+ ot = root.get("text")
+ if isinstance(ot, str) and ot.strip():
+ text_parts.append(ot.strip())
+
+ choices: list[Any] | None = None
+ for root in roots:
+ ch = root.get("choices")
+ if isinstance(ch, list) and ch:
+ choices = ch
+ break
+
if isinstance(choices, list) and choices:
- msg = choices[0].get("message") if isinstance(choices[0], dict) else None
- if isinstance(msg, dict):
- c = msg.get("content")
- if isinstance(c, str):
- text_parts.append(c)
- elif isinstance(c, list):
- for it in c:
- if not isinstance(it, dict):
- continue
- if isinstance(it.get("text"), str):
- text_parts.append(str(it.get("text")))
- elif isinstance(it.get("image"), str):
- images.append(str(it.get("image")))
- elif isinstance(it.get("image_url"), str):
- images.append(str(it.get("image_url")))
- elif isinstance(it.get("image_url"), dict) and isinstance(it["image_url"].get("url"), str):
- images.append(str(it["image_url"]["url"]))
- elif isinstance(it.get("b64_json"), str):
- images.append(f"data:image/png;base64,{it['b64_json']}")
+ for choice in choices:
+ if not isinstance(choice, dict):
+ continue
+ legacy_txt = choice.get("text")
+ if isinstance(legacy_txt, str) and legacy_txt.strip():
+ text_parts.append(legacy_txt.strip())
+ msg = choice.get("message")
+ if isinstance(msg, dict):
+ _parse_choice_message_content(msg, text_parts=text_parts, images=images)
+ elif not isinstance(msg, dict):
+ _harvest_http_urls_from_value(choice, images, depth=0)
+
+ if not text_parts and not images:
+ for root in roots:
+ msgs = root.get("messages")
+ if isinstance(msgs, list) and msgs:
+ _parse_messages_array_for_assistant(msgs, text_parts=text_parts, images=images)
+ if text_parts or images:
+ break
if not images:
- data = resp_json.get("data")
- if isinstance(data, list):
+ for root in roots:
+ data = root.get("data")
+ if not isinstance(data, list):
+ continue
for it in data:
if not isinstance(it, dict):
continue
- if isinstance(it.get("url"), str):
- images.append(str(it.get("url")))
- if isinstance(it.get("b64_json"), str):
+ if isinstance(it.get("url"), str) and str(it.get("url")).strip():
+ images.append(str(it.get("url")).strip())
+ if isinstance(it.get("b64_json"), str) and str(it.get("b64_json")).strip():
images.append(f"data:image/png;base64,{it['b64_json']}")
+
+ if not images:
+ for root in roots:
+ for key in ("results", "artifacts", "task_outputs"):
+ arr = root.get(key)
+ if not isinstance(arr, list):
+ continue
+ for it in arr:
+ if not isinstance(it, dict):
+ continue
+ for ik in ("image", "url", "output_image_url", "image_url"):
+ v = it.get(ik)
+ if isinstance(v, str) and v.strip().startswith(("http://", "https://")):
+ images.append(v.strip())
+ elif isinstance(v, dict) and isinstance(v.get("url"), str):
+ u = str(v.get("url")).strip()
+ if u.startswith(("http://", "https://")):
+ images.append(u)
+
+ # De-dupe URLs while preserving order
+ seen_u: set[str] = set()
+ uniq_images: list[str] = []
+ for u in images:
+ if u not in seen_u:
+ seen_u.add(u)
+ uniq_images.append(u)
+ images = uniq_images
+
return "\n".join([x for x in text_parts if x]).strip(), images
+def _diag_fill_choices(out: dict[str, Any], ch: Any, *, prefix: str) -> None:
+ out[f"{prefix}choices_typename"] = type(ch).__name__
+ out[f"{prefix}choices_len"] = len(ch) if isinstance(ch, list) else None
+ if not isinstance(ch, list) or not ch or not isinstance(ch[0], dict):
+ return
+ c0 = ch[0]
+ out[f"{prefix}choice0_keys"] = sorted(c0.keys())[:24]
+ msg = c0.get("message")
+ if isinstance(msg, dict):
+ out[f"{prefix}msg_keys"] = sorted(msg.keys())[:24]
+ c = msg.get("content")
+ out[f"{prefix}content_typename"] = type(c).__name__
+ if isinstance(c, list):
+ out[f"{prefix}content_len"] = len(c)
+ if c:
+ out[f"{prefix}content0_typename"] = type(c[0]).__name__
+ if isinstance(c[0], dict):
+ out[f"{prefix}content0_keys"] = sorted(c[0].keys())[:24]
+ elif isinstance(c, dict):
+ out[f"{prefix}content_keys"] = sorted(c.keys())[:24]
+ elif c is None:
+ out[f"{prefix}content_is_null"] = True
+
+
+def build_extract_diag_empty(obj: dict[str, Any]) -> dict[str, Any]:
+ """Compact shape hints when extraction yielded nothing (stderr / chat debug)."""
+ out: dict[str, Any] = {"top_level_keys": sorted(obj.keys())[:40]}
+ ch_top = obj.get("choices")
+ if ch_top is not None:
+ _diag_fill_choices(out, ch_top, prefix="")
+ outp = obj.get("output")
+ if isinstance(outp, dict):
+ out["output_keys"] = sorted(outp.keys())[:40]
+ ch = outp.get("choices")
+ _diag_fill_choices(out, ch, prefix="output_")
+ return out
+
+
+def redact_response_for_debug(obj: Any, *, max_chars: int = 2800) -> str:
+ """JSON preview for error surfaces (truncate; redact long base64 / data URLs)."""
+
+ def _walk(x: Any) -> Any:
+ if isinstance(x, dict):
+ return {str(k): _walk(v) for k, v in x.items()}
+ if isinstance(x, list):
+ return [_walk(v) for v in x[:80]]
+ if isinstance(x, str):
+ s = x
+ if ";base64," in s and s.strip().startswith("data:") and len(s) > 120:
+ h, _, _ = s.partition(";base64,")
+ return f"{h};base64,"
+ if len(s) > 500:
+ return s[:400] + f"..."
+ return s
+ return x
+
+ try:
+ txt = json.dumps(_walk(obj), ensure_ascii=False, default=str)
+ except Exception:
+ txt = str(obj)
+ if len(txt) > max_chars:
+ return txt[: max_chars - 24] + "\n…"
+ return txt
+
+
+def format_extract_diag(diag: dict[str, Any], *, max_chars: int = 900) -> str:
+ try:
+ s = json.dumps(diag, ensure_ascii=False, default=str)
+ except Exception:
+ s = str(diag)
+ if len(s) > max_chars:
+ return s[: max_chars - 20] + "…"
+ return s
+
+
__all__ = [
+ "build_extract_diag_empty",
"compress_data_url_image",
+ "dashscope_multimodal_http_ok",
+ "dashscope_native_multimodal_url_from_compatible_base",
+ "download_http_url_bytes",
+ "env_image_expert_download_timeout_sec",
+ "redact_response_for_debug",
+ "env_image_expert_api_key",
+ "env_image_expert_base_url",
+ "env_image_expert_chat_endpoint",
+ "env_image_expert_model",
"env_ocr_lane_api_key",
"env_ocr_lane_base_url",
"env_ocr_lane_chat_endpoint",
"env_ocr_lane_model",
"extract_text_and_images",
+ "format_extract_diag",
"is_data_url",
"join_url",
"post_with_retry",
diff --git a/platform/llm/image_legacy_client.py b/platform/llm/image_legacy_client.py
index 08c63a32..cfa9bd64 100644
--- a/platform/llm/image_legacy_client.py
+++ b/platform/llm/image_legacy_client.py
@@ -1,28 +1,373 @@
"""Legacy image+text payloads for DashScope-style gateways.
Uses ``{"image":...}/{"text":...}`` or typed compatible-mode blocks on ``/chat/completions`` only.
-Prefer :mod:`oclaw.platform.llm.image_ocr_client` for OpenAI-compatible vision (图片专家已改用该路径).
+The **image specialist** uses this module from:
+
+- :mod:`~oclaw.runtime.agents.specialist_agent` (orchestration temp sessions)
+- :mod:`~oclaw.runtime.direct_loop` when ``skill_binding_role=="image"`` (**gateway /chat UI**), so vision
+ turns never hit :class:`~oclaw.platform.llm.transports.openai_responses.OpenAIResponsesModel` unless explicitly disabled via env.
+
+Alignment with Alibaba ``dashscope.MultiModalConversation`` examples (**message shape**):
+ ``messages = [{"role": "user", "content": [{"image": ""}, {"text": "..."}]}]``
+matches our non--``typed`` branch (same ``image`` / ``text`` keys as the SDK doc).
+
+Lane separation (**not** OCR):
+- Resolved as ``kwargs …`` from the user's **chosen chat model/profile** first, then ``AIA_IMAGE_EXPERT_*`` when a field is empty.
+ The **`AIA_OCR_*`** variables remain for **`query_image_attachment`** / OCR downgrade only.
+- Bearer HTTP targets ``BASE_URL`` + ``AIA_IMAGE_EXPERT_CHAT_ENDPOINT`` (default ``/chat/completions``), not ``MultiModalConversation`` SDK.
+
+SDK-style extras:
+- Optional top-level fields (``stream``, ``n``, ``watermark``, ``negative_prompt``, ``prompt_extend``, ``size``, …):
+ use ``DASHSCOPE_IMAGE_*`` env vars or JSON in ``AIA_IMAGE_EXPERT_REQUEST_EXTRA`` (alias: ``AIA_LEGACY_IMAGE_REQUEST_EXTRA``).
+
+Compatibility roots:
+- For OpenAI-compat multimodal, use ``AIA_IMAGE_EXPERT_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1`` etc.
+
+For OpenAI-style ``image_url`` chat payloads (tool OCR / multimodal downgrade), use :mod:`oclaw.platform.llm.image_ocr_client`.
+
+Chat UI **图片专家**端到端说明（与其它链路隔离的变更边界）见仓库内 ``docs/IMAGE_SPECIALIST_LANE.md``.
"""
from __future__ import annotations
+import base64
+import json
+import os
+import sys
from typing import Any
import httpx
+from oclaw.platform.files.attachment_assets import AttachmentAssetStore
from oclaw.platform.llm.image_http_common import (
+ build_extract_diag_empty,
compress_data_url_image,
- env_ocr_lane_api_key,
- env_ocr_lane_base_url,
- env_ocr_lane_model,
- env_ocr_lane_chat_endpoint,
+ dashscope_multimodal_http_ok,
+ dashscope_native_multimodal_url_from_compatible_base,
+ download_http_url_bytes,
+ env_image_expert_api_key,
+ env_image_expert_base_url,
+ env_image_expert_chat_endpoint,
+ env_image_expert_model,
extract_text_and_images,
+ format_extract_diag,
is_data_url,
join_url,
post_with_retry,
+ redact_response_for_debug,
)
from oclaw.runtime.prompt_templates import render_prompt
+IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH = (
+ "请根据用户上传的图片作答：描述可见场景、物体与文字；不确定处请标明。"
+)
+
+
+def collect_legacy_lane_images_from_attachments(
+ attachments: list[dict[str, Any]] | None,
+ *,
+ max_images: int = 3,
+) -> list[str]:
+ """Normalize incoming UI/store attachments to URLs/data URLs for :func:`send_legacy_image_messages`."""
+ from oclaw.platform.files.attachment_assets import attachment_id_to_data_url
+
+ cap = max(1, min(int(max_images), 12))
+ out: list[str] = []
+ for att in attachments or []:
+ if not isinstance(att, dict):
+ continue
+ t = str(att.get("type") or "").strip().lower()
+ if t == "image_ref":
+ aid = str(att.get("attachment_id") or "").strip()
+ if not aid:
+ continue
+ data_url = attachment_id_to_data_url(aid, mime=str(att.get("mime") or ""))
+ if data_url:
+ out.append(data_url)
+ elif t in ("input_image", "image"):
+ raw = str(att.get("image_base64") or att.get("data") or "").strip()
+ if raw:
+ mime = str(att.get("mime") or "image/jpeg")
+ if raw.startswith("data:"):
+ out.append(raw)
+ else:
+ out.append(f"data:{mime};base64,{raw}")
+ elif t == "image_url":
+ u = str(att.get("url") or "").strip()
+ if u:
+ out.append(u)
+ if len(out) >= cap:
+ break
+ return out
+
+
+def normalize_legacy_output_image_urls(resp_images: Any, *, max_items: int = 12) -> list[str]:
+ """Flatten provider ``images`` / content parts to HTTP/data URLs (strings only)."""
+ cap = max(1, min(int(max_items), 24))
+ out: list[str] = []
+ if resp_images is None:
+ return out
+ if isinstance(resp_images, str):
+ s = resp_images.strip()
+ if s:
+ out.append(s)
+ return out[:cap]
+ if not isinstance(resp_images, list):
+ return out
+ for it in resp_images[:cap]:
+ if isinstance(it, str):
+ s = it.strip()
+ if s:
+ out.append(s)
+ continue
+ if not isinstance(it, dict):
+ continue
+ u = it.get("image")
+ if isinstance(u, str) and u.strip():
+ out.append(u.strip())
+ continue
+ u = it.get("url")
+ if isinstance(u, str) and u.strip():
+ out.append(u.strip())
+ continue
+ iu = it.get("image_url")
+ if isinstance(iu, str) and iu.strip():
+ out.append(iu.strip())
+ elif isinstance(iu, dict):
+ u2 = iu.get("url")
+ if isinstance(u2, str) and u2.strip():
+ out.append(u2.strip())
+ return out[:cap]
+
+
+def materialize_legacy_response_output_attachments(
+ resp_images: Any,
+ *,
+ max_images: int = 3,
+) -> list[dict[str, Any]]:
+ """Persist remote/base64 model outputs as ``image_ref`` / ``image_url`` rows for chat UI."""
+ cap = max(1, min(int(max_images), 12))
+ produced: list[dict[str, Any]] = []
+ urls = normalize_legacy_output_image_urls(resp_images, max_items=cap)
+ if not urls:
+ return produced
+ store = AttachmentAssetStore()
+ for idx, item in enumerate(urls[:cap], start=1):
+ s = str(item or "").strip()
+ if not s:
+ continue
+ if s.startswith("data:") and ";base64," in s:
+ head, b64 = s.split(";base64,", 1)
+ mime = head.replace("data:", "", 1) or "image/png"
+ try:
+ blob = base64.b64decode(b64.encode("ascii"))
+ except Exception:
+ continue
+ meta = store.save_bytes(blob, filename=f"image-output-{idx}.png", mime=mime)
+ produced.append(
+ {
+ "type": "image_ref",
+ "attachment_id": meta.attachment_id,
+ "name": meta.name,
+ "mime": meta.mime,
+ "bytes": meta.bytes,
+ "width": meta.width,
+ "height": meta.height,
+ }
+ )
+ elif s.startswith("http://") or s.startswith("https://"):
+ try:
+ blob, ctype = download_http_url_bytes(s)
+ if blob:
+ mime = (ctype.split(";", 1)[0].strip() if ctype else "") or "image/png"
+ ext = ".png"
+ if mime == "image/jpeg":
+ ext = ".jpg"
+ elif mime == "image/webp":
+ ext = ".webp"
+ elif mime == "image/gif":
+ ext = ".gif"
+ meta = store.save_bytes(blob, filename=f"image-output-{idx}{ext}", mime=mime)
+ produced.append(
+ {
+ "type": "image_ref",
+ "attachment_id": meta.attachment_id,
+ "name": meta.name,
+ "mime": meta.mime,
+ "bytes": meta.bytes,
+ "width": meta.width,
+ "height": meta.height,
+ }
+ )
+ continue
+ except Exception:
+ pass
+ produced.append({"type": "image_url", "url": s, "name": f"image-output-{idx}.png"})
+ return produced
+
+
+def legacy_image_turn_bundle(resp: dict[str, Any]) -> tuple[bool, str, list[dict[str, Any]]]:
+ """Interpret ``send_legacy_image_messages`` result for persistence (text-only vision answers allowed)."""
+ ok = bool(resp.get("ok"))
+ text = str(resp.get("text") or "").strip()
+ if not ok:
+ err = str(resp.get("error") or "").strip()
+ return False, f"Image generation failed: {err or 'unknown error'}", []
+ raw_urls = normalize_legacy_output_image_urls(resp.get("images"), max_items=6)
+ imgs = materialize_legacy_response_output_attachments(raw_urls, max_images=3)
+ if not imgs and raw_urls:
+ imgs = [
+ {"type": "image_url", "url": u, "name": f"image-output-{i}.png"}
+ for i, u in enumerate(raw_urls[:3], start=1)
+ if u.startswith(("http://", "https://", "data:"))
+ ]
+ if imgs:
+ return True, text, imgs
+ if text:
+ return True, text, []
+ diag = resp.get("extract_diag")
+ hint = ""
+ if isinstance(diag, dict) and diag:
+ hint = format_extract_diag(diag, max_chars=1400)
+ red = resp.get("provider_response_redacted")
+ if isinstance(red, str) and red.strip():
+ hint = f"{hint}\nprovider_json={red.strip()}" if hint else f"provider_json={red.strip()}"
+ base = "Image specialist failed: empty response from provider."
+ out_msg = f"{base} {hint}".strip()
+ if len(out_msg) > 12_000:
+ out_msg = out_msg[:11_980] + "\n…"
+ return False, out_msg, []
+
+
+def legacy_image_assistant_body_with_placeholder(
+ *,
+ lang: str | None,
+ body_text: str,
+ produced: list[dict[str, Any]] | None,
+) -> str:
+ """If the model returned images but no visible text, use the standard chat placeholder (ZH/EN).
+
+ Shared by ``direct_loop`` (gateway /chat) and ``specialist_agent`` (temp sessions).
+ """
+ if str(body_text or "").strip():
+ return str(body_text or "")
+ if produced:
+ return (
+ "Generated image (see attachment below)."
+ if str(lang or "").startswith("en")
+ else "已生成图片（见下方附件）。"
+ )
+ return str(body_text or "")
+
+
+def _truthy_env(name: str) -> bool:
+ return str(os.getenv(name) or "").strip().lower() in ("1", "true", "yes", "on")
+
+
+def _redact_payload_for_stderr(obj: Any) -> Any:
+ if isinstance(obj, dict):
+ return {str(k): _redact_payload_for_stderr(v) for k, v in obj.items()}
+ if isinstance(obj, list):
+ return [_redact_payload_for_stderr(x) for x in obj]
+ if isinstance(obj, str):
+ s = obj
+ if ";base64," in s and s.strip().startswith("data:") and len(s) > 120:
+ h, _, _ = s.partition(";base64,")
+ return f"{h};base64,"
+ return s
+ return obj
+
+
+def _stderr_debug_image_legacy(url: str, payload: dict[str, Any]) -> None:
+ if not _truthy_env("AIA_IMAGE_EXPERT_DEBUG_PRINT_PAYLOAD"):
+ return
+ try:
+ txt = json.dumps(_redact_payload_for_stderr(dict(payload)), ensure_ascii=False, indent=2, default=str)
+ sys.stderr.write(f"\n[oclaw image_legacy] POST {url}\n{txt}\n\n")
+ sys.stderr.flush()
+ except Exception:
+ pass
+
+
+def _extra_request_fields_from_env() -> dict[str, Any]:
+ """Merge JSON from ``AIA_IMAGE_EXPERT_REQUEST_EXTRA`` (or legacy alias ``AIA_LEGACY_IMAGE_REQUEST_EXTRA``)."""
+ raw = (os.getenv("AIA_IMAGE_EXPERT_REQUEST_EXTRA") or os.getenv("AIA_LEGACY_IMAGE_REQUEST_EXTRA") or "").strip()
+ if not raw:
+ return {}
+ try:
+ parsed = json.loads(raw)
+ return parsed if isinstance(parsed, dict) else {}
+ except Exception:
+ return {}
+
+
+def _dashscope_image_env_kw() -> dict[str, Any]:
+ """Map `_local/system.env.example` ``DASHSCOPE_IMAGE_*`` vars to multimodal/top-level kwargs (parity with SDK samples)."""
+ out: dict[str, Any] = {}
+ raw_n = (os.getenv("DASHSCOPE_IMAGE_N") or "").strip()
+ if raw_n.isdigit():
+ out["n"] = max(1, min(int(raw_n), 6))
+
+ wm = (os.getenv("DASHSCOPE_IMAGE_WATERMARK") or "").strip().lower()
+ if wm in ("1", "true", "yes", "on"):
+ out["watermark"] = True
+ elif wm in ("0", "false", "no", "off"):
+ out["watermark"] = False
+
+ raw_stream = (os.getenv("DASHSCOPE_IMAGE_STREAM") or "").strip().lower()
+ if raw_stream in ("1", "true", "yes", "on"):
+ out["stream"] = True
+ elif raw_stream in ("0", "false", "no", "off"):
+ out["stream"] = False
+
+ neg = os.getenv("DASHSCOPE_IMAGE_NEGATIVE_PROMPT")
+ if neg is not None:
+ ns = str(neg)
+ if ns.strip():
+ out["negative_prompt"] = ns
+
+ pe = (os.getenv("DASHSCOPE_IMAGE_PROMPT_EXTEND") or "").strip().lower()
+ if pe in ("1", "true", "yes", "on"):
+ out["prompt_extend"] = True
+ elif pe in ("0", "false", "no", "off"):
+ out["prompt_extend"] = False
+
+ size = (os.getenv("DASHSCOPE_IMAGE_SIZE") or "").strip()
+ if size:
+ out["size"] = size
+
+ return out
+
+
+def _openai_compatible_vision_content(images: list[str], prompt_text: str) -> list[dict[str, Any]]:
+ """DashScope *compatible-mode* / OpenAI Chat Completions vision shape (NOT ``{"image":..., "text":...}``)."""
+ blocks: list[dict[str, Any]] = []
+ for img in images:
+ blocks.append({"type": "image_url", "image_url": {"url": img}})
+ blocks.append({"type": "text", "text": str(prompt_text or "").strip()})
+ return blocks
+
+
+def _model_triggers_dashscope_native_fallback(model_name: str) -> bool:
+ """``qwen-image`` on OpenAI-compat ``/chat/completions`` often returns ``message.content=null``."""
+ if str(os.getenv("AIA_IMAGE_EXPERT_DISABLE_DASHSCOPE_NATIVE_FALLBACK") or "").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+ "on",
+ ):
+ return False
+ if str(os.getenv("AIA_IMAGE_EXPERT_FORCE_DASHSCOPE_NATIVE") or "").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+ "on",
+ ):
+ return True
+ m = (model_name or "").strip().lower()
+ needle = (os.getenv("AIA_IMAGE_EXPERT_NATIVE_FALLBACK_MODEL_SUBSTR") or "qwen-image").strip().lower()
+ return bool(needle) and needle in m
+
def _http_content_blocks(images: list[str], prompt: str, *, typed: bool) -> list[dict[str, Any]]:
prompt_text = str(prompt or "").strip() or render_prompt("image/default_edit_prompt.zh.md", strict=True)
@@ -44,21 +389,21 @@ def send_legacy_image_messages(
api_key: str | None = None,
base_url: str | None = None,
) -> dict[str, Any]:
- """Legacy multimodal HTTP (non--OpenAI-``image_url`` schema). Optional; specialists use OCR client."""
- resolved_base_url = (base_url or env_ocr_lane_base_url()).strip()
- resolved_api_key = (api_key or env_ocr_lane_api_key()).strip()
- model_name = ((model or "").strip() or env_ocr_lane_model())
- endpoint = env_ocr_lane_chat_endpoint()
+ """Legacy multimodal HTTP (non--OpenAI-``image_url`` schema); image specialist uses this entry point."""
+ resolved_base_url = (base_url or env_image_expert_base_url()).strip()
+ resolved_api_key = (api_key or env_image_expert_api_key()).strip()
+ model_name = ((model or "").strip() or env_image_expert_model())
+ endpoint = env_image_expert_chat_endpoint()
url = join_url(resolved_base_url, endpoint)
if not resolved_api_key or not resolved_base_url:
return {
"ok": False,
- "error": "missing AIA_OCR_API_KEY or AIA_OCR_BASE_URL (or pass api_key and base_url)",
+ "error": "missing AIA_IMAGE_EXPERT_API_KEY or AIA_IMAGE_EXPERT_BASE_URL (or pass api_key and base_url)",
}
if not model_name:
return {
"ok": False,
- "error": "missing AIA_OCR_MODEL (or pass model=...) — no default model id",
+ "error": "missing model (chosen profile/model=… or set AIA_IMAGE_EXPERT_MODEL)",
}
if not images:
return {"ok": False, "error": "at least one image input is required"}
@@ -78,22 +423,28 @@ def send_legacy_image_messages(
if not selected:
return {"ok": False, "error": "no usable image input (expected URL or data URL)"}
- prefer_typed_http = "compatible-mode" in resolved_base_url.lower()
- content_multi = _http_content_blocks(selected, prompt, typed=prefer_typed_http)
- content_multi_fallback = _http_content_blocks(selected, prompt, typed=not prefer_typed_http)
+ # Compatible-mode expects OpenAI-style ``image_url`` + ``text`` parts; DashScope-native HTTP uses plain ``{"image"}`` blocks.
+ use_openai_blocks = "compatible-mode" in resolved_base_url.lower()
+ prompt_plain = str(prompt or "").strip() or render_prompt("image/default_edit_prompt.zh.md", strict=True)
+ if use_openai_blocks:
+ content_multi = _openai_compatible_vision_content(selected, prompt_plain)
+ else:
+ content_multi = _http_content_blocks(selected, prompt, typed=False)
headers = {
"Authorization": f"Bearer {resolved_api_key}",
"Content-Type": "application/json",
}
- payload_multi = {
+ extra_ds = _dashscope_image_env_kw()
+ extra_json = _extra_request_fields_from_env()
+ extra = {**extra_ds, **extra_json}
+ core_multi = {
"model": model_name,
"messages": [{"role": "user", "content": content_multi}],
}
- payload_single = {
- "model": model_name,
- "messages": [{"role": "user", "content": content_multi_fallback[-2:] if len(content_multi_fallback) >= 2 else content_multi_fallback}],
- }
+ payload_multi = {**extra, **core_multi}
+
+ _stderr_debug_image_legacy(url, payload_multi)
with httpx.Client(timeout=float(timeout_sec)) as client:
try:
@@ -102,45 +453,152 @@ def send_legacy_image_messages(
return {"ok": False, "error": f"http request failed: {type(e).__name__}: {e}", "backend_shape": "multi"}
if r.status_code >= 400:
- try:
- r2 = post_with_retry(client, url=url, headers=headers, payload=payload_single)
- except Exception as e:
- return {
- "ok": False,
- "error": f"http fallback request failed: {type(e).__name__}: {e}",
- "backend_shape": "single-fallback-failed",
- }
- if r2.status_code >= 400:
- return {
- "ok": False,
- "error": f"http {r2.status_code}: {r2.text[:500]}",
- "backend_shape": "single-fallback-failed",
- }
- try:
- obj2 = r2.json()
- except Exception:
- return {"ok": False, "error": f"non-json response: {r2.text[:500]}", "backend_shape": "single"}
- text, out_images = extract_text_and_images(obj2 if isinstance(obj2, dict) else {})
- return {
- "ok": True,
- "text": text,
- "images": out_images,
- "backend_shape": "single",
- "input_kind": input_kind[-1:] if input_kind else ["data_url"],
- }
+ return {"ok": False, "error": f"http {r.status_code}: {r.text[:800]}", "backend_shape": "multi"}
try:
obj = r.json()
except Exception:
return {"ok": False, "error": f"non-json response: {r.text[:500]}", "backend_shape": "multi"}
- text, out_images = extract_text_and_images(obj if isinstance(obj, dict) else {})
+ if not isinstance(obj, dict):
+ return {
+ "ok": False,
+ "error": f"expected JSON object from provider, got {type(obj).__name__}",
+ "backend_shape": "multi",
+ "input_kind": input_kind if input_kind else ["data_url"],
+ }
+ body = obj
+ ds_ok, ds_err = dashscope_multimodal_http_ok(body)
+ if not ds_ok:
+ return {
+ "ok": False,
+ "error": ds_err or "provider rejected request (DashScope code/message)",
+ "backend_shape": "multi",
+ "input_kind": input_kind if input_kind else ["data_url"],
+ }
+ text, out_images = extract_text_and_images(body)
+ compat_extract_diag: dict[str, Any] | None = None
+ native_fallback_meta: dict[str, Any] = {}
+
+ if (
+ not str(text or "").strip()
+ and not out_images
+ and use_openai_blocks
+ and _model_triggers_dashscope_native_fallback(model_name)
+ ):
+ compat_extract_diag = build_extract_diag_empty(body)
+ native_url = dashscope_native_multimodal_url_from_compatible_base(resolved_base_url)
+ if not native_url:
+ native_fallback_meta = {
+ "attempted": False,
+ "hint": "set AIA_IMAGE_EXPERT_DASHSCOPE_NATIVE_URL or use a compatible-mode base_url",
+ }
+ else:
+ content_native = _http_content_blocks(selected, prompt, typed=False)
+ param_merge = {**_dashscope_image_env_kw(), **_extra_request_fields_from_env()}
+ param_merge.pop("stream", None)
+ payload_native = {
+ "model": model_name,
+ "input": {"messages": [{"role": "user", "content": content_native}]},
+ "parameters": {"result_format": "message", **param_merge},
+ }
+ _stderr_debug_image_legacy(native_url, payload_native)
+ try:
+ r2 = post_with_retry(client, url=native_url, headers=headers, payload=payload_native)
+ except Exception as e:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "error": f"{type(e).__name__}: {e}",
+ }
+ else:
+ if r2.status_code >= 400:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "http_status": int(r2.status_code),
+ "body_head": r2.text[:600],
+ }
+ else:
+ try:
+ b2 = r2.json()
+ except Exception as e:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "error": f"json: {type(e).__name__}: {e}",
+ }
+ else:
+ if isinstance(b2, dict):
+ ok2, err2 = dashscope_multimodal_http_ok(b2)
+ if not ok2:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "dashscope_error": err2 or "business code",
+ }
+ else:
+ t2, im2 = extract_text_and_images(b2)
+ if str(t2 or "").strip() or im2:
+ text, out_images = t2, im2
+ body = b2
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "succeeded": True,
+ }
+ else:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "succeeded": False,
+ "native_diag": build_extract_diag_empty(b2),
+ }
+ else:
+ native_fallback_meta = {
+ "attempted": True,
+ "url": native_url,
+ "error": "native response not a JSON object",
+ }
+
+ extract_diag: dict[str, Any] | None = None
+ provider_response_redacted: str | None = None
+ if not str(text or "").strip() and not out_images:
+ extract_diag = build_extract_diag_empty(body)
+ if compat_extract_diag is not None:
+ extract_diag["openai_compat_empty"] = compat_extract_diag
+ if native_fallback_meta:
+ extract_diag["dashscope_native_fallback"] = native_fallback_meta
+ provider_response_redacted = redact_response_for_debug(body, max_chars=3200)
+ if _truthy_env("AIA_IMAGE_EXPERT_DEBUG_PRINT_PAYLOAD"):
+ try:
+ sys.stderr.write(
+ "\n[oclaw image_legacy] extract_empty extract_diag="
+ + format_extract_diag(extract_diag, max_chars=4000)
+ + "\n"
+ + (provider_response_redacted or "")
+ + "\n\n"
+ )
+ sys.stderr.flush()
+ except Exception:
+ pass
+ out_images = normalize_legacy_output_image_urls(out_images, max_items=12)
return {
"ok": True,
"text": text,
"images": out_images,
"backend_shape": "multi",
"input_kind": input_kind if input_kind else ["data_url"],
+ "extract_diag": extract_diag,
+ "provider_response_redacted": provider_response_redacted,
}
-__all__ = ["send_legacy_image_messages"]
+__all__ = [
+ "IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH",
+ "collect_legacy_lane_images_from_attachments",
+ "legacy_image_assistant_body_with_placeholder",
+ "legacy_image_turn_bundle",
+ "materialize_legacy_response_output_attachments",
+ "normalize_legacy_output_image_urls",
+ "send_legacy_image_messages",
+]
diff --git a/platform/llm/transports/openai_responses.py b/platform/llm/transports/openai_responses.py
index 479171dd..29046182 100644
--- a/platform/llm/transports/openai_responses.py
+++ b/platform/llm/transports/openai_responses.py
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
import logging
import os
+import sys
import uuid
from typing import Any, Optional
from collections.abc import Callable, Iterable
@@ -12,6 +13,140 @@ from oclaw.platform.llm.transports.base import ChatModel, LLMResponse, LLMToolCa
logger = logging.getLogger(__name__)
+def _redact_nested_json_preview(obj: Any, *, max_chars: int = 4000) -> str:
+ """Best-effort JSON preview for logs (truncate; redact mega data URLs)."""
+
+ try:
+ s = json.dumps(obj, ensure_ascii=False, default=str)
+ except Exception:
+ s = str(obj)
+
+ def _clip_data_urls(t: str) -> str:
+ out_parts: list[str] = []
+ i = 0
+ needle = "data:"
+ while i < len(t):
+ j = t.find(needle, i)
+ if j < 0:
+ out_parts.append(t[i:])
+ break
+ out_parts.append(t[i:j])
+ k = j + len(needle)
+ while k < len(t) and t[k] not in "?;, \n\r\t\"]":
+ k += 1
+ if k < len(t) and t[k:k + 8] == ";base64,":
+ end = k + 8
+ while end < len(t) and t[end] not in "\"}] \n\r\t":
+ end += 1
+ seg = end - (k + 8)
+ out_parts.append(f"data:")
+ i = end
+ continue
+ out_parts.append(t[j:k])
+ i = k
+ return "".join(out_parts)
+
+ s = _clip_data_urls(s)
+ if len(s) > max_chars:
+ return s[: max_chars - 40] + "\n...<>\n..."
+ return s
+
+
+def _deep_redact_for_debug(obj: Any) -> Any:
+ """Recursive copy for logs: shorten giant ``data:...;base64,...`` and very long strings."""
+
+ if isinstance(obj, dict):
+ return {str(k): _deep_redact_for_debug(v) for k, v in obj.items()}
+ if isinstance(obj, list):
+ return [_deep_redact_for_debug(x) for x in obj]
+ if isinstance(obj, str):
+ s = obj
+ if ";base64," in s and s.strip().startswith("data:") and len(s) > 160:
+ head, _, _tail = s.partition(";base64,")
+ return f"{head};base64,"
+ if len(s) > 16_000:
+ return s[:12_000] + f"..."
+ return s
+ return obj
+
+
+def _safe_int(raw: str | None, default: int, *, max_value: int = 2_000_000) -> int:
+ try:
+ value = int(str(raw or "").strip())
+ except Exception:
+ return default
+ if value < 1:
+ return default
+ return min(value, max_value)
+
+
+def _log_openai_responses_wire_kwargs(payload: dict[str, Any], *, variant: str, phase: str) -> None:
+ """Temporary diagnostics: log kwargs passed into ``OpenAI.responses.create`` (after our assembly).
+
+ Enable: ``AIA_RESPONSES_DEBUG_PRINT_PAYLOAD=1`` — emits ``logging.warning`` **and** the same JSON on **stderr**
+ (gateway控制台默认可见，不依赖 logging level)。
+
+ Notes:
+
+ - OpenAI SDK may still apply minor JSON transforms on send; this matches **our** arguments.
+ - Base64 / long ``data:`` URLs are redacted to keep logs readable.
+ """
+ raw = str(os.getenv("AIA_RESPONSES_DEBUG_PRINT_PAYLOAD") or "").strip().lower()
+ if raw not in ("1", "true", "yes", "on"):
+ return
+ try:
+ dbg = _deep_redact_for_debug(dict(payload))
+ txt = json.dumps(dbg, ensure_ascii=False, indent=2, default=str)
+ cap = _safe_int(os.getenv("AIA_RESPONSES_DEBUG_PRINT_MAX_CHARS"), 120_000, max_value=500_000)
+ if len(txt) > cap:
+ txt = txt[: max(cap - 80, 0)] + "\n...<>\n"
+ logger.warning(
+ "openai_responses DEBUG wire kwargs [%s] variant=%s (%d chars):\n%s",
+ phase,
+ variant,
+ len(txt),
+ txt,
+ )
+ try:
+ sys.stderr.write(
+ f"\n[oclaw openai_responses DEBUG] phase={phase} variant={variant} chars={len(txt)}\n{txt}\n\n"
+ )
+ sys.stderr.flush()
+ except Exception:
+ pass
+ except Exception as exc:
+ logger.warning("openai_responses DEBUG payload serialization failed: %s", exc)
+
+
+def _openai_sdk_diagnostic_text(exc: BaseException) -> str:
+ """``str(APIStatusError)`` is often only ``Error code: 400``; validation detail lives in ``body['message']``."""
+ chunks: list[str] = [str(exc)]
+ msg = getattr(exc, "message", None)
+ if isinstance(msg, str) and msg.strip() and msg not in chunks:
+ chunks.append(msg)
+ bod = getattr(exc, "body", None)
+ if isinstance(bod, dict):
+ bm = bod.get("message")
+ if isinstance(bm, str) and bm.strip():
+ chunks.append(bm)
+ err = bod.get("error")
+ if isinstance(err, dict):
+ em = err.get("message")
+ if isinstance(em, str) and em.strip():
+ chunks.append(em)
+ return "\n".join(chunks)
+
+
+def _is_input_messages_validation_error(exc: BaseException) -> bool:
+ """Third-party gateways often return 400/422 with Pydantic paths like ``input.messages.0.role``."""
+ m = _openai_sdk_diagnostic_text(exc).lower()
+ if "input.messages" in m:
+ return True
+ if "input should be 'user'" in m and "content" in m:
+ return True
+ return False
+
+
def _as_dict(obj: Any) -> dict[str, Any] | None:
if obj is None:
return None
@@ -131,14 +266,56 @@ class OpenAIResponsesModel(ChatModel):
self._client = OpenAI(**kw)
@staticmethod
- def _normalize_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ def _strip_leading_system_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
+ """Separate leading ``role=system`` rows for ``instructions=` (Responses API expects user-shaped ``input.messages``)."""
+ msgs = messages or []
+ sys_chunks: list[str] = []
+ i = 0
+ while i < len(msgs):
+ m = msgs[i]
+ if not isinstance(m, dict):
+ break
+ if str(m.get("role") or "").strip().lower() != "system":
+ break
+ c = m.get("content")
+ if isinstance(c, str) and c.strip():
+ sys_chunks.append(c.strip())
+ elif isinstance(c, list):
+ texts: list[str] = []
+ for it in c:
+ if not isinstance(it, dict):
+ continue
+ tt = str(it.get("type") or "").strip().lower()
+ if tt in ("text", "input_text"):
+ xs = str(it.get("text") or "").strip()
+ if xs:
+ texts.append(xs)
+ elif isinstance(it.get("text"), str) and str(it.get("text")).strip():
+ texts.append(str(it.get("text")).strip())
+ if texts:
+ sys_chunks.append("\n".join(texts))
+ i += 1
+ joined = "\n\n".join(sys_chunks).strip()
+ return joined, list(msgs[i:])
+
+ @staticmethod
+ def _normalize_messages(
+ messages: list[dict[str, Any]],
+ *,
+ envelope_openai_message: bool = True,
+ content_chat_completions_parts: bool = False,
+ image_detail_auto: bool = True,
+ ) -> list[dict[str, Any]]:
"""
- Normalize to the strictest OpenAI-compatible `responses` shape:
- - input.messages[*].role MUST be "user"
- - input.messages[*].content MUST be a list (content blocks)
- - text blocks use {"type":"text","text":...}
- - image blocks use {"type":"image_url","image_url":{"url":"..."}} or {"type":"input_image","image_url":"..."} depending on gateway;
- we prefer the common "image_url" block here.
+ Build Response input rows used under ``input`` (flat array) or ``input.messages``.
+
+ - **Responses content** (default when ``content_chat_completions_parts=False``): each part is
+ ``{"type":"input_text",...}`` / ``{"type":"input_image","image_url":"","detail":"auto"}``
+ (`OpenAI response_input_*` TypedDict surface).
+ - **Chat completions content** (``content_chat_completions_parts=True``): ``text`` /
+ ``image_url`` multimodal blocks.
+ - **Envelope** ``envelope_openai_message=True``: ``{"type":"message","role","content":[...]}``;
+ ``False``: ``{"role","content":[...]}`` (per some Bailian curls).
"""
out: list[dict[str, Any]] = []
for m in messages or []:
@@ -147,6 +324,22 @@ class OpenAIResponsesModel(ChatModel):
role = str(m.get("role") or "user").strip().lower() or "user"
content = m.get("content")
norm_content: list[dict[str, Any]] = []
+
+ def _append_image_part(url_value: str) -> None:
+ if content_chat_completions_parts:
+ norm_content.append({"type": "image_url", "image_url": {"url": url_value}})
+ else:
+ p: dict[str, Any] = {"type": "input_image", "image_url": url_value}
+ if image_detail_auto:
+ p["detail"] = "auto"
+ norm_content.append(p)
+
+ def _append_text_part(txt: str) -> None:
+ if content_chat_completions_parts:
+ norm_content.append({"type": "text", "text": txt})
+ else:
+ norm_content.append({"type": "input_text", "text": txt})
+
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "input_image":
@@ -154,39 +347,118 @@ class OpenAIResponsesModel(ChatModel):
b64 = normalize_image_b64_payload(item.get("image_base64") or item.get("data"))
if not b64:
continue
- norm_content.append({"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}})
+ _append_image_part(f"data:{mime};base64,{b64}")
+ continue
+ if isinstance(item, dict) and str(item.get("type") or "").strip().lower() == "image_url":
+ iu = item.get("image_url")
+ url = ""
+ if isinstance(iu, dict):
+ url = str(iu.get("url") or "").strip()
+ elif isinstance(iu, str):
+ url = iu.strip()
+ if url.startswith(("http://", "https://", "data:")):
+ _append_image_part(url)
continue
if isinstance(item, dict) and item.get("type") in ("input_text", "text"):
t = str(item.get("text") or "").strip()
if t:
- norm_content.append({"type": "text", "text": t})
+ _append_text_part(t)
continue
if isinstance(item, dict):
- # Best-effort coerce unknown blocks into valid text blocks.
+ itype = str(item.get("type") or "").strip().lower()
+ img_key = item.get("image")
+ if isinstance(img_key, str) and img_key.strip() and itype != "input_image":
+ uu = img_key.strip()
+ if uu.startswith(("http://", "https://", "data:")):
+ _append_image_part(uu)
+ continue
+ if isinstance(item.get("text"), str) and str(item.get("text") or "").strip():
+ _append_text_part(str(item.get("text") or "").strip())
+ continue
s = str(item.get("text") or "").strip()
if s:
- norm_content.append({"type": "text", "text": s})
+ _append_text_part(s)
continue
s = str(item or "").strip()
if s:
- norm_content.append({"type": "text", "text": s})
+ _append_text_part(s)
elif isinstance(content, str):
txt = content.strip()
if txt:
- norm_content.append({"type": "text", "text": txt})
+ _append_text_part(txt)
elif content is not None:
s = str(content).strip()
if s:
- norm_content.append({"type": "text", "text": s})
+ _append_text_part(s)
+
if not norm_content:
- # Keep shape valid even for empty messages.
continue
- # Enforce role=user and preserve other roles via prefix tag.
+ coerced_role = "user"
if role != "user":
prefix = "assistant" if role == "assistant" else ("system" if role == "system" else role)
- norm_content.insert(0, {"type": "text", "text": f"[{prefix}]"})
- out.append({"role": "user", "content": norm_content})
+ norm_content.insert(
+ 0,
+ OpenAIResponsesModel._role_prefix_part(prefix, content_chat_completions_parts),
+ )
+
+ if envelope_openai_message:
+ out.append({"type": "message", "role": coerced_role, "content": norm_content})
+ else:
+ out.append({"role": coerced_role, "content": norm_content})
+ return out
+
+ @staticmethod
+ def _role_prefix_part(prefix: str, chat_parts: bool) -> dict[str, Any]:
+ t = f"[{prefix}]"
+ if chat_parts:
+ return {"type": "text", "text": t}
+ return {"type": "input_text", "text": t}
+
+ @staticmethod
+ def _responses_input_candidates(
+ msgs: list[dict[str, Any]],
+ *,
+ flat_responses: bool,
+ prefer_envelope: bool,
+ prefer_chat_parts: bool,
+ ) -> list[tuple[str, Any]]:
+ """Several gateways validate ``input.messages`` differently; try a small deterministic set."""
+
+ def N(env: bool, chat: bool) -> list[dict[str, Any]]:
+ return OpenAIResponsesModel._normalize_messages(
+ msgs,
+ envelope_openai_message=env,
+ content_chat_completions_parts=chat,
+ )
+
+ keys_seen: set[str] = set()
+ out: list[tuple[str, Any]] = []
+
+ def push(tag: str, inp: Any) -> None:
+ try:
+ sk = json.dumps(inp, ensure_ascii=False, sort_keys=True, default=str)
+ except Exception:
+ sk = repr(inp)
+ if sk in keys_seen:
+ return
+ keys_seen.add(sk)
+ out.append((tag, inp))
+
+ if flat_responses:
+ for env in (prefer_envelope, not prefer_envelope):
+ push(f"flat_envelope_{env}", N(env, False))
+ return out
+
+ # Prefer profile/env defaults first (``AIA_RESPONSES_NESTED_CHAT_PARTS`` / envelope toggles), then a fixed
+ # fallback ring so picky gateways still get Chat-shaped ``input.messages`` without dropping multimodal pixels.
+ primary_combo = (prefer_envelope, prefer_chat_parts)
+ fallback_ring = [(False, True), (False, False), (True, False), (True, True)]
+ combos = [primary_combo] + [p for p in fallback_ring if p != primary_combo]
+ for env, chat in combos:
+ norm = N(env, chat)
+ push(f"e{int(env)}c{int(chat)}_messages", {"messages": norm})
+ push(f"e{int(env)}c{int(chat)}_flat_input", norm)
return out
def chat(
@@ -196,8 +468,96 @@ class OpenAIResponsesModel(ChatModel):
*,
on_token: Optional[Callable[[str], None]] = None,
) -> LLMResponse:
- norm = self._normalize_messages(messages)
- # OpenAI-compatible gateways differ: some require input={"messages":[...]} with role=user only.
+ sys_instructions, rest = OpenAIResponsesModel._strip_leading_system_messages(messages)
+ def _env_truthy(name: str) -> bool:
+ return str(os.getenv(name) or "").strip().lower() in ("1", "true", "yes", "on")
+
+ if _env_truthy("AIA_RESPONSES_DEBUG_PRINT_PAYLOAD"):
+ try:
+ sys.stderr.write(
+ "[oclaw openai_responses] chat() ENTERED "
+ f"model={self.model!r} base_url={self.base_url!r} "
+ f"messages={len(messages)} tools={len(tools or [])}\n"
+ )
+ sys.stderr.flush()
+ except Exception:
+ pass
+
+ # Default: nested ``input.messages`` rows use **Responses** parts (``input_text`` / ``input_image``).
+ # Flat Bailian ``input`` array: ``AIA_RESPONSES_INPUT_IS_FLAT_LIST=1`` (optional ``AIA_RESPONSES_FLAT_OPENAI_ENVELOPE``).
+ flat_raw = os.getenv("AIA_RESPONSES_INPUT_IS_FLAT_LIST")
+ if flat_raw is not None and str(flat_raw).strip():
+ flat_responses = _env_truthy("AIA_RESPONSES_INPUT_IS_FLAT_LIST")
+ else:
+ legacy = os.getenv("AIA_RESPONSES_INPUT_USE_MESSAGES_ARRAY")
+ if legacy is None or not str(legacy).strip():
+ flat_responses = False
+ else:
+ flat_responses = not _env_truthy("AIA_RESPONSES_INPUT_USE_MESSAGES_ARRAY")
+
+ # Nested ``input.messages`` defaults to **Responses** content parts (``input_*``) + ``type:message``,
+ # matching OpenAI ``EasyInputMessageParam`` / Model Studio expanded examples. Some proxies wrongly
+ # expect Chat ``text``/``image_url`` parts → ``AIA_RESPONSES_NESTED_CHAT_PARTS=1``.
+ if flat_responses:
+ envelope_openai_message = _env_truthy("AIA_RESPONSES_FLAT_OPENAI_ENVELOPE")
+ nested_chat_parts = False
+ else:
+ envelope_openai_message = True
+ nested_chat_parts = _env_truthy("AIA_RESPONSES_NESTED_CHAT_PARTS")
+
+ def _normalize_batch(msgs: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ return OpenAIResponsesModel._normalize_messages(
+ msgs,
+ envelope_openai_message=envelope_openai_message,
+ content_chat_completions_parts=nested_chat_parts,
+ )
+
+ msgs_for_norm: list[dict[str, Any]]
+ if rest:
+ msgs_for_norm = rest
+ norm = _normalize_batch(rest)
+ inst_kw: dict[str, Any] = {}
+ if sys_instructions.strip():
+ si = sys_instructions.strip()
+ if len(si) > 80_000:
+ si = si[:80_000] + "\n...[truncated]"
+ inst_kw = {"instructions": si}
+ if not norm:
+ msgs_for_norm = messages
+ norm = _normalize_batch(messages)
+ inst_kw = {}
+ else:
+ msgs_for_norm = messages
+ norm = _normalize_batch(messages)
+ inst_kw = {}
+
+ explicit_variants = os.getenv("AIA_RESPONSES_INPUT_VARIANTS")
+ bu_norm = (self.base_url or "").strip().lower()
+ # CRITICAL: do **not** treat empty ``base_url`` as "official" here. Clients often omit storing the
+ # default URL in profiles while still hitting third-party gateways via env/SDK defaults—but then we
+ # would wrongly disable alternate ``input`` shapes and only ever send ``primary``.
+ host_is_explicit_official_api = bool(bu_norm) and ("api.openai.com" in bu_norm)
+ if explicit_variants is not None and str(explicit_variants).strip():
+ multi_shape = _env_truthy("AIA_RESPONSES_INPUT_VARIANTS")
+ else:
+ multi_shape = not host_is_explicit_official_api
+
+ if flat_responses:
+ primary_input: Any = norm
+ else:
+ primary_input = {"messages": norm}
+
+ input_candidates: list[tuple[str, Any]]
+ if multi_shape and not _env_truthy("AIA_RESPONSES_DISABLE_INPUT_VARIANTS"):
+ input_candidates = OpenAIResponsesModel._responses_input_candidates(
+ msgs_for_norm,
+ flat_responses=flat_responses,
+ prefer_envelope=envelope_openai_message,
+ prefer_chat_parts=nested_chat_parts,
+ )
+ else:
+ input_candidates = [("primary", primary_input)]
+
stream_errors: list[str] = []
b = str(self.base_url or "").strip().lower()
force_disable = str(os.getenv("AIA_LLM_THINKING_FORCE_DISABLED") or "").strip().lower() in ("1", "true", "yes", "on")
@@ -217,26 +577,33 @@ class OpenAIResponsesModel(ChatModel):
reasoning_effort = str(getattr(self, "reasoning_effort", "") or "").strip().lower()
if reasoning_effort not in ("low", "medium", "high"):
reasoning_effort = ""
- stream_variants: list[dict[str, Any]] = [
- {
- **thinking,
- **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
- "model": self.model,
- "input": {"messages": norm},
- "tools": tools or None,
- "stream": True,
- },
- {
- **thinking,
- **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
- "model": self.model,
- "input": norm,
- "tools": tools or None,
- "stream": True,
- },
- ]
+ base_stream_template: dict[str, Any] = {
+ **thinking,
+ **inst_kw,
+ **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
+ "model": self.model,
+ "tools": tools or None,
+ "stream": True,
+ }
try:
- for payload in stream_variants:
+ for cand_label, responses_input in input_candidates:
+ stream_payload = {**base_stream_template, "input": responses_input}
+ if _env_truthy("AIA_RESPONSES_LOG_PAYLOAD_SUMMARY"):
+ logger.warning(
+ "openai_responses [%s] payload summary (redacted): %s",
+ cand_label,
+ _redact_nested_json_preview(
+ {
+ "model": self.model,
+ "input": responses_input,
+ "has_instructions": bool(inst_kw.get("instructions")),
+ "tools_n": len(tools or []),
+ "stream": True,
+ }
+ ),
+ )
+ payload = stream_payload
+ _log_openai_responses_wire_kwargs(payload, variant=str(cand_label), phase="stream")
try:
stream = self._client.responses.create(**payload)
text, tool_calls, final_resp = parse_openai_responses_stream_events(stream, on_token=on_token)
@@ -246,6 +613,8 @@ class OpenAIResponsesModel(ChatModel):
text = ot
if on_token:
on_token(text)
+ if cand_label != input_candidates[0][0]:
+ logger.info("openai_responses: succeeded with input variant %s", cand_label)
return LLMResponse(content=text, tool_calls=tool_calls)
except Exception as exc:
emsg = str(exc)
@@ -258,6 +627,11 @@ class OpenAIResponsesModel(ChatModel):
eb["thinking"] = {"type": "disabled"}
forced["extra_body"] = eb
forced.pop("reasoning_effort", None)
+ _log_openai_responses_wire_kwargs(
+ forced,
+ variant=f"{cand_label}_thinking_disabled",
+ phase="stream_retry_thinking",
+ )
stream = self._client.responses.create(**forced)
text, tool_calls, final_resp = parse_openai_responses_stream_events(stream, on_token=on_token)
if (not text.strip()) and final_resp:
@@ -266,31 +640,36 @@ class OpenAIResponsesModel(ChatModel):
text = ot
if on_token:
on_token(text)
+ if cand_label != input_candidates[0][0]:
+ logger.info("openai_responses: succeeded with input variant %s", cand_label)
return LLMResponse(content=text, tool_calls=tool_calls)
except Exception:
pass
+ if _env_truthy("AIA_RESPONSES_LOG_API_ERROR_DETAIL"):
+ eb = getattr(exc, "body", None)
+ if eb is not None:
+ logger.warning(
+ "responses.create stream [%s] body: %s", cand_label, eb
+ )
+ if len(input_candidates) > 1 and _is_input_messages_validation_error(exc):
+ stream_errors.append(f"{cand_label}:{emsg}")
+ continue
stream_errors.append(emsg)
- continue
+ break
raise RuntimeError("; ".join(stream_errors) or "responses_stream_all_variants_failed")
except Exception as exc:
logger.info("responses stream failed; fallback to non-stream (%s)", exc)
nonstream_errors: list[str] = []
- for payload in (
- {
- **thinking,
- **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
- "model": self.model,
- "input": {"messages": norm},
- "tools": tools or None,
- },
- {
- **thinking,
- **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
- "model": self.model,
- "input": norm,
- "tools": tools or None,
- },
- ):
+ base_nonstream = {
+ **thinking,
+ **inst_kw,
+ **({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
+ "model": self.model,
+ "tools": tools or None,
+ }
+ for cand_label, responses_input in input_candidates:
+ payload = {**base_nonstream, "input": responses_input}
+ _log_openai_responses_wire_kwargs(payload, variant=str(cand_label), phase="non_stream")
try:
resp = self._client.responses.create(**payload)
d = _as_dict(resp) or {}
@@ -298,6 +677,8 @@ class OpenAIResponsesModel(ChatModel):
tool_calls = _collect_tool_calls_from_response_dict(d)
if on_token and text:
on_token(text)
+ if cand_label != input_candidates[0][0]:
+ logger.info("openai_responses non-stream: succeeded with input variant %s", cand_label)
return LLMResponse(content=text, tool_calls=tool_calls)
except Exception as e2:
emsg2 = str(e2)
@@ -309,17 +690,36 @@ class OpenAIResponsesModel(ChatModel):
eb["thinking"] = {"type": "disabled"}
forced["extra_body"] = eb
forced.pop("reasoning_effort", None)
+ _log_openai_responses_wire_kwargs(
+ forced,
+ variant=f"{cand_label}_thinking_disabled",
+ phase="non_stream_retry_thinking",
+ )
resp = self._client.responses.create(**forced)
d = _as_dict(resp) or {}
text = str(d.get("output_text") or "")
tool_calls = _collect_tool_calls_from_response_dict(d)
if on_token and text:
on_token(text)
+ if cand_label != input_candidates[0][0]:
+ logger.info(
+ "openai_responses non-stream: succeeded with input variant %s",
+ cand_label,
+ )
return LLMResponse(content=text, tool_calls=tool_calls)
except Exception:
pass
+ if _env_truthy("AIA_RESPONSES_LOG_API_ERROR_DETAIL"):
+ eb = getattr(e2, "body", None)
+ if eb is not None:
+ logger.warning(
+ "responses.create non-stream [%s] body: %s", cand_label, eb
+ )
+ if len(input_candidates) > 1 and _is_input_messages_validation_error(e2):
+ nonstream_errors.append(f"{cand_label}:{emsg2}")
+ continue
nonstream_errors.append(emsg2)
- continue
+ break
raise RuntimeError(
"openai_responses_request_failed: "
+ " | ".join([str(exc)] + nonstream_errors[-2:])
diff --git a/platform/persistence/sqlite_store.py b/platform/persistence/sqlite_store.py
index 270d5c25..6f49ece8 100644
--- a/platform/persistence/sqlite_store.py
+++ b/platform/persistence/sqlite_store.py
@@ -3,6 +3,7 @@ from __future__ import annotations
import base64
import json
import os
+import re
import sqlite3
import sys
import hashlib
@@ -1436,6 +1437,78 @@ class SqliteStore:
(aid, tid, uid, sid, src, ts),
)
+ @staticmethod
+ def _attachment_ids_for_acl_from_payload_items(items: list[dict[str, Any]]) -> list[str]:
+ """Collect stable attachment ids from message attachment JSON (for ACL rows)."""
+ out: list[str] = []
+ seen: set[str] = set()
+ ref_types = {"image_ref", "video_ref", "text_ref", "binary_ref"}
+ relay_re = re.compile(r"^relay://attachments/[^/]+/([a-f0-9]{8,64})$", re.IGNORECASE)
+ for a in items:
+ if not isinstance(a, dict):
+ continue
+ typ = str(a.get("type") or "").strip().lower()
+ aid = str(a.get("attachment_id") or a.get("attachmentId") or "").strip().lower()
+ if typ in ref_types and aid:
+ if aid not in seen:
+ seen.add(aid)
+ out.append(aid)
+ continue
+ if typ != "relay_pointer":
+ continue
+ if not aid:
+ uri = str(a.get("pointer_uri") or "").strip()
+ m = relay_re.match(uri)
+ if m:
+ aid = str(m.group(1) or "").strip().lower()
+ if aid and aid not in seen:
+ seen.add(aid)
+ out.append(aid)
+ return out
+
+ def sync_attachment_acl_from_chat_message_attachments(
+ self,
+ *,
+ session_id: str,
+ role: str,
+ event_type: str | None,
+ attachments: Any,
+ ) -> None:
+ """Best-effort: link chat_message attachments to session owner so strict ACL downloads work.
+
+ Tool results already link in runtime; assistant rows (image specialist, inbound, etc.)
+ historically did not, which breaks ``AIA_ATTACHMENT_ACL_STRICT=1``.
+ """
+ sid = str(session_id or "").strip()
+ if not sid or attachments is None:
+ return
+ owner = self.get_ui_session_owner(session_id=sid) or {}
+ tid = str(owner.get("tenant_id") or "").strip()
+ uid = str(owner.get("user_id") or "").strip()
+ if not tid or not uid:
+ return
+ items: list[dict[str, Any]] = []
+ if isinstance(attachments, list):
+ items = [x for x in attachments if isinstance(x, dict)]
+ elif isinstance(attachments, dict):
+ items = [attachments]
+ else:
+ return
+ ids = self._attachment_ids_for_acl_from_payload_items(items)
+ if not ids:
+ return
+ r = str(role or "").strip() or "-"
+ ev = str(event_type or "").strip() or "-"
+ src = f"chat_message:{r}:{ev}"[:240]
+ for aid in ids:
+ self.link_attachment_acl(
+ tenant_id=tid,
+ user_id=uid,
+ session_id=sid,
+ attachment_id=aid,
+ source=src,
+ )
+
def attachment_acl_allows_user(self, *, tenant_id: str, user_id: str, attachment_id: str) -> bool:
tid = str(tenant_id or "").strip()
uid = str(user_id or "").strip()
@@ -1922,6 +1995,15 @@ class SqliteStore:
"UPDATE chat_session SET last_message_at = ? WHERE id = ?",
(ts, session_id),
)
+ try:
+ self.sync_attachment_acl_from_chat_message_attachments(
+ session_id=str(session_id or "").strip(),
+ role=str(role or ""),
+ event_type=str(event_type or "").strip() or None,
+ attachments=attachments,
+ )
+ except Exception:
+ pass
return ChatMessage(
id=msg_id,
session_id=session_id,
diff --git a/runtime/agent_core_attempt.py b/runtime/agent_core_attempt.py
index 92379768..fa30c897 100644
--- a/runtime/agent_core_attempt.py
+++ b/runtime/agent_core_attempt.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import json
from dataclasses import dataclass
from typing import Any, Callable, Optional
@@ -74,36 +75,67 @@ class AttemptRunnerOutput:
outcome: TurnRunOutcome
+def _exception_reason_for_chat(exc: BaseException, *, max_chars: int = 12000) -> str:
+ """Human-visible attempt failure text (WS/chat reads ``AttemptState.reason``).
+
+ OpenAI Python SDK's ``APIError`` often carries a decoded JSON ``body``; include it so oclaw /chat
+ can show the full upstream validation payload instead of only ``str(exc)``.
+ """
+ head = f"{type(exc).__name__}:{exc}"
+ blob = ""
+ cur: BaseException | None = exc
+ seen_ids: set[int] = set()
+ depth = 0
+ while cur is not None and depth < 16:
+ if id(cur) in seen_ids:
+ break
+ seen_ids.add(id(cur))
+ depth += 1
+ body = getattr(cur, "body", None)
+ if body is not None:
+ try:
+ blob = json.dumps(body, ensure_ascii=False, indent=2) if isinstance(body, (dict, list)) else str(body)
+ except Exception:
+ blob = str(body)
+ break
+ cur = cur.__cause__
+ out = head if not blob else f"{head}\n\nupstream_json_body:\n{blob}"
+ if len(out) > max_chars:
+ out = out[: max_chars - 24] + "\n...[truncated]"
+ return out
+
+
def _classify_attempt_error(exc: Exception) -> tuple[str, str, bool]:
raw = f"{type(exc).__name__}:{exc}"
low = raw.lower()
+ reason = _exception_reason_for_chat(exc)
if "relay_envelope_unsupported_version" in low:
- return ("relay_envelope_unsupported_version", raw[:500], False)
+ return ("relay_envelope_unsupported_version", reason, False)
if "relay_envelope_invalid" in low:
- return ("relay_envelope_invalid", raw[:500], False)
+ return ("relay_envelope_invalid", reason, False)
if "interrupted" in low or "cancel" in low or "stopped" in low:
- return ("control_interrupted", raw[:500], False)
+ return ("control_interrupted", reason, False)
if "api_key" in low or "invalid api key" in low or "unauthorized" in low or "401" in low or "forbidden" in low or "403" in low:
- return ("auth_invalid_credentials", raw[:500], False)
+ return ("auth_invalid_credentials", reason, False)
if "invalid tool_result sequence" in low or "unexpected tool_use_id" in low:
- return ("tool_replay_protocol_mismatch", raw[:500], True)
+ return ("tool_replay_protocol_mismatch", reason, True)
if "invalid_request" in low or "bad_request" in low or "400" in low:
- return ("input_invalid_request", raw[:500], False)
+ return ("input_invalid_request", reason, False)
if "context_length" in low or "token limit" in low or "max context" in low:
- return ("context_overflow", raw[:500], True)
+ return ("context_overflow", reason, True)
if "tool_loop_guard" in low:
- return ("tool_loop_guard", raw[:500], False)
+ return ("tool_loop_guard", reason, False)
if "tool_timeout_or_failed" in low or "tool_execution_error" in low:
- return ("tool_execution_failed", raw[:500], True)
+ return ("tool_execution_failed", reason, True)
if "timeout" in low:
- return ("provider_timeout", raw[:500], True)
+ return ("provider_timeout", reason, True)
if "rate" in low and "limit" in low:
- return ("provider_rate_limited", raw[:500], True)
+ return ("provider_rate_limited", reason, True)
if "temporary" in low or "temporar" in low:
- return ("provider_temporary_error", raw[:500], True)
+ return ("provider_temporary_error", reason, True)
if "connection" in low or "network" in low or "503" in low or "502" in low:
- return ("provider_unavailable", raw[:500], True)
- return ("runtime_unknown_error", raw[:500], False)
+ return ("provider_unavailable", reason, True)
+ return ("runtime_unknown_error", reason, False)
def run_attempt(*, store: Any, data: AttemptRunnerInput) -> AttemptRunnerOutput:
diff --git a/runtime/agents/specialist_agent.py b/runtime/agents/specialist_agent.py
index c9b95aff..e658fae1 100644
--- a/runtime/agents/specialist_agent.py
+++ b/runtime/agents/specialist_agent.py
@@ -2,10 +2,9 @@ from __future__ import annotations
import json
import os
+import sys
import time
-import base64
import hashlib
-import httpx
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Optional
@@ -14,8 +13,13 @@ from oclaw.runtime.chat.agent import Agent
from oclaw.runtime.chat.agent import GenerationInterrupted
from oclaw.runtime.agents.network_ops_agent import NetworkOpsAgent
from oclaw.platform.persistence.sqlite_store import SqliteStore
-from oclaw.platform.files.attachment_assets import AttachmentAssetStore, attachment_id_to_data_url
-from oclaw.platform.llm.image_ocr_client import send_ocr_image_messages
+from oclaw.platform.llm.image_legacy_client import (
+ IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH,
+ collect_legacy_lane_images_from_attachments,
+ legacy_image_assistant_body_with_placeholder,
+ legacy_image_turn_bundle,
+ send_legacy_image_messages,
+)
from oclaw.runtime.tools import default_registry
from oclaw.runtime.agents.specialists import expert_name_for_specialist
@@ -233,56 +237,44 @@ class SpecialistAgentRunner:
try:
if step.specialist == "image":
image_protocol = "messages.content.image"
- selected_images: list[str] = []
- for att in parent_task.attachments or []:
- if not isinstance(att, dict):
- continue
- t = str(att.get("type") or "").strip().lower()
- if t == "image_ref":
- aid = str(att.get("attachment_id") or "").strip()
- if not aid:
- continue
- data_url = attachment_id_to_data_url(aid, mime=str(att.get("mime") or ""))
- if data_url:
- selected_images.append(data_url)
- elif t in ("input_image", "image"):
- raw = str(att.get("image_base64") or att.get("data") or "").strip()
- if raw:
- mime = str(att.get("mime") or "image/jpeg")
- if raw.startswith("data:"):
- selected_images.append(raw)
- else:
- selected_images.append(f"data:{mime};base64,{raw}")
- elif t == "image_url":
- u = str(att.get("url") or "").strip()
- if u:
- selected_images.append(u)
- if len(selected_images) >= 3:
- break
+ selected_images = collect_legacy_lane_images_from_attachments(
+ list(parent_task.attachments or []),
+ max_images=3,
+ )
image_input_count = len(selected_images)
image_input_kind = ["data_url" if s.startswith("data:") else "url" for s in selected_images]
if not selected_images:
output = "Image specialist received no image input."
ok = False
else:
+ # Use the user's chosen model/session profile (same as specialist routing UI). Wrong model ⇒ upstream HTTP error as-is (no OCR lane, no alternate payload).
_, chosen_model, _ = self._resolve_profile_and_model(step.specialist)
- model_name = str(
- os.getenv("AIA_OCR_MODEL") or getattr(chosen_model, "model", None) or ""
- ).strip() or None
- api_key = str(getattr(chosen_model, "api_key", "") or "").strip() or None
- base_url = str(getattr(chosen_model, "base_url", "") or "").strip() or None
text_parts = [
str(x).strip()
for x in (step.objective, step.input_text, parent_task.user_text)
if str(x or "").strip()
]
- user_text = "\n".join(text_parts) if text_parts else "请根据用户上传的图片作答：描述可见场景、物体与文字；不确定处请标明。"
- resp = send_ocr_image_messages(
+ user_text = "\n".join(text_parts) if text_parts else IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH
+ if str(os.getenv("AIA_IMAGE_EXPERT_DEBUG_PRINT_PAYLOAD") or "").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+ "on",
+ ):
+ try:
+ sys.stderr.write(
+ "[oclaw specialist:image] lane=legacy_http → send_legacy_image_messages "
+ "(NOT OpenAIResponsesModel).\n"
+ )
+ sys.stderr.flush()
+ except Exception:
+ pass
+ resp = send_legacy_image_messages(
images=selected_images,
prompt=user_text,
- model=model_name,
- api_key=api_key,
- base_url=base_url,
+ model=str(getattr(chosen_model, "model", "") or "").strip() or None,
+ api_key=str(getattr(chosen_model, "api_key", "") or "").strip() or None,
+ base_url=str(getattr(chosen_model, "base_url", "") or "").strip() or None,
)
image_debug_schema = str(resp.get("debug_used_schema") or "").strip()
dbg = resp.get("debug_used_debug")
@@ -290,91 +282,12 @@ class SpecialistAgentRunner:
image_debug_payload = dbg
elif dbg is not None:
image_debug_payload = str(dbg)
- ok = bool(resp.get("ok"))
- output = str(resp.get("text") or "").strip()
- if not ok:
- err = str(resp.get("error") or "").strip()
- output = f"Image generation failed: {err or 'unknown error'}"
- elif not output:
- output = "Image processed."
- # persist output images as attachment assets for UI rendering
- produced_attachments: list[dict[str, Any]] = []
- if ok:
- out_images = resp.get("images")
- if isinstance(out_images, list):
- store = AttachmentAssetStore()
- for idx, item in enumerate(out_images[:3], start=1):
- s = str(item or "").strip()
- if not s:
- continue
- if s.startswith("data:") and ";base64," in s:
- head, b64 = s.split(";base64,", 1)
- mime = head.replace("data:", "", 1) or "image/png"
- try:
- blob = base64.b64decode(b64.encode("ascii"))
- except Exception:
- continue
- meta = store.save_bytes(
- blob,
- filename=f"image-output-{idx}.png",
- mime=mime,
- )
- produced_attachments.append(
- {
- "type": "image_ref",
- "attachment_id": meta.attachment_id,
- "name": meta.name,
- "mime": meta.mime,
- "bytes": meta.bytes,
- "width": meta.width,
- "height": meta.height,
- }
- )
- elif s.startswith("http://") or s.startswith("https://"):
- try:
- with httpx.Client(timeout=20.0, follow_redirects=True) as client:
- r = client.get(s)
- if r.status_code < 400 and r.content:
- mime = str(r.headers.get("content-type") or "image/png").split(";", 1)[0].strip() or "image/png"
- ext = ".png"
- if mime == "image/jpeg":
- ext = ".jpg"
- elif mime == "image/webp":
- ext = ".webp"
- elif mime == "image/gif":
- ext = ".gif"
- meta = store.save_bytes(
- r.content,
- filename=f"image-output-{idx}{ext}",
- mime=mime,
- )
- produced_attachments.append(
- {
- "type": "image_ref",
- "attachment_id": meta.attachment_id,
- "name": meta.name,
- "mime": meta.mime,
- "bytes": meta.bytes,
- "width": meta.width,
- "height": meta.height,
- }
- )
- continue
- except Exception:
- pass
- produced_attachments.append(
- {
- "type": "image_url",
- "url": s,
- "name": f"image-output-{idx}.png",
- }
- )
- # Treat missing image outputs as failure to avoid false "generated" state.
- if not produced_attachments:
- ok = False
- output = (
- "Image generation failed: response succeeded but no image output was returned."
- )
+ ok, output, produced_attachments = legacy_image_turn_bundle(resp)
+ output = legacy_image_assistant_body_with_placeholder(
+ lang=self.lang,
+ body_text=output,
+ produced=produced_attachments if ok else None,
+ )
self.store.add_message(
session_id=session_id,
role="assistant",
diff --git a/runtime/chat/agent_messages.py b/runtime/chat/agent_messages.py
index f9d19c9e..9915e404 100644
--- a/runtime/chat/agent_messages.py
+++ b/runtime/chat/agent_messages.py
@@ -210,8 +210,10 @@ def build_llm_messages(
:func:`~oclaw.runtime.direct_loop._guard_tool_results_for_llm_context`). Omit or leave empty
to apply image-blob stripping for every tool row (safe default for callers without turn context).
- Only the **last** user message may expand attachments into native multimodal ``input_image``;
- older user attachments are replayed as text metadata only.
+ Only the **last** user message may expand attachments into native multimodal ``input_image``
+ blocks shaped like ``{"type":"input_image","image_base64","mime"}``; ``OpenAIChatModel`` and
+ ``OpenAIResponsesModel`` both accept this shape (see transport normalization code). Older user
+ attachments are replayed as text metadata only.
"""
out: list[dict[str, Any]] = [{"role": "system", "content": (system_prompt or "").strip()}]
dropped_unpaired_tool_rows = 0
diff --git a/runtime/direct_loop.py b/runtime/direct_loop.py
index a74da5e8..dd4adbca 100644
--- a/runtime/direct_loop.py
+++ b/runtime/direct_loop.py
@@ -1102,6 +1102,101 @@ def _execute_tool_step(
return int((time.perf_counter() - t0) * 1000), results_by_id
+def _maybe_image_specialist_legacy_gateway_turn(
+ *,
+ store: Any,
+ session_id: str,
+ turn_uuid: str,
+ lang: str,
+ model: ChatModel,
+ user_text: str,
+ attachments: list[dict[str, Any]] | None,
+ skill_binding_role: str | None,
+ on_token: Optional[Callable[[str], None]],
+ on_progress: Optional[Callable[[str], None]],
+) -> TurnRunOutcome | None:
+ """When the UI selects **image** specialist, skip Responses/chat-model transports.
+
+ Vision/gen HTTP goes through :func:`oclaw.platform.llm.image_legacy_client.send_legacy_image_messages`
+ (``/chat/completions`` lane). Disable with ``AIA_IMAGE_SPECIALIST_DISABLE_LEGACY_GATEWAY_LANE=1``.
+
+ End-to-end notes and safe edit boundaries: ``docs/IMAGE_SPECIALIST_LANE.md``.
+ """
+ if str(os.getenv("AIA_IMAGE_SPECIALIST_DISABLE_LEGACY_GATEWAY_LANE") or "").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+ "on",
+ ):
+ return None
+ if str(skill_binding_role or "").strip().lower() != "image":
+ return None
+
+ from oclaw.platform.llm.image_legacy_client import (
+ IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH,
+ collect_legacy_lane_images_from_attachments,
+ legacy_image_assistant_body_with_placeholder,
+ legacy_image_turn_bundle,
+ send_legacy_image_messages,
+ )
+
+ imgs = collect_legacy_lane_images_from_attachments(attachments)
+ if not imgs:
+ hint_en = "Image specialist received no image input. Attach an image and try again."
+ hint_zh = "图片专家未收到可用的图片输入；请先上传或附上图片后再试。"
+ hint = hint_en if str(lang or "").startswith("en") else hint_zh
+ store.add_message(
+ session_id=session_id,
+ role="assistant",
+ content=hint,
+ turn_uuid=turn_uuid,
+ event_type="assistant_text",
+ )
+ return TurnRunOutcome(
+ final_text=hint,
+ tool_traces=tuple(),
+ handoff_note="image_specialist_legacy_missing_attachment",
+ turn_uuid=turn_uuid,
+ )
+
+ if on_progress:
+ on_progress("oclaw: image specialist (legacy multimodal HTTP)…")
+
+ prompt_plain = str(user_text or "").strip()
+ if not prompt_plain:
+ prompt_plain = IMAGE_SPECIALIST_DEFAULT_PROMPT_ZH
+
+ resp = send_legacy_image_messages(
+ images=imgs,
+ prompt=prompt_plain,
+ model=str(getattr(model, "model", "") or "").strip() or None,
+ api_key=str(getattr(model, "api_key", "") or "").strip() or None,
+ base_url=str(getattr(model, "base_url", "") or "").strip() or None,
+ )
+ ok, body_text, produced = legacy_image_turn_bundle(resp)
+ body_text = legacy_image_assistant_body_with_placeholder(
+ lang=lang,
+ body_text=body_text,
+ produced=produced if ok else None,
+ )
+ store.add_message(
+ session_id=session_id,
+ role="assistant",
+ content=body_text,
+ turn_uuid=turn_uuid,
+ event_type="assistant_text",
+ attachments=(produced or None) if ok else None,
+ )
+ if ok and on_token and body_text:
+ on_token(body_text)
+ return TurnRunOutcome(
+ final_text=body_text,
+ tool_traces=tuple(),
+ handoff_note="image_specialist_legacy_http" if ok else "image_specialist_legacy_upstream_failed",
+ turn_uuid=turn_uuid,
+ )
+
+
def run_oclaw_direct_loop(
*,
store: Any,
@@ -1149,6 +1244,20 @@ def run_oclaw_direct_loop(
turn_uuid=turn_uuid,
event_type="user_text",
)
+ legacy_early = _maybe_image_specialist_legacy_gateway_turn(
+ store=store,
+ session_id=session_id,
+ turn_uuid=turn_uuid,
+ lang=lang,
+ model=model,
+ user_text=str(user_text or ""),
+ attachments=attachments,
+ skill_binding_role=skill_binding_role,
+ on_token=on_token,
+ on_progress=on_progress,
+ )
+ if legacy_early is not None:
+ return legacy_early
skill_exec = SkillExecutor(config=ToolExecutionConfig(max_workers=max(1, min(int(max_tool_workers or 8), 32))))
tool_traces: list[dict[str, Any]] = []
diff --git a/runtime/operations/scripts/probe_openai_responses_image.py b/runtime/operations/scripts/probe_openai_responses_image.py
new file mode 100644
index 00000000..219130c7
--- /dev/null
+++ b/runtime/operations/scripts/probe_openai_responses_image.py
@@ -0,0 +1,106 @@
+#!/usr/bin/env python3
+"""POST minimal vision payloads to an OpenAI-compatible ``/responses`` endpoint.
+
+Run locally: loads ``oclaw/_local/system.env`` the same way the gateway does, then POSTs variants.
+
+Example::
+
+ set OPENAI_BASE_URL=https://...
+ set OPENAI_API_KEY=sk-...
+ set OPENAI_MODEL=qwen-vl-plus
+ python runtime/operations/scripts/probe_openai_responses_image.py
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+if str(_REPO_ROOT) not in sys.path:
+ sys.path.insert(0, str(_REPO_ROOT))
+
+# Same bootstrap as gateway: ``interfaces/http/fastapi_app.py`` calls ``load_system_env()`` so
+# ``oclaw/_local/system.env`` is merged before reading ``OPENAI_*``.
+try:
+ from oclaw.platform.config.bootstrap_env import load_system_env
+
+ load_system_env()
+except ImportError:
+ pass
+
+try:
+ import httpx
+except ImportError:
+ print("install httpx: pip install httpx", file=sys.stderr)
+ raise SystemExit(2)
+
+# 1x1 transparent PNG
+_PNG_B64 = (
+ "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
+)
+
+
+def _variants(model: str, data_uri: str) -> list[tuple[str, dict]]:
+ user_block_resp = {
+ "type": "message",
+ "role": "user",
+ "content": [
+ {"type": "input_text", "text": "What color is this pixel image? One word."},
+ {"type": "input_image", "image_url": data_uri, "detail": "auto"},
+ ],
+ }
+ user_plain_resp = {
+ "role": "user",
+ "content": [
+ {"type": "input_text", "text": "What color is this pixel image? One word."},
+ {"type": "input_image", "image_url": data_uri, "detail": "auto"},
+ ],
+ }
+ user_chat = {
+ "role": "user",
+ "content": [
+ {"type": "text", "text": "What color is this pixel image? One word."},
+ {"type": "image_url", "image_url": {"url": data_uri}},
+ ],
+ }
+ return [
+ ("flat_resp_envelope", {"model": model, "input": [user_block_resp]}),
+ ("flat_resp_plain", {"model": model, "input": [user_plain_resp]}),
+ ("nested_messages_resp_env", {"model": model, "input": {"messages": [user_block_resp]}}),
+ ("nested_messages_resp_plain", {"model": model, "input": {"messages": [user_plain_resp]}}),
+ ("nested_messages_chat", {"model": model, "input": {"messages": [user_chat]}}),
+ ]
+
+
+def main() -> None:
+ base = (os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1").strip().rstrip("/")
+ key = (os.getenv("OPENAI_API_KEY") or "").strip()
+ model = (os.getenv("OPENAI_MODEL") or "gpt-4o-mini").strip()
+ if not key:
+ print("OPENAI_API_KEY is required", file=sys.stderr)
+ raise SystemExit(2)
+
+ url = f"{base}/responses"
+ data_uri = f"data:image/png;base64,{_PNG_B64}"
+ headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
+
+ print(f"POST {url}", flush=True)
+ for label, body in _variants(model, data_uri):
+ try:
+ r = httpx.post(url, headers=headers, json=body, timeout=120.0)
+ except httpx.HTTPError as exc:
+ print(f"\n=== {label} transport_error {exc}", flush=True)
+ continue
+ tail = (r.text or "")[:2400]
+ print(f"\n=== {label} status={r.status_code}", flush=True)
+ print(tail, flush=True)
+ if r.status_code < 400:
+ print("(first successful variant — use this shape for your gateway)", flush=True)
+ break
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_attachment_acl_backfill.py b/tests/test_attachment_acl_backfill.py
index 52be39ce..bb3fdc49 100644
--- a/tests/test_attachment_acl_backfill.py
+++ b/tests/test_attachment_acl_backfill.py
@@ -58,7 +58,7 @@ class AttachmentAclBackfillTests(unittest.TestCase):
def _h(self) -> dict[str, str]:
return {"authorization": f"Bearer {self.token}"}
- def test_backfill_enables_strict_acl_download(self) -> None:
+ def test_add_message_links_acl_for_strict_download(self) -> None:
store = SqliteStore(str(self.db))
sess = store.create_session_for_user(title="t", tenant_id=self.tenant_id, user_id=self.alice_id)
ast = AttachmentAssetStore()
@@ -75,11 +75,12 @@ class AttachmentAclBackfillTests(unittest.TestCase):
prev = os.environ.get("AIA_ATTACHMENT_ACL_STRICT")
os.environ["AIA_ATTACHMENT_ACL_STRICT"] = "1"
try:
- # Strict mode: without backfill, this should be forbidden (not avatar).
+ # Strict mode: add_message should have written attachment_acl (not only tool_result rows).
r0 = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h())
- self.assertEqual(r0.status_code, 403, r0.text)
+ self.assertEqual(r0.status_code, 200, r0.text)
+ self.assertTrue(len(r0.content) > 10)
- # Backfill ACL for tenant, then strict download should succeed.
+ # Backfill remains idempotent.
res = store.backfill_attachment_acl_from_messages(tenant_id=self.tenant_id, limit_messages=5000)
self.assertTrue(res.get("ok"), res)
r1 = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h())
diff --git a/tests/test_image_expert_env_lane.py b/tests/test_image_expert_env_lane.py
new file mode 100644
index 00000000..a54e77f8
--- /dev/null
+++ b/tests/test_image_expert_env_lane.py
@@ -0,0 +1,60 @@
+"""Image specialist lane must use ``AIA_IMAGE_EXPERT_*``, not ``AIA_OCR_*``."""
+
+from __future__ import annotations
+
+from oclaw.platform.llm.image_legacy_client import send_legacy_image_messages
+
+
+def test_legacy_fails_when_only_ocr_env_configured(monkeypatch: object) -> None:
+ monkeypatch.setenv("AIA_OCR_API_KEY", "k-ocr")
+ monkeypatch.setenv("AIA_OCR_BASE_URL", "https://ocr.example/v1")
+ monkeypatch.setenv("AIA_OCR_MODEL", "ocr-model")
+ for k in ("AIA_IMAGE_EXPERT_API_KEY", "AIA_IMAGE_EXPERT_BASE_URL", "AIA_IMAGE_EXPERT_MODEL"):
+ monkeypatch.delenv(k, raising=False)
+ out = send_legacy_image_messages(
+ images=["https://example.com/x.png"],
+ prompt="hi",
+ model=None,
+ api_key=None,
+ base_url=None,
+ )
+ assert out.get("ok") is False
+ err = str(out.get("error") or "")
+ assert "AIA_IMAGE_EXPERT" in err
+
+
+def test_legacy_accepts_explicit_kwargs_without_expert_env(monkeypatch: object) -> None:
+ """Explicit api_key/base_url/model override env (for scripts/tests only)."""
+ for k in (
+ "AIA_OCR_API_KEY",
+ "AIA_OCR_BASE_URL",
+ "AIA_IMAGE_EXPERT_API_KEY",
+ "AIA_IMAGE_EXPERT_BASE_URL",
+ ):
+ monkeypatch.delenv(k, raising=False)
+
+ recorded: dict[str, object] = {}
+
+ def fake_post(client, *, url: str, headers: dict[str, str], payload: dict[str, object]) -> object:
+ recorded["url"] = url
+
+ class R:
+ status_code = 200
+
+ def json(_self): # noqa: PLR6301
+ return {"choices": [{"message": {"content": [{"text": "ok"}]}}]}
+
+ return R()
+
+ import oclaw.platform.llm.image_legacy_client as mod
+
+ monkeypatch.setattr(mod, "post_with_retry", fake_post)
+ out = send_legacy_image_messages(
+ images=["https://example.com/x.png"],
+ prompt="hello",
+ model="m-special",
+ api_key="explicit-key",
+ base_url="https://gw.example/expert/v1",
+ )
+ assert out.get("ok") is True
+ assert "/chat/completions" in str(recorded.get("url") or "")
diff --git a/tests/test_image_legacy_gateway_lane.py b/tests/test_image_legacy_gateway_lane.py
new file mode 100644
index 00000000..e9dc72c1
--- /dev/null
+++ b/tests/test_image_legacy_gateway_lane.py
@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+from oclaw.platform.llm.image_http_common import (
+ dashscope_multimodal_http_ok,
+ dashscope_native_multimodal_url_from_compatible_base,
+ extract_text_and_images,
+)
+from oclaw.platform.llm.image_http_common import build_extract_diag_empty
+from oclaw.platform.llm.image_legacy_client import (
+ collect_legacy_lane_images_from_attachments,
+ legacy_image_assistant_body_with_placeholder,
+ legacy_image_turn_bundle,
+ normalize_legacy_output_image_urls,
+)
+
+
+def test_normalize_legacy_output_dict_image_parts() -> None:
+ u = "https://dashscope-result-sz.oss-cn-shenzhen.aliyuncs.com/x.png?Expires=1"
+ assert normalize_legacy_output_image_urls([{"image": u}]) == [u]
+ assert normalize_legacy_output_image_urls([{"image_url": {"url": u}}]) == [u]
+
+
+def test_legacy_turn_bundle_coerces_dict_images_to_attachments() -> None:
+ ok, text, att = legacy_image_turn_bundle(
+ {"ok": True, "text": "", "images": [{"image": "https://example.invalid/a.png"}]}
+ )
+ assert ok is True
+ assert len(att) == 1
+ assert att[0]["type"] == "image_url"
+ assert att[0]["url"] == "https://example.invalid/a.png"
+
+
+def test_dashscope_native_url_from_compatible_base() -> None:
+ assert (
+ dashscope_native_multimodal_url_from_compatible_base(
+ "https://dashscope.aliyuncs.com/compatible-mode/v1"
+ )
+ == "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"
+ )
+ assert (
+ dashscope_native_multimodal_url_from_compatible_base("https://example.com/openai/v1") is None
+ )
+
+
+def test_collect_legacy_lane_images_image_url() -> None:
+ atts = [{"type": "image_url", "url": "https://example.invalid/x.png"}]
+ assert collect_legacy_lane_images_from_attachments(atts) == ["https://example.invalid/x.png"]
+
+
+def test_collect_legacy_lane_images_raw_base64() -> None:
+ atts = [{"type": "input_image", "mime": "image/png", "image_base64": "SGVsbG8="}]
+ got = collect_legacy_lane_images_from_attachments(atts)
+ assert len(got) == 1
+ assert got[0].startswith("data:image/png;base64,")
+
+
+def test_legacy_turn_bundle_text_only_success() -> None:
+ ok, text, att = legacy_image_turn_bundle({"ok": True, "text": "caption only", "images": []})
+ assert ok is True
+ assert text == "caption only"
+ assert att == []
+
+
+def test_legacy_image_assistant_placeholder_zh_en() -> None:
+ produced = [{"type": "image_ref", "attachment_id": "a" * 64}]
+ assert "附件" in legacy_image_assistant_body_with_placeholder(
+ lang="zh", body_text="", produced=produced
+ )
+ assert "attachment" in legacy_image_assistant_body_with_placeholder(
+ lang="en", body_text="", produced=produced
+ ).lower()
+ assert legacy_image_assistant_body_with_placeholder(lang="zh", body_text="x", produced=produced) == "x"
+
+
+def test_legacy_turn_bundle_upstream_error() -> None:
+ ok, text, att = legacy_image_turn_bundle({"ok": False, "error": "rate"})
+ assert ok is False
+ assert "rate" in text
+ assert att == []
+
+
+def test_extract_harvests_nested_https_under_message() -> None:
+ url = "https://cdn.example.invalid/generated.png"
+ text, images = extract_text_and_images(
+ {
+ "choices": [
+ {
+ "message": {
+ "role": "assistant",
+ "content": None,
+ "metadata": {"preview_image": url},
+ }
+ }
+ ]
+ }
+ )
+ assert text == ""
+ assert images == [url]
+
+
+def test_build_extract_diag_top_level_openai_choices() -> None:
+ d = build_extract_diag_empty(
+ {
+ "choices": [],
+ "model": "x",
+ }
+ )
+ assert d.get("choices_len") == 0
+
+
+def test_legacy_turn_bundle_includes_provider_redacted() -> None:
+ ok, msg, att = legacy_image_turn_bundle(
+ {
+ "ok": True,
+ "text": "",
+ "images": [],
+ "extract_diag": {"choices_len": 0},
+ "provider_response_redacted": '{"choices":[]}',
+ }
+ )
+ assert ok is False
+ assert "provider_json=" in msg
+ assert att == []
+
+
+def test_legacy_turn_bundle_empty_ok_response_fails() -> None:
+ ok, text, att = legacy_image_turn_bundle({"ok": True, "text": "", "images": []})
+ assert ok is False
+ assert att == []
+
+
+def test_extract_text_and_images_dashscope_output_wrapper() -> None:
+ """Native multimodal HTTP wraps ``choices`` under ``output`` (not top-level OpenAI shape)."""
+ url = "https://dashscope-result-hz.oss-cn-hangzhou.aliyuncs.com/x.png?Expires=1"
+ payload = {
+ "status_code": 200,
+ "request_id": "959afba6-544e-487e-b58a-6bd9fea97xxx",
+ "code": "",
+ "message": "",
+ "output": {
+ "text": None,
+ "finish_reason": None,
+ "choices": [
+ {
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": [{"image": url}],
+ },
+ }
+ ],
+ "audio": None,
+ },
+ "usage": {
+ "input_tokens": 0,
+ "output_tokens": 0,
+ "image_count": 1,
+ "width": 2048,
+ "height": 2048,
+ },
+ }
+ text, images = extract_text_and_images(payload)
+ assert text == ""
+ assert images == [url]
+ assert dashscope_multimodal_http_ok(payload)[0] is True
+
+
+def test_extract_text_and_images_content_dict_not_list() -> None:
+ """Some gateways return a single object for ``message.content`` instead of an array."""
+ url = "https://dashscope-result-hz.oss-cn-hangzhou.aliyuncs.com/out.png"
+ text, images = extract_text_and_images(
+ {
+ "output": {
+ "choices": [
+ {
+ "message": {
+ "role": "assistant",
+ "content": {"image": url},
+ }
+ }
+ ]
+ }
+ }
+ )
+ assert text == ""
+ assert images == [url]
+
+
+def test_extract_text_and_images_messages_fallback() -> None:
+ text, images = extract_text_and_images(
+ {
+ "output": {
+ "messages": [
+ {"role": "user", "content": "x"},
+ {
+ "role": "assistant",
+ "content": [{"image": "https://example.invalid/gen.png"}],
+ },
+ ]
+ }
+ }
+ )
+ assert images == ["https://example.invalid/gen.png"]
+
+
+def test_extract_text_and_images_typed_image_url_part() -> None:
+ text, images = extract_text_and_images(
+ {
+ "choices": [
+ {
+ "message": {
+ "content": [
+ {
+ "type": "image_url",
+ "image_url": {"url": "https://example.invalid/v.png"},
+ }
+ ]
+ }
+ }
+ ]
+ }
+ )
+ assert images == ["https://example.invalid/v.png"]
+
+
+def test_dashscope_envelope_rejects_non_success_code() -> None:
+ ok, msg = dashscope_multimodal_http_ok({"code": "InvalidParameter", "message": "bad"})
+ assert ok is False
+ assert "bad" in msg
+
+
+def test_extract_text_and_images_openai_top_level_unchanged() -> None:
+ text, images = extract_text_and_images(
+ {
+ "choices": [
+ {
+ "message": {
+ "content": [
+ {"type": "text", "text": "hi"},
+ {"image": "https://example.invalid/a.jpg"},
+ ]
+ }
+ }
+ ]
+ }
+ )
+ assert "hi" in text
+ assert images == ["https://example.invalid/a.jpg"]
diff --git a/tests/test_openai_responses_normalize.py b/tests/test_openai_responses_normalize.py
new file mode 100644
index 00000000..d197c8ef
--- /dev/null
+++ b/tests/test_openai_responses_normalize.py
@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import httpx
+from openai import BadRequestError
+
+from oclaw.platform.llm.transports.openai_responses import OpenAIResponsesModel, _is_input_messages_validation_error
+
+
+def test_input_messages_validation_detects_body_not_str_exc() -> None:
+ """OpenAI SDK ``str(exc)`` is usually only ``Error code: 400``; gateway detail is in ``body``."""
+ req = httpx.Request("POST", "http://example.invalid/v1/responses")
+ resp = httpx.Response(400, request=req)
+ body = {
+ "message": (
+ "Input should be 'user': input.messages.0.role & Input should be a valid list: "
+ "input.messages.0.content"
+ ),
+ "type": "invalid_request_error",
+ "code": "invalid_parameter_error",
+ "param": None,
+ }
+ exc = BadRequestError("Error code: 400", response=resp, body=body)
+ assert "input.messages" not in str(exc).lower()
+ assert _is_input_messages_validation_error(exc) is True
+
+
+def test_strip_leading_system_to_instructions_kw() -> None:
+ msgs = [
+ {"role": "system", "content": "You are helpful."},
+ {"role": "user", "content": "Hi"},
+ ]
+ txt, tail = OpenAIResponsesModel._strip_leading_system_messages(msgs)
+ assert txt == "You are helpful."
+ assert tail == [{"role": "user", "content": "Hi"}]
+
+
+def test_normalize_default_responses_parts_and_openai_envelope() -> None:
+ msgs = [
+ {
+ "role": "user",
+ "content": [
+ {"type": "image_url", "image_url": {"url": "https://example.invalid/x.png"}},
+ {"type": "text", "text": "what?"},
+ ],
+ }
+ ]
+ out = OpenAIResponsesModel._normalize_messages(msgs)
+ assert len(out) == 1
+ assert out[0]["type"] == "message"
+ assert out[0]["role"] == "user"
+ cc = out[0]["content"]
+ assert isinstance(cc, list)
+ assert any(
+ x.get("type") == "input_image"
+ and isinstance(x.get("image_url"), str)
+ and x.get("detail") == "auto"
+ for x in cc
+ )
+ assert any(x.get("type") == "input_text" and x.get("text") == "what?" for x in cc)
+
+
+def test_normalize_nested_chat_parts_opt_in() -> None:
+ msgs = [
+ {
+ "role": "user",
+ "content": [{"type": "image_url", "image_url": {"url": "https://example.invalid/x.png"}}],
+ }
+ ]
+ out = OpenAIResponsesModel._normalize_messages(
+ msgs, envelope_openai_message=False, content_chat_completions_parts=True
+ )
+ assert "type" not in out[0]
+ assert out[0]["role"] == "user"
+ assert any(x.get("type") == "image_url" for x in out[0]["content"])
+
+
+def test_normalize_dashscope_shorthand_image_text_blocks() -> None:
+ msgs = [
+ {"role": "user", "content": [{"image": "https://example.invalid/y.png"}, {"text": "caption"}]},
+ ]
+ out = OpenAIResponsesModel._normalize_messages(msgs)
+ assert out[0]["type"] == "message"
+ assert out[0]["role"] == "user"
+ parts = out[0]["content"]
+ assert isinstance(parts, list)
+ imgs = [p for p in parts if isinstance(p, dict) and p.get("type") == "input_image"]
+ assert len(imgs) == 1 and imgs[0].get("detail") == "auto"
+ txts = [p for p in parts if isinstance(p, dict) and p.get("type") == "input_text"]
+ assert any("caption" in str(p.get("text")) for p in txts)
+
+
+def test_responses_input_candidates_cover_flat_and_nested() -> None:
+ msgs = [{"role": "user", "content": "hi"}]
+ flat = OpenAIResponsesModel._responses_input_candidates(
+ msgs, flat_responses=True, prefer_envelope=True, prefer_chat_parts=False
+ )
+ assert len(flat) >= 1
+ nested = OpenAIResponsesModel._responses_input_candidates(
+ msgs, flat_responses=False, prefer_envelope=True, prefer_chat_parts=False
+ )
+ tags = [t for t, _ in nested]
+ assert any("_messages" in t for t in tags)
+ assert any("_flat_input" in t for t in tags)
+
+
+def test_responses_input_candidates_primary_combo_first() -> None:
+ msgs = [{"role": "user", "content": "hi"}]
+ nested = OpenAIResponsesModel._responses_input_candidates(
+ msgs, flat_responses=False, prefer_envelope=True, prefer_chat_parts=False
+ )
+ assert nested[0][0] == "e1c0_messages"
+ assert nested[1][0] == "e1c0_flat_input"
+
+ nested2 = OpenAIResponsesModel._responses_input_candidates(
+ msgs, flat_responses=False, prefer_envelope=False, prefer_chat_parts=True
+ )
+ assert nested2[0][0] == "e0c1_messages"
+ assert nested2[1][0] == "e0c1_flat_input"
+
+
+def test_normalize_agent_messages_style_input_image() -> None:
+ """Matches ``build_llm_messages`` last-turn multimodal blocks (``input_image`` + ``image_base64``)."""
+ msgs = [
+ {
+ "role": "user",
+ "content": [
+ {"type": "input_image", "image_base64": "SGk=", "mime": "image/png"},
+ {"type": "text", "text": "what is this"},
+ ],
+ }
+ ]
+ out = OpenAIResponsesModel._normalize_messages(msgs)
+ assert len(out) == 1
+ assert out[0]["type"] == "message"
+ parts = out[0]["content"]
+ imgs = [p for p in parts if isinstance(p, dict) and p.get("type") == "input_image"]
+ assert len(imgs) == 1
+ assert imgs[0]["image_url"].startswith("data:image/png;base64,")
+ assert imgs[0].get("detail") == "auto"
+ txts = [p for p in parts if isinstance(p, dict) and p.get("type") == "input_text"]
+ assert any(p.get("text") == "what is this" for p in txts)
+
+ chat_parts = OpenAIResponsesModel._normalize_messages(
+ msgs, envelope_openai_message=False, content_chat_completions_parts=True
+ )
+ cp = chat_parts[0]["content"]
+ assert any(
+ isinstance(p, dict) and p.get("type") == "image_url" and "base64" in str(p.get("image_url", {}).get("url"))
+ for p in cp
+ )
+ assert any(p.get("type") == "text" and p.get("text") == "what is this" for p in cp)
+
+
+def test_image_legacy_compatible_mode_uses_image_url_parts() -> None:
+ from oclaw.platform.llm.image_legacy_client import _openai_compatible_vision_content
+
+ cc = _openai_compatible_vision_content(["data:image/jpeg;base64,SGk="], "go")
+ assert cc[-1]["type"] == "text"
+ assert cc[-1]["text"] == "go"
+ assert cc[0]["type"] == "image_url"