fix: bug on openai compat file upload

application/llm/openai.py

@@ -1,5 +1,6 @@
 import base64
 import hashlib
+import io
 import json
 import logging
 
@@ -172,6 +173,12 @@ class OpenAILLM(BaseLLM):
         # fallback restream of the already-delivered answer.
         self._stream_reached_finish = False
         self._imported_response_id = None
+        # Files-API ids for inline ``file_data`` content parts already
+        # uploaded, keyed by content hash. First-line cache for the
+        # in-request tool loop; the Redis-backed cross-request cache
+        # (see ``_inline_file_id_cache_*``) covers /v1 clients that
+        # resend the same ``file_data`` on every turn.
+        self._inline_file_ids = {}
 
     def responses_chain_key(self) -> str:
         """Return a credential- and endpoint-scoped Responses chain key.
@@ -236,6 +243,137 @@ class OpenAILLM(BaseLLM):
         self._imported_response_id = None
         self._last_finish_reason = None
 
+    def _resolve_file_part(self, item):
+        """Resolve a ``file`` content part into a Files-API reference.
+
+        Clients (the /v1 passthrough in particular) may send OpenAI-style
+        file parts carrying an inline ``file_data`` data-URI. Azure's
+        Responses API rejects inline data with ``unsupported_file`` even for
+        valid PDFs, and string-content-only chat deployments 4xx on any file
+        part — so upload the bytes once and swap in the ``file_id`` the
+        deployments do accept. A part with neither ``file_id`` nor decodable
+        ``file_data``, or whose upload fails (e.g. the endpoint has no Files
+        API), degrades to a text note instead of a certain provider 4xx.
+        """
+        file_obj = item.get("file") or {}
+        if file_obj.get("file_id"):
+            # Normalize: a client that sends both ``file_id`` and
+            # ``file_data`` would otherwise leak the inline payload to
+            # ``_responses_content_parts``, which copies every truthy key
+            # into ``input_file`` — and Azure Responses then rejects on
+            # the ``file_data`` regardless of the ``file_id``.
+            return {"type": "file", "file": {"file_id": file_obj["file_id"]}}
+        filename = file_obj.get("filename") or "upload.pdf"
+        file_data = file_obj.get("file_data")
+        if file_data:
+            content_hash = hashlib.sha256(file_data.encode()).hexdigest()
+            cached = self._inline_file_ids.get(content_hash)
+            if cached:
+                return {"type": "file", "file": {"file_id": cached}}
+            cached = self._inline_file_id_cache_get(content_hash)
+            if cached:
+                self._inline_file_ids[content_hash] = cached
+                return {"type": "file", "file": {"file_id": cached}}
+            try:
+                payload = file_data
+                if payload.startswith("data:"):
+                    _, _, payload = payload.partition(",")
+                # MIME-wrapped encoders (``base64.encodebytes``, some
+                # JSON pretty-printers) insert whitespace/newlines that
+                # ``validate=True`` rejects — strip so recoverable data
+                # doesn't get thrown into the text-note degrade path.
+                payload = "".join(payload.split())
+                if not payload:
+                    # A data URI missing the comma (or one with an empty
+                    # payload) would otherwise decode to zero bytes and
+                    # upload an empty artifact; degrade deliberately.
+                    raise ValueError("empty file_data payload")
+                raw = base64.b64decode(payload, validate=True)
+                file_id = self.client.files.create(
+                    file=(filename, io.BytesIO(raw)),
+                    purpose="assistants",
+                ).id
+                self._inline_file_ids[content_hash] = file_id
+                self._inline_file_id_cache_set(content_hash, file_id)
+                return {"type": "file", "file": {"file_id": file_id}}
+            except Exception as e:
+                logging.warning(
+                    "Could not resolve inline file_data part '%s' to a "
+                    "file_id (%s); degrading to a text note",
+                    filename,
+                    e,
+                )
+        else:
+            logging.warning(
+                "File content part '%s' has neither file_id nor file_data; "
+                "degrading to a text note",
+                filename,
+            )
+        return {
+            "type": "text",
+            "text": f"[File '{filename}' could not be processed]",
+        }
+
+    def _inline_file_id_cache_key(self, content_hash: str) -> str:
+        """Redis key for the inline-file-data → file_id cache.
+
+        Scoped by ``(provider_name, base_url, api_key)``: a Files-API
+        ``file_id`` is only valid for the endpoint + credential it was
+        uploaded to, so a shared key across providers would return
+        ids the current call can't use.
+        """
+        creds = "\0".join(
+            (
+                self.provider_name or "",
+                self._effective_base_url or "",
+                self.api_key or "",
+            )
+        )
+        creds_hash = hashlib.sha256(creds.encode("utf-8")).hexdigest()[:16]
+        return f"openai_inline_file:{creds_hash}:{content_hash}"
+
+    def _inline_file_id_cache_get(self, content_hash: str):
+        """Look up a previously-uploaded file_id for this content.
+
+        Returns None on cache miss or on any Redis error — the caller
+        then uploads normally. Never raises.
+        """
+        try:
+            from application.cache import get_redis_instance
+
+            r = get_redis_instance()
+            if r is None:
+                return None
+            value = r.get(self._inline_file_id_cache_key(content_hash))
+            if value is None:
+                return None
+            return value.decode() if isinstance(value, (bytes, bytearray)) else value
+        except Exception as e:
+            logging.debug("inline_file_id cache read failed: %s", e)
+            return None
+
+    def _inline_file_id_cache_set(self, content_hash: str, file_id: str) -> None:
+        """Persist a fresh (content_hash → file_id) mapping.
+
+        24 h TTL: well inside Azure's 30-day retention for
+        ``purpose="assistants"``, long enough to cover multi-turn
+        conversations that resend the same ``file_data`` every turn.
+        Silent on failure.
+        """
+        try:
+            from application.cache import get_redis_instance
+
+            r = get_redis_instance()
+            if r is None:
+                return
+            r.setex(
+                self._inline_file_id_cache_key(content_hash),
+                86400,
+                file_id,
+            )
+        except Exception as e:
+            logging.debug("inline_file_id cache write failed: %s", e)
+
     def _clean_messages_openai(self, messages):
         cleaned_messages = []
         for message in messages:
@@ -346,7 +484,7 @@ class OpenAILLM(BaseLLM):
                             if "type" in item and item["type"] == "text" and "text" in item:
                                 content_parts.append(item)
                             elif "type" in item and item["type"] == "file" and "file" in item:
-                                content_parts.append(item)
+                                content_parts.append(self._resolve_file_part(item))
                             elif "type" in item and item["type"] == "image_url" and "image_url" in item:
                                 content_parts.append(item)
                             elif "text" in item and "type" not in item:

tests/conftest.py

@@ -138,6 +138,29 @@ def pg_conn(pg_engine):
 # ---------------------------------------------------------------------------
 
 
+@pytest.fixture(autouse=True)
+def _no_real_redis(monkeypatch):
+    """Force the Redis-absent baseline CI has.
+
+    CI runs without a Redis service, so ``get_redis_instance()`` /
+    ``get_pubsub_redis_instance()`` return None there. A dev machine with
+    a live localhost Redis diverges: code under test silently reads and
+    writes real keys, leaking state between tests and between runs (seen
+    with the ``openai_inline_file:*`` upload cache). Flipping the
+    creation-failed flags makes the real accessors return None everywhere
+    regardless of import style — consumers that did ``from
+    application.cache import get_redis_instance`` still hit these module
+    globals at call time. Tests that want Redis behavior keep injecting
+    fakes by patching the accessor at the consumer module, which bypasses
+    this guard. A test targeting the accessor's own construction path
+    must reset the two flags first.
+    """
+    monkeypatch.setattr("application.cache._redis_instance", None)
+    monkeypatch.setattr("application.cache._redis_creation_failed", True)
+    monkeypatch.setattr("application.cache._pubsub_redis_instance", None)
+    monkeypatch.setattr("application.cache._pubsub_redis_creation_failed", True)
+
+
 @pytest.fixture
 def mock_llm():
     llm = Mock()

tests/llm/test_openai.py

@@ -14,8 +14,9 @@ Extends coverage beyond test_openai_llm.py:
   - _get_base64_image / _upload_file_to_openai
 """
 
+import base64
 import types
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -1540,3 +1541,332 @@ class TestUploadFileToOpenAIError:
         )
         with pytest.raises(FileNotFoundError, match="File not found"):
             llm._upload_file_to_openai({"path": "/missing.pdf"})
+
+
+# _clean_messages_openai — inline file_data resolution (ledger #1 unsupported_file)
+
+
+_TINY_PDF_BYTES = b"%PDF-1.4 tiny e2e stub"
+_TINY_PDF_B64 = base64.b64encode(_TINY_PDF_BYTES).decode()
+
+
+class _CountingFiles:
+    """Files-API stub that records uploads and can simulate an endpoint
+    without a Files API (the OpenAI-compatible fallback deployments)."""
+
+    def __init__(self, fail=False):
+        self.calls = []
+        self.fail = fail
+
+    def create(self, file=None, purpose=None):
+        self.calls.append((file, purpose))
+        if self.fail:
+            raise RuntimeError("files API unavailable")
+        return types.SimpleNamespace(id=f"file-{len(self.calls)}")
+
+
+class TestInlineFilePartResolution:
+    def _msg(self, file_obj):
+        return [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "what is in the file?"},
+                    {"type": "file", "file": file_obj},
+                ],
+            }
+        ]
+
+    def test_file_data_uri_uploaded_and_swapped_for_file_id(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "filename": "Scan-48.pdf",
+                    "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}",
+                }
+            )
+        )
+        parts = out[0]["content"]
+        assert parts[0] == {"type": "text", "text": "what is in the file?"}
+        assert parts[1] == {"type": "file", "file": {"file_id": "file-1"}}
+        ((uploaded, purpose),) = files.calls
+        assert purpose == "assistants"
+        assert uploaded[0] == "Scan-48.pdf"
+        assert uploaded[1].read() == _TINY_PDF_BYTES
+
+    def test_raw_base64_without_data_uri_prefix(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg({"filename": "a.pdf", "file_data": _TINY_PDF_B64})
+        )
+        assert out[0]["content"][1]["file"] == {"file_id": "file-1"}
+
+    def test_same_file_data_uploaded_once(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        msg = self._msg(
+            {
+                "filename": "a.pdf",
+                "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}",
+            }
+        )
+        llm._clean_messages_openai(msg)
+        out2 = llm._clean_messages_openai(msg)
+        assert len(files.calls) == 1
+        assert out2[0]["content"][1]["file"] == {"file_id": "file-1"}
+
+    def test_upload_failure_degrades_to_text_note(self, llm):
+        llm.client.files = _CountingFiles(fail=True)
+        out = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "filename": "Scan-48.pdf",
+                    "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}",
+                }
+            )
+        )
+        parts = out[0]["content"]
+        assert parts[1]["type"] == "text"
+        assert "Scan-48.pdf" in parts[1]["text"]
+
+    def test_invalid_base64_degrades_to_text_note(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg({"filename": "x.pdf", "file_data": "data:application/pdf;base64,%%%not-base64%%%"})
+        )
+        assert out[0]["content"][1]["type"] == "text"
+        assert files.calls == []
+
+    def test_existing_file_id_part_untouched(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(self._msg({"file_id": "file-abc"}))
+        assert out[0]["content"][1] == {
+            "type": "file",
+            "file": {"file_id": "file-abc"},
+        }
+        assert files.calls == []
+
+    def test_empty_file_part_degrades_to_text_note(self, llm):
+        # The classic ledger-#1 flavor: {"type": "file", "file": {"file_id": None}}
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(self._msg({"file_id": None}))
+        parts = out[0]["content"]
+        assert parts[1]["type"] == "text"
+        assert files.calls == []
+
+    def test_responses_parts_conversion_gets_file_id(self, llm):
+        files = _CountingFiles()
+        llm.client.files = files
+        cleaned = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "filename": "a.pdf",
+                    "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}",
+                }
+            )
+        )
+        parts = OpenAILLM._responses_content_parts("user", cleaned[0]["content"])
+        assert {"type": "input_file", "file_id": "file-1"} in parts
+
+    # -- Fix #1: file_id + file_data both present must drop file_data --
+
+    def test_file_id_and_file_data_normalized_to_file_id_only(self, llm):
+        """A client sending both keys would otherwise leak ``file_data``
+        into ``_responses_content_parts`` (Azure Responses then 400s on
+        the inline payload regardless of the ``file_id``)."""
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "file_id": "file-preexisting",
+                    "filename": "Scan-48.pdf",
+                    "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}",
+                }
+            )
+        )
+        part = out[0]["content"][1]
+        # Normalized: only ``file_id`` survives — no ``file_data``, no
+        # stray ``filename`` that _responses_content_parts would ship.
+        assert part == {"type": "file", "file": {"file_id": "file-preexisting"}}
+        # No spurious upload — the client already had a file_id.
+        assert files.calls == []
+        # And the Responses translator no longer sees any inline data.
+        rparts = OpenAILLM._responses_content_parts("user", out[0]["content"])
+        assert {"type": "input_file", "file_id": "file-preexisting"} in rparts
+        assert not any("file_data" in p for p in rparts)
+
+    # -- Fix #3: whitespace/newline-wrapped base64 must decode --
+
+    def test_mime_wrapped_base64_decodes_cleanly(self, llm):
+        """``base64.encodebytes`` line-wraps every 76 chars; pretty-printed
+        JSON producers may inject ``\\n``. With ``validate=True`` those
+        would raise, throwing recoverable payloads into the degrade path."""
+        files = _CountingFiles()
+        llm.client.files = files
+        wrapped_b64 = base64.encodebytes(_TINY_PDF_BYTES).decode()  # ends with \n
+        assert "\n" in wrapped_b64
+        out = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "filename": "wrapped.pdf",
+                    "file_data": f"data:application/pdf;base64,{wrapped_b64}",
+                }
+            )
+        )
+        assert out[0]["content"][1] == {"type": "file", "file": {"file_id": "file-1"}}
+        ((uploaded, _),) = files.calls
+        # The decoded bytes must equal the ORIGINAL — no whitespace bleed.
+        assert uploaded[1].read() == _TINY_PDF_BYTES
+
+    # -- Fix #4: no-comma data URI must degrade, not upload zero bytes --
+
+    def test_data_uri_missing_comma_degrades_without_upload(self, llm):
+        """``"data:application/pdf;base64".partition(",")`` -> empty payload;
+        the previous code decoded ``b""`` and shipped a zero-byte file."""
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg(
+                {
+                    "filename": "no-comma.pdf",
+                    "file_data": "data:application/pdf;base64",
+                }
+            )
+        )
+        part = out[0]["content"][1]
+        assert part["type"] == "text"
+        assert "no-comma.pdf" in part["text"]
+        assert files.calls == []  # no zero-byte artifact left in provider bucket
+
+    def test_data_uri_with_comma_but_empty_body_degrades(self, llm):
+        """Guard covers the near-neighbour: comma present, body empty."""
+        files = _CountingFiles()
+        llm.client.files = files
+        out = llm._clean_messages_openai(
+            self._msg({"filename": "empty.pdf", "file_data": "data:application/pdf;base64,"})
+        )
+        assert out[0]["content"][1]["type"] == "text"
+        assert files.calls == []
+
+    # -- Fix #2: Redis cache — cross-request dedup for /v1 replays --
+
+    def _patch_redis(self, cache):
+        """Patch ``application.cache.get_redis_instance`` to return the
+        provided fake redis (a dict-backed stub) — one context per test.
+
+        The helpers ``_inline_file_id_cache_*`` do ``from application.cache
+        import get_redis_instance`` INSIDE the function, so patching the
+        module attribute is enough — no import-time capture to worry about.
+        """
+        return patch("application.cache.get_redis_instance", return_value=cache)
+
+    class _FakeRedis:
+        def __init__(self):
+            self.store = {}
+            self.setex_calls = []
+            self.get_calls = []
+
+        def get(self, key):
+            self.get_calls.append(key)
+            return self.store.get(key)
+
+        def setex(self, key, ttl, value):
+            self.setex_calls.append((key, ttl, value))
+            self.store[key] = value.encode() if isinstance(value, str) else value
+
+    def test_redis_cache_hit_skips_upload_across_instances(self, llm):
+        """Two independent OpenAILLM instances (per-request creation is how
+        LLMCreator works) sharing the same credential must both see the
+        cached file_id after the first uploads it."""
+        cache = self._FakeRedis()
+        files_1 = _CountingFiles()
+        llm.client.files = files_1
+        msg = self._msg(
+            {"filename": "Scan-48.pdf",
+             "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}"}
+        )
+        with self._patch_redis(cache):
+            llm._clean_messages_openai(msg)
+        # First call: upload happened, cache populated with a real TTL.
+        assert len(files_1.calls) == 1
+        assert len(cache.setex_calls) == 1
+        key, ttl, val = cache.setex_calls[0]
+        assert key.startswith("openai_inline_file:") and ttl == 86400 and val == "file-1"
+
+        # Second call, second instance (simulates the next /v1 turn's
+        # freshly-constructed LLM), same credentials → cache-hit path.
+        llm2 = OpenAILLM(api_key="sk-test", user_api_key=None)
+        llm2.client = types.SimpleNamespace(files=_CountingFiles())
+        with self._patch_redis(cache):
+            out2 = llm2._clean_messages_openai(msg)
+        assert out2[0]["content"][1] == {"type": "file", "file": {"file_id": "file-1"}}
+        # No new upload — the second instance's Files stub is untouched.
+        assert llm2.client.files.calls == []
+
+    def test_redis_cache_key_isolates_by_credential(self, llm):
+        """A file_id is only valid on the endpoint+credential it was
+        uploaded to; the key must therefore not collide across credentials
+        (else a foundry-uploaded id would be handed to a byom endpoint)."""
+        cache = self._FakeRedis()
+        llm._effective_base_url = "https://foundry.example/openai/v1"
+        llm.api_key = "sk-foundry"
+        llm.client.files = _CountingFiles()
+        msg = self._msg(
+            {"filename": "x.pdf",
+             "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}"}
+        )
+        with self._patch_redis(cache):
+            llm._clean_messages_openai(msg)
+
+        llm2 = OpenAILLM(api_key="sk-byom", user_api_key=None)
+        llm2._effective_base_url = "https://byom.example/v1"
+        llm2.client = types.SimpleNamespace(files=_CountingFiles())
+        with self._patch_redis(cache):
+            llm2._clean_messages_openai(msg)
+        # Second credential missed the cache → uploaded again.
+        assert len(llm2.client.files.calls) == 1
+        # Two distinct cache keys landed in Redis, one per credential.
+        assert len(cache.setex_calls) == 2
+        assert cache.setex_calls[0][0] != cache.setex_calls[1][0]
+
+    def test_redis_unreachable_falls_back_to_upload(self, llm):
+        """If Redis is down, ``get_redis_instance`` returns None — the
+        code must upload normally and never raise."""
+        llm.client.files = _CountingFiles()
+        msg = self._msg(
+            {"filename": "x.pdf",
+             "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}"}
+        )
+        with self._patch_redis(None):
+            out = llm._clean_messages_openai(msg)
+        assert out[0]["content"][1] == {"type": "file", "file": {"file_id": "file-1"}}
+        assert len(llm.client.files.calls) == 1
+
+    def test_redis_read_exception_swallowed(self, llm):
+        """A transient Redis error mid-request must not surface — degrade
+        to an upload silently."""
+
+        class BrokenRedis:
+            def get(self, key):
+                raise RuntimeError("connection reset")
+
+            def setex(self, key, ttl, value):
+                raise RuntimeError("connection reset")
+
+        llm.client.files = _CountingFiles()
+        msg = self._msg(
+            {"filename": "x.pdf",
+             "file_data": f"data:application/pdf;base64,{_TINY_PDF_B64}"}
+        )
+        with self._patch_redis(BrokenRedis()):
+            out = llm._clean_messages_openai(msg)
+        # Upload happened because the read errored → treated as miss.
+        assert len(llm.client.files.calls) == 1
+        assert out[0]["content"][1] == {"type": "file", "file": {"file_id": "file-1"}}