diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 04cca5f06..5d0ab434d 100644 --- a/backend/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index ab42770b0..206fd50bf 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/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]: diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index cac9b1c57..8c2f63007 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/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 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 "" in injected_content + assert "prior-report.txt" in injected_content + assert "previous messages" in injected_content # historical section header