fix(api): accept OpenAI-style multimodal message content for vision

contextual_orchestrator/cost_ledger.py

@@ -583,6 +583,7 @@ def _seed_dimension_catalog(self) -> None:
         ph = self._placeholder()
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
+            # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
             cur.execute(
                 f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
                 (name,),
@@ -602,6 +603,7 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
+        # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
         cur.execute(
             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,6 +624,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()
+        # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
         cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 

contextual_orchestrator/orchestrator.py

@@ -29,7 +29,9 @@
 from .credentials import NotConfigured, get_credential
 
 
-ChatMessage = dict[str, str]
+# `content` is usually str; a multimodal (vision) message's content is an
+# OpenAI-style parts list instead -- see server._validate_message_content.
+ChatMessage = dict[str, Any]
 
 class BudgetExceededError(RuntimeError):
     """Raised when an operator-configured spend budget is already exhausted."""
@@ -230,6 +232,7 @@ def __init__(
     @staticmethod
     def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext:
         if not verify_tls:
+            # nosemgrep: python.lang.security.unverified-ssl-context.unverified-ssl-context
             return ssl._create_unverified_context()  # nosec B323 - explicit dev-only provider TLS opt-out.
         if ca_bundle:
             if not os.path.isfile(ca_bundle):
@@ -307,6 +310,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."""
+        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
         return urllib.request.urlopen(  # nosec B310 - request URL comes from _provider_url after provider validation.
             request,
             timeout=self.timeout,
@@ -479,6 +483,7 @@ def _validate_provider(self, agent: ModelAgent) -> None:
                 or ip_address.is_link_local
                 or ip_address.is_multicast
                 or ip_address.is_reserved
+                or ip_address.is_unspecified
             ):
                 raise RuntimeError(f"{agent.id} provider resolves to non-public address")
 
@@ -492,9 +497,11 @@ def _provider_url(self, agent: ModelAgent, path: str) -> str:
         return f"{agent.base_url.rstrip('/')}{path}"
 
     def _mock(self, agent: ModelAgent, messages: list[ChatMessage]) -> str:
-        last = next((m["content"] for m in reversed(messages) if m.get("role") == "user"), "")
+        last = next(
+            (_message_content_text(m["content"]) for m in reversed(messages) if m.get("role") == "user"), ""
+        )
         role = "worker"
-        system = messages[0]["content"] if messages and messages[0].get("role") == "system" else ""
+        system = _message_content_text(messages[0]["content"]) if messages and messages[0].get("role") == "system" else ""
         match = re.search(r"Role: ([a-z]+)", system)
         if match:
             role = match.group(1)
@@ -644,6 +651,36 @@ def _coerce_input_text(value: Any) -> str:
     return " ".join(parts)
 
 
+def _message_content_text(content: Any) -> str:
+    """Best-effort text from a chat message's ``content`` for routing/
+    complexity heuristics -- a multimodal (vision) message's content is a
+    parts list (``{"type": "text", ...}`` / ``{"type": "image_url", ...}``);
+    only the text parts feed routing, the image data is opaque to it.
+    """
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return " ".join(
+            part["text"]
+            for part in content
+            if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str)
+        )
+    return ""
+
+
+def _message_image_parts(content: Any) -> list[dict[str, Any]]:
+    """Non-text content parts (``image_url``, etc.) from a chat message's ``content``.
+
+    ``_message_content_text`` deliberately drops these for routing/complexity
+    heuristics; callers that forward a message to a provider must reattach them
+    separately, or a vision request silently loses its image once it enters a
+    text-only path (e.g. ``conduct``'s generated step prompts).
+    """
+    if not isinstance(content, list):
+        return []
+    return [part for part in content if isinstance(part, dict) and part.get("type") != "text"]
+
+
 def load_agents(path: str) -> list[ModelAgent]:  # pragma: no cover
     """Load model agent definitions from an agents JSON file."""
     with open(path, encoding="utf-8") as handle:
@@ -1367,6 +1404,14 @@ def route_once(self, messages: list[ChatMessage]) -> dict[str, Any]:
     def conduct(self, messages: list[ChatMessage]) -> dict[str, Any]:
         """Run a planned workflow: fixed template, or a Conductor-style generated plan."""
         task = self._latest_user_text(messages)
+        # The original user message's image_url (etc.) parts -- `task` above is
+        # text-only (routing/complexity heuristics only), so a vision request that
+        # falls into this workflow path would otherwise never reach a provider
+        # with its image attached. Reattached to every step's synthesized prompt.
+        latest_user_content = next(
+            (m.get("content") for m in reversed(messages) if m.get("role") == "user"), None
+        )
+        image_parts = _message_image_parts(latest_user_content)
         plan_source = "template"
         if self.policy.workflow_planning == "generated":
             try:
@@ -1383,6 +1428,10 @@ def conduct(self, messages: list[ChatMessage]) -> dict[str, Any]:
         for step in steps:
             agent = self._agent(step.agent_id)
             prior = "\n\n".join(f"Step {i}: {outputs[i]}" for i in step.access)
+            user_text = f"Original task:\n{task}\n\nAccessed prior work:\n{prior}\n\nSubtask:\n{step.subtask}"
+            user_content: Any = (
+                [{"type": "text", "text": user_text}, *image_parts] if image_parts else user_text
+            )
             step_messages = [
                 {
                     "role": "system",
@@ -1394,7 +1443,7 @@ def conduct(self, messages: list[ChatMessage]) -> dict[str, Any]:
                 },
                 {
                     "role": "user",
-                    "content": f"Original task:\n{task}\n\nAccessed prior work:\n{prior}\n\nSubtask:\n{step.subtask}",
+                    "content": user_content,
                 },
             ]
             start = time.perf_counter()
@@ -1600,7 +1649,10 @@ 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 next(
+            (_message_content_text(m.get("content", "")) for m in reversed(messages) if m.get("role") == "user"),
+            "",
+        )  # pragma: no cover
 
     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).

contextual_orchestrator/server.py

@@ -183,17 +183,50 @@ def _validate_mode(mode: Any) -> str:
     return mode
 
 
-def _validate_messages(messages: Any) -> list[dict[str, str]]:
+def _validate_message_content(content: Any) -> str | list[dict[str, Any]]:
+    """A message's ``content`` is either plain text, or an OpenAI-style
+    multimodal parts array (vision requests: ``[{"type": "text", ...},
+    {"type": "image_url", "image_url": {"url": "data:...;base64,..."}}]``)
+    -- omni-modal callers (e.g. LineageWeave's image OCR/captioning) send
+    the latter, and rejecting it here was the same "channel silently
+    degrades to unavailable" failure this codebase's own pluggable-client
+    discipline is built to avoid everywhere else.
+    """
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list) and content:
+        parts: list[dict[str, Any]] = []
+        for part in content:
+            if not isinstance(part, dict):
+                raise RequestError(400, "invalid_message", "message content part must be an object")
+            part_type = part.get("type")
+            if part_type == "text":
+                if not isinstance(part.get("text"), str):
+                    raise RequestError(400, "invalid_message", "text content part requires a string 'text'")
+            elif part_type == "image_url":
+                image_url = part.get("image_url")
+                if not isinstance(image_url, dict) or not isinstance(image_url.get("url"), str):
+                    raise RequestError(
+                        400, "invalid_message", "image_url content part requires image_url.url as a string"
+                    )
+            else:
+                raise RequestError(400, "invalid_message", "content part 'type' must be text or image_url")
+            parts.append(part)
+        return parts
+    raise RequestError(400, "invalid_message", "message role or content is invalid")
+
+
+def _validate_messages(messages: Any) -> list[dict[str, Any]]:
     if not isinstance(messages, list) or not messages:
         raise RequestError(400, "invalid_message", "messages must be a non-empty array")
-    validated: list[dict[str, str]] = []
+    validated: list[dict[str, Any]] = []
     for message in messages:
         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 = _validate_message_content(message.get("content"))
         validated.append({"role": role, "content": content})
     return validated
 

tests/test_paper_contracts.py

@@ -47,6 +47,33 @@ def test_trinity_contract_has_explicit_thinker_worker_verifier_roles() -> None:
     assert ["thinker", "worker", "verifier"] == [step["role"] for step in result["trace"][:3]]
 
 
+def test_conduct_forwards_original_image_parts_to_every_step() -> None:
+    """A vision request that falls into the multi-step conduct() workflow path
+    (e.g. via an "analyze" hint) must not silently lose its image -- only its
+    text feeds routing/complexity heuristics; the image_url part must still
+    reach every step's provider call.
+    """
+    client = RecordingClient()
+    image_part = {"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="}}
+    build(client).conduct(
+        [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Analyze this diagram and implement a safe parser."},
+                    image_part,
+                ],
+            }
+        ]
+    )
+
+    assert client.calls  # the workflow actually dispatched steps
+    for _agent_id, messages in client.calls:
+        user_content = messages[1]["content"]
+        assert isinstance(user_content, list), "image request must not collapse to a plain-text prompt"
+        assert image_part in user_content
+
+
 def test_conductor_contract_uses_access_lists_to_control_context() -> None:
     client = RecordingClient()
     build(client).conduct([{"role": "user", "content": "Analyze, implement, verify, and synthesize."}])

tests/test_security_hardening.py

@@ -132,6 +132,66 @@ def test_http_api_validates_mode_and_request_shape() -> None:
     assert body["error"]["code"] in {"invalid_message", "invalid_mode"}
 
 
+def test_http_api_accepts_multimodal_vision_content() -> None:
+    """An OpenAI-style content-parts array (text + image_url) -- what a
+    vision/omni-modal caller like LineageWeave's image OCR sends -- must be
+    accepted, not rejected as an invalid message; this was a real bug:
+    _validate_messages required content to be a plain string.
+    """
+    server = build_server(build(), port=0, security=SecurityConfig(auth_token="secret_token"))
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    port = server.server_address[1]
+    payload = {
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "describe this receipt"},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="},
+                    },
+                ],
+            }
+        ]
+    }
+
+    try:
+        status, body = post_json(
+            f"http://127.0.0.1:{port}/v1/chat/completions", payload, token="secret_token"
+        )
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert status == 200
+    assert "describe this receipt" in body["choices"][0]["message"]["content"]
+
+
+def test_http_api_rejects_malformed_multimodal_content_part() -> None:
+    server = build_server(build(), port=0, security=SecurityConfig(auth_token="secret_token"))
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    port = server.server_address[1]
+    payload = {
+        "messages": [
+            {"role": "user", "content": [{"type": "image_url", "image_url": {"missing_url_key": True}}]}
+        ]
+    }
+
+    try:
+        status, body = post_json(
+            f"http://127.0.0.1:{port}/v1/chat/completions", payload, token="secret_token"
+        )
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert status == 400
+    assert body["error"]["code"] == "invalid_message"
+
+
 def test_http_api_rejects_unknown_request_fields() -> None:
     server = build_server(build(), port=0, security=SecurityConfig(auth_token="secret_token"))
     thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -254,6 +314,28 @@ def test_external_provider_requires_resolvable_credential_and_public_https() ->
         set_backend(None)
 
 
+def test_external_provider_rejects_unspecified_addresses() -> None:
+    """0.0.0.0 and [::] are unrouteable "any interface" addresses, not public --
+    Strix-flagged SSRF gap: they cleared every prior is_private/is_loopback/
+    is_link_local/is_multicast/is_reserved check.
+    """
+    client = ModelClient()
+    backend = InMemoryCredentialBackend()
+    backend.set("MODEL_KEY", "sk-unspecified-check")
+    set_backend(backend)
+    try:
+        for host in ("0.0.0.0", "[::]"):
+            agent = ModelAgent("unspecified_agent", "gpt-example", f"https://{host}/v1", "MODEL_KEY")
+            try:
+                client._validate_provider(agent)
+            except RuntimeError as exc:
+                assert "non-public address" in str(exc)
+            else:
+                raise AssertionError(f"unspecified-address provider {host} should fail")
+    finally:
+        set_backend(None)
+
+
 def test_external_provider_rejects_insecure_or_unlisted_hosts() -> None:
     client = ModelClient()
     insecure_agent = ModelAgent("insecure_agent", "gpt-example", "http://api.openai.com/v1", "MODEL_KEY")