From a88fb523829fa9ab1bd7763c426b23d37fd1b63c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 13:03:41 +0900 Subject: [PATCH] feat(api): accept OpenAI image_url content parts with vision passthrough Vision clients send messages with image_url content parts. Validate url/detail shapes, accept multi-modal content arrays, and force single-agent passthrough so multi-agent verifiers do not fuse images. Fix _latest_user_text for list content. Include audited Semgrep FP nosemgrep on main-based cost_ledger SQL and orchestrator TLS/urllib. --- contextual_orchestrator/cost_ledger.py | 8 +- contextual_orchestrator/orchestrator.py | 29 ++++- contextual_orchestrator/server.py | 101 ++++++++++++++++- tests/test_image_url_content_parts.py | 141 ++++++++++++++++++++++++ 4 files changed, 267 insertions(+), 12 deletions(-) create mode 100644 tests/test_image_url_content_parts.py diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..8ca3c6dfd 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -583,12 +583,12 @@ def _seed_dimension_catalog(self) -> None: ph = self._placeholder() cur = self._conn.cursor() for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG): - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}", # nosec B608 - ph is a DB-API placeholder. (name,), ) if cur.fetchone() is None: - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound. "INSERT INTO cost_attribution_dimensions " f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})", # nosec B608 - ph is a DB-API placeholder. (name, label, order), @@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None: placeholders = ", ".join(ph for _ in _USAGE_COLUMNS) columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})", # nosec B608 - columns are fixed _USAGE_COLUMNS. tuple(row.get(column) for column in _USAGE_COLUMNS), ) @@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[ where = f" WHERE {' AND '.join(clauses)}" if clauses else "" columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. + cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..0da483f41 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -230,7 +230,7 @@ def __init__( @staticmethod def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext: if not verify_tls: - return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. + return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints. if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -307,7 +307,7 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str: def _open_provider(self, request: urllib.request.Request) -> Any: """Open a provider request built from a validated provider URL.""" - return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. request, timeout=self.timeout, context=self._ssl_context, @@ -1600,7 +1600,30 @@ def _needs_workflow(self, text: str) -> bool: return hits >= self.policy.conduct_hint_threshold or len(text) > 700 def _latest_user_text(self, messages: list[ChatMessage]) -> str: - return next((m.get("content", "") for m in reversed(messages) if m.get("role") == "user"), "") # pragma: no cover + """Return the latest user message as plain text for agent selection. + + OpenAI multi-modal messages may use a content-part array; extract text + chunks and ignore image_url parts so selection still works. + """ + for message in reversed(messages): + if message.get("role") != "user": + continue + content = message.get("content", "") + if isinstance(content, str): + return content + if isinstance(content, list): + parts: list[str] = [] + for chunk in content: + if isinstance(chunk, str) and chunk: + parts.append(chunk) + elif isinstance(chunk, dict): + text = chunk.get("text") + if isinstance(text, str) and text: + parts.append(text) + if parts: + return "\n".join(parts) + return "[image]" + return "" def _model_judge_verification(self, task: str, fallback: dict[str, Any]) -> dict[str, Any]: """Ask a model to judge the verifier report (fixes term-matching false negatives). diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb79..609f626ef 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -183,6 +183,90 @@ def _validate_mode(mode: Any) -> str: return mode + +def _message_has_image_content(messages: Any) -> bool: + """True when any message content part is an OpenAI image_url part.""" + if not isinstance(messages, list): + return False + for message in messages: + if not isinstance(message, dict): + continue + content = message.get("content") + if not isinstance(content, list): + continue + for part in content: + if isinstance(part, dict) and part.get("type") == "image_url": + return True + return False + + +def _normalize_message_content(content: Any) -> str: + """Normalize OpenAI message content to a plain string for orchestration. + + Accepts a string or an array of content parts. ``text`` parts are joined. + ``image_url`` parts are accepted for schema parity but force single-agent + passthrough (see chat completions handler); the orchestrated multi-agent + path cannot fuse vision inputs across workers, so callers with images are + routed to passthrough before orchestration runs. + """ + if isinstance(content, str): + return content + if not isinstance(content, list) or not content: + raise RequestError(400, "invalid_message", "message role or content is invalid") + text_parts: list[str] = [] + has_image = False + for part in content: + if isinstance(part, str): + if part: + text_parts.append(part) + continue + if not isinstance(part, dict): + raise RequestError(400, "invalid_message", "message role or content is invalid") + part_type = part.get("type", "text") + if part_type == "text": + text = part.get("text") + if not isinstance(text, str): + raise RequestError(400, "invalid_message", "message role or content is invalid") + if text: + text_parts.append(text) + continue + if part_type == "image_url": + image_url = part.get("image_url") + if not isinstance(image_url, dict): + raise RequestError( + 400, + "invalid_message_content", + "image_url content part must include an image_url object", + ) + url = image_url.get("url") + if not isinstance(url, str) or not url.strip(): + raise RequestError( + 400, + "invalid_message_content", + "image_url.url must be a non-empty string", + ) + detail = image_url.get("detail") + if detail is not None and detail not in {"auto", "low", "high"}: + raise RequestError( + 400, + "invalid_message_content", + "image_url.detail must be auto, low, or high when present", + ) + has_image = True + continue + raise RequestError( + 400, + "invalid_message_content", + f"content part type {part_type!r} is not supported", + {"part_type": part_type}, + ) + if not text_parts and not has_image: + raise RequestError(400, "invalid_message", "message role or content is invalid") + if not text_parts and has_image: + return "[image]" + return "\n".join(text_parts) + + def _validate_messages(messages: Any) -> list[dict[str, str]]: if not isinstance(messages, list) or not messages: raise RequestError(400, "invalid_message", "messages must be a non-empty array") @@ -191,9 +275,9 @@ def _validate_messages(messages: Any) -> list[dict[str, str]]: if not isinstance(message, dict): raise RequestError(400, "invalid_message", "each message must be an object") role = message.get("role") - content = message.get("content") - if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES or not isinstance(content, str): + if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES: raise RequestError(400, "invalid_message", "message role or content is invalid") + content = _normalize_message_content(message.get("content")) validated.append({"role": role, "content": content}) return validated @@ -713,9 +797,16 @@ def do_POST(self) -> None: # noqa: N802 if path == "/v1/chat/completions": _reject_unknown_keys(body, ALLOWED_CHAT_KEYS) - if PASSTHROUGH_TRIGGER_KEYS & set(body): - # response_format / tools cannot be merged across agents; - # proxy the full request to one agent and return it verbatim. + # Validate message shapes first (including image_url content parts) + # so invalid multi-modal payloads fail with 400 before proxy. + if isinstance(body.get("messages"), list): + for message in body["messages"]: + if isinstance(message, dict) and "content" in message: + _normalize_message_content(message.get("content")) + if PASSTHROUGH_TRIGGER_KEYS & set(body) or _message_has_image_content(body.get("messages")): + # response_format / tools / vision image_url parts cannot be + # merged across multi-agent verifiers; proxy to one agent. + started_at = time.perf_counter() proxied = self._run( lambda: orchestrator.proxy_completion(body, endpoint="chat/completions") diff --git a/tests/test_image_url_content_parts.py b/tests/test_image_url_content_parts.py new file mode 100644 index 000000000..9a7ff0939 --- /dev/null +++ b/tests/test_image_url_content_parts.py @@ -0,0 +1,141 @@ +"""OpenAI image_url content parts on chat messages force vision passthrough.""" + +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import ( # noqa: E402 + RequestError, + SecurityConfig, + _message_has_image_content, + _normalize_message_content, + build_server, +) + +_TEST_AUTH_TOKEN = "img_url_token" # noqa: S105 + + +def build() -> TaskOrchestrator: + return TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))] + ) + + +def test_normalize_and_detect_image_parts() -> None: + assert _normalize_message_content("hi") == "hi" + assert _normalize_message_content( + [{"type": "text", "text": "what is this?"}, {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}}] + ) == "what is this?" + assert _normalize_message_content( + [{"type": "image_url", "image_url": {"url": "https://example.com/a.png", "detail": "low"}}] + ) == "[image]" + assert _message_has_image_content( + [{"role": "user", "content": [{"type": "image_url", "image_url": {"url": "https://x/y"}}]}] + ) + try: + _normalize_message_content( + [{"type": "image_url", "image_url": {"url": ""}}] + ) + raise AssertionError("expected invalid empty url") + except RequestError as exc: + assert exc.code == "invalid_message_content" + try: + _normalize_message_content( + [{"type": "image_url", "image_url": {"url": "https://x", "detail": "ultra"}}] + ) + raise AssertionError("expected invalid detail") + except RequestError as exc: + assert exc.code == "invalid_message_content" + + +def _post(port: int, path: str, payload: dict) -> tuple[int, dict]: + request = urllib.request.Request( + f"http://127.0.0.1:{port}{path}", + data=json.dumps(payload).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=10) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def test_http_chat_accepts_image_url_parts() -> None: + server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = _post( + port, + "/v1/chat/completions", + { + "model": "mock-generalist", + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "describe"}, + { + "type": "image_url", + "image_url": { + "url": "https://example.com/cat.png", + "detail": "auto", + }, + }, + ], + } + ], + }, + ) + assert status in {200, 202}, body + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_http_chat_rejects_bad_image_url() -> None: + server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = _post( + port, + "/v1/chat/completions", + { + "model": "mock-generalist", + "messages": [ + { + "role": "user", + "content": [{"type": "image_url", "image_url": {"url": ""}}], + } + ], + }, + ) + assert status == 400, body + assert body["error"]["code"] == "invalid_message_content" + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_normalize_and_detect_image_parts() + test_http_chat_accepts_image_url_parts() + test_http_chat_rejects_bad_image_url() + print("ok")