fix(uploads): keep malformed metadata from aborting chat turns (#6882)

* fix(uploads): keep malformed metadata from aborting chat turns

* fix(uploads): log skipped probes and bound size labels

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2417,6 +2417,14 @@ AIO directory listings discard missing shell sessions so the next request can re
 After a dropped connection, directory listings and persistent shell commands report an
 unknown outcome without replaying the operation; later calls use a fresh session.
 
+Current-upload context tolerates malformed attachment metadata: invalid filename
+references and files that cannot be checked are skipped individually, while valid
+attachments in the same message remain available. Unusable or display-overflowing
+file sizes use the existing zero-size display fallback without dropping the file.
+Probe errors are logged at debug level. Sizes of 1 TB or more that can be rendered
+safely display as `>= 1 TB` to keep the prompt compact; stored byte counts remain
+unchanged.
+
 Uploaded Markdown outlines recognize ATX heading syntax, clean closing markers with a linear suffix scan, and skip HTML comment blocks and fenced and indented code examples, so hashtags and code comments do not
 crowd out real document sections from the agent's heading preview. Indented bold examples are also excluded; PDF-style bold headings with up to three leading spaces remain supported.
 Split-bold numeric table rows, including parenthesized years, signed values, and

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -74,7 +74,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
 4. **PiiRedactionMiddleware** - *(optional, `pii_redaction.enabled`, default off, #3190)* Rewrites PII in genuine user messages (`wrap_model_call`) and remote-content tool results (`wrap_tool_call`, the tool-result allowlist) to irreversible value-derived placeholders. Deterministic regex detectors only. Innermost Layer-1 wrapper; compaction input, summaries and title input use `redact_text`. Memory admission redacts content, parsed/invalid/legacy call arguments and supported provenance before backend filtering.
 
 5. **ThreadDataMiddleware** - Creates per-thread directories under the user's isolation scope (`backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/{workspace,uploads,outputs}`); resolves identity via `resolve_runtime_user_id(runtime)`, including Gateway runtime context and standalone LangGraph Server auth, then falls back to the request ContextVar / `"default"`
-6. **UploadsMiddleware** - Injects current uploads (lead only). Check existence in the runtime user's bucket; coerce invalid `files[*].size` to `0`. Copy the user message when adding context to preserve caller metadata, including `response_metadata`.
+6. **UploadsMiddleware** - Lead-only current uploads: exact basenames in runtime owner's bucket; skip malformed references/probe errors (debug-log errors). Invalid/display-overflowing sizes use `0`; render >=1 TB compactly without changing stored sizes. Copy messages with metadata, including `response_metadata`.
 7. **SandboxMiddleware** - Acquires sandbox, stores `sandbox_id` in state. The
    lead runtime normally owns the thread's physical Agent-skill projection;
    delegated subagents and the prompt-only bootstrap agent are non-owners, so

backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py

@@ -5,6 +5,7 @@ on demand via the ``list_uploaded_files`` tool.
 """
 
 import logging
+import math
 from collections import Counter
 from pathlib import Path
 from typing import NotRequired, override
@@ -25,6 +26,7 @@ from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_t
 logger = logging.getLogger(__name__)
 
 _MAX_FILES_PER_CONTEXT_SECTION = 10
+_MAX_DISPLAY_FILE_SIZE_BYTES = 1024**4
 
 
 def _extension_label(file: dict) -> str:
@@ -84,8 +86,11 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         document cannot embed blocked authority tags inside the trusted
         ``<current_uploads>`` wrapper.
         """
-        size_kb = file["size"] / 1024
-        size_str = f"{size_kb:.1f} KB" if size_kb < 1024 else f"{size_kb / 1024:.1f} MB"
+        if file["size"] >= _MAX_DISPLAY_FILE_SIZE_BYTES:
+            size_str = ">= 1 TB"
+        else:
+            size_kb = file["size"] / 1024
+            size_str = f"{size_kb:.1f} KB" if size_kb < 1024 else f"{size_kb / 1024:.1f} MB"
         lines.append(f"- {neutralize_untrusted_tags(file['filename'])} ({size_str})")
         lines.append(f"  Path: {neutralize_untrusted_tags(file['path'])}")
         if file.get("selection_reason") == "query_match":
@@ -168,16 +173,20 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         and every other field in ``_files_from_kwargs`` is validated fail-soft,
         so an unusable size must never abort the run: anything that is not a
         non-negative finite number (``"abc"``, a list, a bool, ``-5``, ``inf``)
-        falls back to ``0``, the same value a missing size gets. Numeric
-        strings such as ``"2048"`` keep working.
+        falls back to ``0``, the same value a missing size gets. Values that
+        parse as integers but overflow the floating-point KB display also use
+        that fallback. Numeric strings such as ``"2048"`` keep working.
         """
         if value is None or isinstance(value, bool):
             return 0
         try:
             size = int(value)
+            if size < 0 or not math.isfinite(size / 1024):
+                return 0
         except (TypeError, ValueError, OverflowError):
+            # Both int(value) and the subsequent KB division can overflow.
             return 0
-        return size if size >= 0 else 0
+        return size
 
     def _files_from_kwargs(self, message: HumanMessage, uploads_dir: Path | None = None) -> list[dict] | None:
         """Extract file info from message additional_kwargs.files.
@@ -202,17 +211,26 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         for f in kwargs_files:
             if not isinstance(f, dict):
                 continue
-            filename = f.get("filename") or ""
-            if not filename or Path(filename).name != filename or is_upload_staging_file(filename):
+            filename = f.get("filename")
+            if not isinstance(filename, str) or not filename or "\x00" in filename:
                 continue
-            if uploads_dir is not None and not (uploads_dir / filename).is_file():
+            try:
+                file_path = Path(filename)
+                if file_path.name != filename or is_upload_staging_file(filename):
+                    continue
+                if uploads_dir is not None and not (uploads_dir / filename).is_file():
+                    continue
+            except (OSError, ValueError) as error:
+                # Invalid references and unavailable entries are per-file
+                # failures; they must not suppress valid sibling uploads.
+                logger.debug("Skipping uploaded file %r after %s during its filesystem probe: %s", filename, type(error).__name__, error)
                 continue
             files.append(
                 {
                     "filename": filename,
                     "size": self._coerce_file_size(f.get("size")),
                     "path": f"/mnt/user-data/uploads/{filename}",
-                    "extension": Path(filename).suffix,
+                    "extension": file_path.suffix,
                 }
             )
         return files if files else None

backend/tests/blocking_io/test_uploads_middleware.py

@@ -35,7 +35,16 @@ class _FakeModel(FakeMessagesListChatModel):
         return self
 
 
-async def test_before_agent_uploads_scan_does_not_block_event_loop(tmp_path: Path) -> None:
+@pytest.mark.parametrize(
+    "files,size,size_label",
+    [
+        (None, None, None),
+        ([{"filename": 42}, {"filename": "existing.txt", "size": "1" + "0" * 400}], 0, "0.0 KB"),
+        ([{"filename": "existing.txt", "size": 10**308}], 10**308, ">= 1 TB"),
+    ],
+    ids=["no-current-files", "malformed-with-valid-sibling", "huge-finite-size"],
+)
+async def test_before_agent_uploads_scan_does_not_block_event_loop(tmp_path: Path, files, size, size_label) -> None:
     from langchain.agents import create_agent
 
     from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
@@ -49,8 +58,12 @@ async def test_before_agent_uploads_scan_does_not_block_event_loop(tmp_path: Pat
     agent = await asyncio.to_thread(lambda: create_agent(model=_FakeModel(responses=[AIMessage(content="ok")]), tools=[], middleware=[mw]))
 
     result = await agent.ainvoke(
-        {"messages": [HumanMessage(content="hi")]},
+        {"messages": [HumanMessage(content="hi", additional_kwargs={"files": files})]},
         {"configurable": {"thread_id": "t1"}},
     )
 
     assert result["messages"]
+    if files:
+        assert [file["filename"] for file in result["uploaded_files"]] == ["existing.txt"]
+        assert result["uploaded_files"][0]["size"] == size
+        assert f"existing.txt ({size_label})" in result["messages"][0].content

backend/tests/test_upload_metadata_admission.py

@@ -0,0 +1,202 @@
+"""Malformed client upload metadata must not abort a valid chat turn."""
+
+import asyncio
+import logging
+import os
+from copy import deepcopy
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
+from pydantic import Field
+
+from app.gateway.services import _normalize_input_messages
+from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
+from deerflow.tools.builtins.list_uploaded_files_tool import _list_uploaded_files_impl
+from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text
+
+
+@pytest.fixture
+def upload_context(tmp_path):
+    middleware = UploadsMiddleware(base_dir=str(tmp_path))
+    context = {"thread_id": "upload-thread", "user_id": "alice"}
+    directory = middleware._paths.sandbox_uploads_dir(**context)
+    directory.mkdir(parents=True)
+    (directory / "current.md").write_text("# Current document\n\nNew evidence", encoding="utf-8")
+    (directory / "historical.md").write_text("# Historical document", encoding="utf-8")
+    return middleware, directory, context
+
+
+def _gateway_message(files):
+    return _normalize_input_messages(
+        [{"role": "user", "content": "Read the attachment", "additional_kwargs": {"files": files}}],
+        location="input.messages",
+    )[0]
+
+
+@pytest.mark.parametrize("filename", [True, 123, 1.5, ["current.md"], {"name": "current.md"}])
+def test_gateway_metadata_keeps_valid_siblings_when_a_filename_is_not_text(upload_context, filename):
+    middleware, directory, context = upload_context
+    files = [{"filename": filename}, {"filename": "current.md", "size": 24}]
+    original = deepcopy(files)
+    message = _gateway_message(files)
+    assert message.additional_kwargs["files"] == files
+
+    result = middleware.before_agent({"messages": [message]}, SimpleNamespace(context=context))
+
+    assert [file["filename"] for file in result["uploaded_files"]] == ["current.md"]
+    assert "# Current document" not in message.content
+    assert "Current document" in result["messages"][-1].content
+    assert message.additional_kwargs["files"] == original
+    assert result["messages"][-1].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "Read the attachment"
+
+
+@pytest.mark.parametrize("filename", ["bad\x00.md", "../current.md", "/current.md", ".upload-active.part"])
+@pytest.mark.parametrize("check_disk", [True, False])
+def test_unusable_filename_is_skipped_before_filesystem_access(upload_context, monkeypatch, filename, check_disk):
+    middleware, directory, _context = upload_context
+    original_is_file = Path.is_file
+
+    def guarded_is_file(path):
+        if path.name != "current.md":
+            pytest.fail("an unusable reference must not probe the filesystem")
+        return original_is_file(path)
+
+    monkeypatch.setattr(Path, "is_file", guarded_is_file)
+    files = [{"filename": filename}, {"filename": "current.md"}]
+    result = middleware._files_from_kwargs(_gateway_message(files), directory if check_disk else None)
+    assert [file["filename"] for file in result] == ["current.md"]
+
+
+@pytest.mark.parametrize("size", [10**400, "1" + "0" * 400, 10**4000], ids=["integer", "numeric-string", "long-integer"])
+def test_parseable_size_cannot_overflow_context_rendering(upload_context, size):
+    middleware, _directory, context = upload_context
+    message = _gateway_message([{"filename": "current.md", "size": size}])
+
+    result = middleware.before_agent({"messages": [message]}, SimpleNamespace(context=context))
+
+    assert result["uploaded_files"][0]["size"] == 0
+    assert "current.md (0.0 KB)" in result["messages"][-1].content
+    assert "Current document" in result["messages"][-1].content
+
+
+@pytest.mark.parametrize("size", [1024**4, 1024**4 + 1, 10**308], ids=["display-boundary", "above-boundary", "huge-finite"])
+def test_large_renderable_size_keeps_metadata_with_a_bounded_display(upload_context, size):
+    middleware, _directory, context = upload_context
+    message = _gateway_message([{"filename": "current.md", "size": size}])
+    result = middleware.before_agent({"messages": [message]}, SimpleNamespace(context=context))
+    assert result["uploaded_files"][0]["size"] == size
+    assert "current.md (>= 1 TB)" in result["messages"][-1].content
+    assert message.additional_kwargs["files"][0]["size"] == size
+
+
+@pytest.mark.parametrize("error", [PermissionError("unreadable"), OSError("file unavailable"), ValueError("invalid path")])
+def test_file_probe_failure_does_not_drop_a_valid_sibling(upload_context, monkeypatch, caplog, error):
+    middleware, directory, context = upload_context
+    original_is_file = Path.is_file
+
+    def unavailable(path):
+        if path.name == "unavailable.md":
+            raise error
+        return original_is_file(path)
+
+    monkeypatch.setattr(Path, "is_file", unavailable)
+    message = _gateway_message([{"filename": "unavailable.md"}, {"filename": "current.md"}])
+    with caplog.at_level(logging.DEBUG, logger="deerflow.agents.middlewares.uploads_middleware"):
+        result = middleware.before_agent({"messages": [message]}, SimpleNamespace(context=context))
+    assert [file["filename"] for file in result["uploaded_files"]] == ["current.md"]
+    assert "unavailable.md" not in result["messages"][-1].content
+    dropped = [record for record in caplog.records if "unavailable.md" in record.getMessage()]
+    assert len(dropped) == 1
+    assert dropped[0].levelno == logging.DEBUG
+    assert type(error).__name__ in dropped[0].getMessage()
+    assert str(error) in dropped[0].getMessage()
+
+
+def test_unexpected_probe_error_is_not_silenced(upload_context, monkeypatch):
+    middleware, directory, _context = upload_context
+
+    def unexpected(_path):
+        raise RuntimeError("implementation bug")
+
+    monkeypatch.setattr(Path, "is_file", unexpected)
+    with pytest.raises(RuntimeError, match="implementation bug"):
+        middleware._files_from_kwargs(_gateway_message([{"filename": "current.md"}]), directory)
+
+
+def test_all_invalid_metadata_clears_stale_current_uploads(upload_context):
+    middleware, _directory, context = upload_context
+    message = _gateway_message([{"filename": True}, {"filename": "bad\x00.md"}])
+    state = {"messages": [message], "uploaded_files": [{"filename": "historical.md"}]}
+    assert middleware.before_agent(state, SimpleNamespace(context=context)) == {"uploaded_files": []}
+    assert message.content == "Read the attachment"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Names legal on POSIX but not Windows")
+@pytest.mark.parametrize("filename", ["CON.md", "report.md "])
+def test_existing_posix_upload_names_are_not_rewritten(upload_context, filename):
+    middleware, directory, _context = upload_context
+    (directory / filename).write_text("Existing upload", encoding="utf-8")
+    result = middleware._files_from_kwargs(_gateway_message([{"filename": filename}]), directory)
+    assert result[0]["filename"] == filename
+    assert result[0]["path"] == f"/mnt/user-data/uploads/{filename}"
+
+
+def test_historical_discovery_only_excludes_admitted_current_files(upload_context):
+    middleware, _directory, context = upload_context
+    message = _gateway_message([{"filename": {"name": "historical.md"}}, {"filename": "current.md"}])
+    update = middleware.before_agent({"messages": [message]}, SimpleNamespace(context=context))
+    runtime = SimpleNamespace(context=context, state=update)
+    listing = _list_uploaded_files_impl(runtime=runtime, _paths=middleware._paths)
+    assert [file["filename"] for file in listing["files"]] == ["historical.md"]
+    runtime.state = middleware.before_agent({"messages": [HumanMessage(content="Continue")]}, runtime)
+    listing = _list_uploaded_files_impl(runtime=runtime, _paths=middleware._paths)
+    assert {file["filename"] for file in listing["files"]} == {"current.md", "historical.md"}
+
+
+class _UploadModel(FakeMessagesListChatModel):
+    seen_messages: list[list[BaseMessage]] = Field(default_factory=list)
+
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        self.seen_messages.append(deepcopy(messages))
+        return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invoke_async", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize("size,size_label", [("1" + "0" * 400, "0.0 KB"), (10**308, ">= 1 TB")], ids=["overflowing", "huge-finite"])
+async def test_agent_graph_consumes_only_valid_owner_upload_context(upload_context, invoke_async, size, size_label):
+    middleware, _directory, context = upload_context
+    other_directory = middleware._paths.sandbox_uploads_dir("upload-thread", user_id="bob")
+    other_directory.mkdir(parents=True)
+    (other_directory / "other-owner.md").write_text("# Private other-owner evidence", encoding="utf-8")
+    model = _UploadModel(responses=[AIMessage(content="Processed")])
+    agent = create_agent(model=model, tools=[], middleware=[middleware])
+    message = _gateway_message(
+        [{"filename": 42}, {"filename": "current.md", "size": size}, {"filename": "other-owner.md"}],
+    )
+
+    state = {"messages": [message]}
+    config = {"configurable": {"thread_id": "upload-thread"}}
+    if invoke_async:
+        result = await agent.ainvoke(state, config, context=context)
+    else:
+        result = await asyncio.to_thread(agent.invoke, state, config, context=context)
+
+    assert result["messages"][-1].content == "Processed"
+    assert [file["filename"] for file in result["uploaded_files"]] == ["current.md"]
+    rendered = message_content_to_text(result["messages"][0].content)
+    assert "Current document" in rendered
+    assert "Private other-owner evidence" not in rendered
+    assert len(model.seen_messages) == 1
+    model_input = message_content_to_text(model.seen_messages[0][0].content)
+    assert f"current.md ({size_label})" in model_input
+    assert "Current document" in model_input
+    assert "Private other-owner evidence" not in model_input
+    assert message.content == "Read the attachment"