refactor(middlewares): declarative layered builder; pin chain order with tests (#3809)

Split `_build_runtime_middlewares`'s flat list into three named declarative
sublists (outer_wrappers / thread_hooks / tail) and drop the
`middlewares.insert(2, UploadsMiddleware())` magic-index pattern. The
declarative structure makes the layering self-documenting and immune to
position drift when the head of the list changes.

Move UploadsMiddleware to run after ThreadDataMiddleware in the chain.
Under the previous order (a magic-index artifact introduced when #3662
prepended InputSanitizationMiddleware), UploadsMiddleware scanned the
uploads directory before ThreadDataMiddleware created it under
lazy_init=False, so historical files could be missed on the first run of
a thread. Narrow path — the upload endpoint normally pre-creates the
directory — but the order is the correct semantic and is now locked.

Documentation:
- backend/AGENTS.md middleware chain renumbered: ThreadDataMiddleware is
  now #3, UploadsMiddleware #4 (was reversed).

Tests (backend/tests/test_tool_error_handling_middleware.py):
- test_build_lead_runtime_middlewares_orders_thread_data_before_uploads
  — focused td_idx < um_idx assertion.
- test_build_lead_runtime_middlewares_chain_order_matches_agents_md
  — full-chain order pin using real classes, so a swap between any pair
    is caught (the existing FakeMiddleware-stubbed tests cannot detect
    this).
- test_lead_runtime_chain_finds_historical_uploads_under_lazy_init_false
  — integration anchor: under lazy_init=False, ThreadDataMiddleware
    creates the dirs, then UploadsMiddleware surfaces a pre-existing
    historical file in the injected <uploaded_files> context.

backend/AGENTS.md

@@ -207,8 +207,8 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 
 1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages
 2. **ToolOutputBudgetMiddleware** - Caps tool output size (per app config) before it re-enters the model context
-3. **UploadsMiddleware** - Tracks and injects newly uploaded files into conversation (lead agent only)
-4. **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 `user_id` via `get_effective_user_id()` (falls back to `"default"` in no-auth mode)
+3. **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 `user_id` via `get_effective_user_id()` (falls back to `"default"` in no-auth mode)
+4. **UploadsMiddleware** - Tracks and injects newly uploaded files into conversation (lead agent only)
 5. **SandboxMiddleware** - Acquires sandbox, stores `sandbox_id` in state
 6. **DanglingToolCallMiddleware** - Injects placeholder ToolMessages for AIMessage tool_calls that lack responses (e.g., user interruption), preserving raw provider tool-call payloads in `additional_kwargs["tool_calls"]`
 7. **LLMErrorHandlingMiddleware** - Normalizes provider/model invocation failures into recoverable assistant-facing errors before later stages run

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

@@ -140,27 +140,31 @@ def _build_runtime_middlewares(
     from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware
     from deerflow.sandbox.middleware import SandboxMiddleware
 
+    # Layer 1 — outermost wrap_model_call wrappers (listed outer→inner).
     # InputSanitizationMiddleware is first so it becomes the outermost
-    # wrap_model_call wrapper — sanitised messages are what every inner
-    # middleware (including LLMErrorHandlingMiddleware retries) sees.
-    middlewares: list[AgentMiddleware] = [
+    # wrapper — sanitised messages are what every inner middleware sees.
+    outer_wrappers: list[AgentMiddleware] = [
         InputSanitizationMiddleware(),
         ToolOutputBudgetMiddleware.from_app_config(app_config),
-        ThreadDataMiddleware(lazy_init=lazy_init),
-        SandboxMiddleware(lazy_init=lazy_init),
     ]
 
+    # Layer 2 — before_agent hooks that read/annotate thread-scoped data.
+    thread_hooks: list[AgentMiddleware] = [
+        ThreadDataMiddleware(lazy_init=lazy_init),
+    ]
     if include_uploads:
         from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
 
-        middlewares.insert(2, UploadsMiddleware())
+        thread_hooks.append(UploadsMiddleware())
+    thread_hooks.append(SandboxMiddleware(lazy_init=lazy_init))
 
+    # Layer 3 — post-processing append-only middlewares.
+    tail: list[AgentMiddleware] = []
     if include_dangling_tool_call_patch:
         from deerflow.agents.middlewares.dangling_tool_call_middleware import DanglingToolCallMiddleware
 
-        middlewares.append(DanglingToolCallMiddleware())
-
-    middlewares.append(LLMErrorHandlingMiddleware(app_config=app_config))
+        tail.append(DanglingToolCallMiddleware())
+    tail.append(LLMErrorHandlingMiddleware(app_config=app_config))
 
     # Guardrail middleware (if configured)
     guardrails_config = app_config.guardrails
@@ -183,13 +187,14 @@ def _build_runtime_middlewares(
             except (ValueError, TypeError):
                 pass
         provider = provider_cls(**provider_kwargs)
-        middlewares.append(GuardrailMiddleware(provider, fail_closed=guardrails_config.fail_closed, passport=guardrails_config.passport))
+        tail.append(GuardrailMiddleware(provider, fail_closed=guardrails_config.fail_closed, passport=guardrails_config.passport))
 
     from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware
 
-    middlewares.append(SandboxAuditMiddleware())
-    middlewares.append(ToolErrorHandlingMiddleware())
-    return middlewares
+    tail.append(SandboxAuditMiddleware())
+    tail.append(ToolErrorHandlingMiddleware())
+
+    return [*outer_wrappers, *thread_hooks, *tail]
 
 
 def build_lead_runtime_middlewares(*, app_config: AppConfig, lazy_init: bool = True) -> list[AgentMiddleware]:

backend/tests/test_tool_error_handling_middleware.py

@@ -7,6 +7,7 @@ from langgraph.errors import GraphInterrupt
 
 from deerflow.agents.middlewares.tool_error_handling_middleware import (
     ToolErrorHandlingMiddleware,
+    build_lead_runtime_middlewares,
     build_subagent_runtime_middlewares,
 )
 from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
@@ -152,6 +153,73 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
     assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware)
 
 
+def test_build_lead_runtime_middlewares_orders_thread_data_before_uploads():
+    """ThreadDataMiddleware must run before UploadsMiddleware so the uploads
+    directory is guaranteed to exist when UploadsMiddleware scans it under
+    lazy_init=False. This is the narrow functional concern the chain order
+    protects; a regression here would silently drop historical files on the
+    first run of a thread when the directory has not been pre-created by the
+    upload endpoint.
+    """
+    from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware
+    from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
+
+    app_config = _make_app_config()
+    middlewares = build_lead_runtime_middlewares(app_config=app_config)
+
+    td_indices = [i for i, m in enumerate(middlewares) if isinstance(m, ThreadDataMiddleware)]
+    um_indices = [i for i, m in enumerate(middlewares) if isinstance(m, UploadsMiddleware)]
+
+    assert td_indices and len(td_indices) == 1, f"expected exactly one ThreadDataMiddleware, got {td_indices}"
+    assert um_indices and len(um_indices) == 1, f"expected exactly one UploadsMiddleware, got {um_indices}"
+    assert td_indices[0] < um_indices[0], f"ThreadDataMiddleware (idx {td_indices[0]}) must come before UploadsMiddleware (idx {um_indices[0]}) so the uploads directory exists when UploadsMiddleware scans it under lazy_init=False."
+
+
+def test_build_lead_runtime_middlewares_chain_order_matches_agents_md():
+    """Pin the AGENTS.md middleware numbering for the shared runtime base.
+
+    The existing tests stub most middlewares as a single ``FakeMiddleware``,
+    which cannot detect a reorder. This test uses the real classes so an
+    index swap between any pair (e.g. Uploads vs ThreadData, Sandbox vs
+    DanglingToolCall) is caught. If a future refactor legitimately reorders
+    these, update backend/AGENTS.md "Middleware Chain" in the same change.
+    """
+    from deerflow.agents.middlewares.dangling_tool_call_middleware import DanglingToolCallMiddleware
+    from deerflow.agents.middlewares.input_sanitization_middleware import InputSanitizationMiddleware
+    from deerflow.agents.middlewares.llm_error_handling_middleware import LLMErrorHandlingMiddleware
+    from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware
+    from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware
+    from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware
+    from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
+    from deerflow.sandbox.middleware import SandboxMiddleware
+
+    app_config = _make_app_config()
+    middlewares = build_lead_runtime_middlewares(app_config=app_config)
+
+    def idx_of(cls, *, label: str) -> int:
+        matches = [i for i, m in enumerate(middlewares) if isinstance(m, cls)]
+        assert matches, f"{label} missing from chain"
+        assert len(matches) == 1, f"expected exactly one {label}, got indices {matches}"
+        return matches[0]
+
+    # Mirrors AGENTS.md "Shared runtime base" items 1-10 (non-optional spine).
+    expected_order: list[tuple[str, type]] = [
+        ("InputSanitizationMiddleware", InputSanitizationMiddleware),
+        ("ToolOutputBudgetMiddleware", ToolOutputBudgetMiddleware),
+        ("ThreadDataMiddleware", ThreadDataMiddleware),
+        ("UploadsMiddleware", UploadsMiddleware),
+        ("SandboxMiddleware", SandboxMiddleware),
+        ("DanglingToolCallMiddleware", DanglingToolCallMiddleware),
+        ("LLMErrorHandlingMiddleware", LLMErrorHandlingMiddleware),
+        ("SandboxAuditMiddleware", SandboxAuditMiddleware),
+        ("ToolErrorHandlingMiddleware", ToolErrorHandlingMiddleware),
+    ]
+    actual = [(label, idx_of(cls, label=label)) for label, cls in expected_order]
+
+    for (name_a, idx_a), (name_b, idx_b) in zip(actual, actual[1:]):
+        assert idx_a < idx_b, f"{name_a} (idx {idx_a}) must come before {name_b} (idx {idx_b}); full chain: {actual}"
+
+
 def test_wrap_tool_call_passthrough_on_success():
     middleware = ToolErrorHandlingMiddleware()
     req = _request()
@@ -301,3 +369,57 @@ def test_subagent_runtime_middlewares_skip_deferred_filter_without_names(monkeyp
     for setup in (None, DeferredToolSetup(None, frozenset(), None)):
         middlewares = build_subagent_runtime_middlewares(app_config=app_config, deferred_setup=setup)
         assert not any(isinstance(m, DeferredToolFilterMiddleware) for m in middlewares)
+
+
+def test_lead_runtime_chain_finds_historical_uploads_under_lazy_init_false(tmp_path, monkeypatch):
+    """Integration anchor for the ThreadData → Uploads ordering.
+
+    Under lazy_init=False, ThreadDataMiddleware eagerly creates the thread
+    directories in before_agent. UploadsMiddleware then scans the uploads
+    directory. Running both middlewares via the real build_lead_runtime_middlewares
+    chain (TD before UM) must surface pre-existing historical files in the
+    injected <uploaded_files> context.
+
+    This complements the static order contract
+    (test_build_lead_runtime_middlewares_orders_thread_data_before_uploads):
+    that test pins the chain position; this test pins the observable behavior
+    at that position.
+    """
+    from langchain_core.messages import HumanMessage
+    from langgraph.runtime import Runtime
+
+    from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware
+    from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
+    from deerflow.config.paths import Paths
+    from deerflow.runtime.user_context import get_effective_user_id
+
+    thread_id = "thread-historical-files"
+    user_id = get_effective_user_id()
+
+    paths = Paths(str(tmp_path))
+    uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=user_id)
+    uploads_dir.mkdir(parents=True, exist_ok=True)
+    (uploads_dir / "prior-report.txt").write_bytes(b"historical payload")
+
+    td = ThreadDataMiddleware(base_dir=str(tmp_path), lazy_init=False)
+    um = UploadsMiddleware(base_dir=str(tmp_path))
+
+    runtime = Runtime(context={"thread_id": thread_id, "run_id": "run-1"})
+    state = {"messages": [HumanMessage(content="please summarise the prior upload")]}
+
+    td_result = td.before_agent(state, runtime)
+    assert td_result is not None, "ThreadDataMiddleware must run and produce state updates"
+    # Sanity: under lazy_init=False the directories were created (not just computed).
+    assert uploads_dir.exists(), "ThreadDataMiddleware should have ensured the uploads directory exists"
+
+    # ThreadDataMiddleware rewrites the last HumanMessage (annotating run_id/timestamp);
+    # carry its updated messages into the UploadsMiddleware input state, mirroring
+    # how LangGraph chains before_agent outputs into the next middleware.
+    um_input = {**state, "messages": td_result["messages"]}
+    um_result = um.before_agent(um_input, runtime)
+
+    assert um_result is not None, "UploadsMiddleware must inject context when historical files exist"
+    injected_content = um_result["messages"][-1].content
+    assert "<uploaded_files>" in injected_content
+    assert "prior-report.txt" in injected_content
+    assert "previous messages" in injected_content  # historical section header