fix(memory): exclude invalid and raw assistant tool calls (#6379)

README.md

@@ -2221,6 +2221,10 @@ behavior, and current boundaries.
 
 Across sessions, DeerFlow builds a persistent memory of your profile, preferences, and accumulated knowledge. The more you use it, the better it knows you — your writing style, your technical stack, your recurring workflows. Memory is stored locally and stays under your control.
 
+DeerMem and mem0 capture user inputs and final assistant replies. Assistant
+turns that request tools, including malformed or provider-raw calls, are
+excluded from memory extraction even when they contain visible text.
+
 DeerMem remains the default local backend. An opt-in `mem0` backend is also
 available for the hosted mem0 Platform API or API-compatible self-hosted
 servers. Its token-bearing `base_url` must use HTTPS by default; plaintext HTTP

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

@@ -87,6 +87,12 @@ fact-only import payload supplies empty summary defaults.
 It captures `user_id` when it enqueues work.
 This identity survives the background timer boundary.
 
+DeerMem and mem0 exclude assistant tool-call intent in every representation:
+`tool_calls`, `invalid_tool_calls`, and provider-raw `tool_calls`/`function_call`.
+An empty parsed call list does not make an attempted call a final response.
+Keep both portable backend filters aligned without importing host helpers;
+test the real queue/HTTP write boundary in `tests/test_memory_tool_call_intent.py`.
+
 `memory.mode: tool` registers the four memory tools.
 The model chooses when to search or change facts.
 Tool mode still uses `MemoryMiddleware` for passive writes on supported remote backends.

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/message_processing.py

@@ -199,12 +199,14 @@ def filter_messages_for_memory(messages: list[Any], *, should_keep_hidden_messag
                 filtered.append(msg)
                 skip_next_ai = False
         elif msg_type == "ai":
-            tool_calls = getattr(msg, "tool_calls", None)
-            if not tool_calls:
-                if skip_next_ai:
-                    skip_next_ai = False
-                    continue
-                filtered.append(msg)
+            # Invalid/provider-raw calls are still tool turns, not final replies.
+            additional_kwargs = getattr(msg, "additional_kwargs", {}) or {}
+            if getattr(msg, "tool_calls", None) or getattr(msg, "invalid_tool_calls", None) or additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"):
+                continue
+            if skip_next_ai:
+                skip_next_ai = False
+                continue
+            filtered.append(msg)
 
     return filtered
 

backend/packages/harness/deerflow/agents/memory/backends/mem0/message_filtering.py

@@ -84,7 +84,9 @@ def filter_messages_for_memory(messages: list[Any]) -> list[Any]:
                 filtered.append(msg)
                 skip_next_ai = False
         elif msg_type == "ai":
-            if getattr(msg, "tool_calls", None):
+            # Invalid/provider-raw calls are still tool turns, not final replies.
+            additional_kwargs = getattr(msg, "additional_kwargs", {}) or {}
+            if getattr(msg, "tool_calls", None) or getattr(msg, "invalid_tool_calls", None) or additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"):
                 continue
             if skip_next_ai:
                 skip_next_ai = False

backend/tests/test_memory_tool_call_intent.py

@@ -0,0 +1,129 @@
+"""Tool-call assistant turns must not become final-response memory inputs."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, messages_from_dict, messages_to_dict
+from langchain_openai.chat_models.base import _convert_dict_to_message
+
+from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem
+from deerflow.agents.memory.backends.deermem.deermem.core.message_processing import filter_messages_for_memory as filter_deermem
+from deerflow.agents.memory.backends.mem0.client import Mem0Client
+from deerflow.agents.memory.backends.mem0.mem0_manager import Mem0Manager
+from deerflow.agents.memory.backends.mem0.message_filtering import filter_messages_for_memory as filter_mem0
+
+_ATTEMPT_TEXT = "I will save the nightly production deployment preference."
+_USER_TEXT = "Inspect the deployment settings before changing anything."
+_FINAL_TEXT = "The settings were inspected; no deployment preference was saved."
+
+
+def _assistant_attempt(kind: str) -> AIMessage:
+    if kind == "parsed":
+        return AIMessage(content=_ATTEMPT_TEXT, tool_calls=[{"name": "memory_add", "args": {}, "id": "call-1"}])
+    if kind == "invalid":
+        return AIMessage(content=_ATTEMPT_TEXT, invalid_tool_calls=[{"name": "memory_add", "args": "{broken", "id": "call-1"}])
+    if kind == "legacy":
+        return _convert_dict_to_message({"role": "assistant", "content": _ATTEMPT_TEXT, "function_call": {"name": "memory_add", "arguments": "{}"}})
+    raw = [{"id": "call-1", "type": "function", "function": {"name": "memory_add", "arguments": "{broken"}}]
+    if kind == "raw":
+        # Provider-raw fallback with neither structured view (e.g. an older checkpoint).
+        return AIMessage(content=_ATTEMPT_TEXT).model_copy(update={"additional_kwargs": {"tool_calls": raw}})
+    message = _convert_dict_to_message({"role": "assistant", "content": _ATTEMPT_TEXT, "tool_calls": raw})
+    assert isinstance(message, AIMessage)
+    assert not message.tool_calls and message.invalid_tool_calls
+    # Exercise the message serialization used when persisting/reloading history.
+    return messages_from_dict(messages_to_dict([message]))[0]
+
+
+@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"])
+@pytest.mark.parametrize("kind", ["parsed", "invalid", "provider-invalid", "raw", "legacy"])
+def test_filter_excludes_all_tool_call_views_without_mutating_history(filter_messages, kind: str) -> None:
+    attempt = _assistant_attempt(kind)
+    messages = [HumanMessage(content=_USER_TEXT), attempt, ToolMessage(content="Not executed", tool_call_id="call-1", status="error"), AIMessage(content=_FINAL_TEXT)]
+    original = messages_to_dict(messages)
+
+    kept = filter_messages(messages)
+
+    assert kept == [messages[0], messages[-1]]
+    assert kept[0] is messages[0] and kept[1] is messages[-1]
+    assert messages_to_dict(messages) == original
+
+
+@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"])
+@pytest.mark.parametrize("kwargs", [{}, {"tool_calls": []}, {"function_call": None}, {"tool_calls": [], "function_call": {}}])
+def test_empty_call_views_keep_final_response(filter_messages, kwargs: dict) -> None:
+    final = AIMessage(content=[{"type": "text", "text": _FINAL_TEXT}], additional_kwargs=kwargs)
+    assert filter_messages([final]) == [final]
+
+
+@pytest.mark.parametrize("filter_messages", [filter_deermem, filter_mem0], ids=["deermem", "mem0"])
+def test_invalid_attempt_does_not_consume_upload_ack_filter(filter_messages) -> None:
+    upload = HumanMessage(content="<current_uploads>\nsettings.yaml\n</current_uploads>")
+    attempt = _assistant_attempt("provider-invalid")
+    ack = AIMessage(content="I see the uploaded settings.")
+    user = HumanMessage(content=_USER_TEXT)
+    final = AIMessage(content=_FINAL_TEXT)
+    assert filter_messages([upload, attempt, ack, user, final]) == [user, final]
+
+
+@pytest.mark.parametrize("method", ["add", "add_nowait"])
+@pytest.mark.parametrize("include_final", [False, True])
+def test_deermem_queue_contains_only_user_and_final_response(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, method: str, include_final: bool) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
+    # Keep the real queue deterministic; do not launch background extraction.
+    monkeypatch.setattr(manager._queue, "_schedule_timer", lambda *args, **kwargs: None)
+    messages = [HumanMessage(content=_USER_TEXT), _assistant_attempt("provider-invalid")]
+    if include_final:
+        messages.append(AIMessage(content=_FINAL_TEXT))
+    original = messages_to_dict(messages)
+    try:
+        getattr(manager, method)("thread-1", messages, user_id="alice", agent_name="deployment-agent")
+        contexts = manager._queue._items
+        if include_final:
+            assert len(contexts) == 1
+            context = contexts[0]
+            assert context.messages == [messages[0], messages[-1]]
+            assert (context.thread_id, context.user_id, context.agent_name) == ("thread-1", "alice", "deployment-agent")
+        else:
+            # An attempted call is not the assistant reply required for extraction.
+            assert contexts == []
+        assert messages_to_dict(messages) == original
+    finally:
+        manager.cancel_by_agent("deployment-agent", user_id="alice")
+        manager.close()
+
+
+@pytest.mark.parametrize("method", ["add", "aadd"])
+@pytest.mark.parametrize("include_final", [False, True])
+def test_mem0_http_payload_excludes_invalid_attempt(monkeypatch: pytest.MonkeyPatch, method: str, include_final: bool) -> None:
+    monkeypatch.setenv("MEM0_API_KEY", "offline-test-key")
+    payloads: list[dict] = []
+
+    def record(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST" and request.url.path == "/v3/memories/add/"
+        payloads.append(json.loads(request.content))
+        return httpx.Response(200, json={"status": "PENDING"})
+
+    manager = Mem0Manager(backend_config={"startup_policy": "tolerate"})
+    manager.close()
+    manager._client = Mem0Client(base_url="https://mem0.test", api_key="offline-test-key", transport=httpx.MockTransport(record))
+    messages = [HumanMessage(content=_USER_TEXT), _assistant_attempt("provider-invalid")]
+    if include_final:
+        messages.append(AIMessage(content=_FINAL_TEXT))
+    original = messages_to_dict(messages)
+    try:
+        result = getattr(manager, method)("thread-1", messages, user_id="alice", agent_name="deployment-agent")
+        if method == "aadd":
+            asyncio.run(result)
+        expected = [{"role": "user", "content": _USER_TEXT}]
+        if include_final:
+            expected.append({"role": "assistant", "content": _FINAL_TEXT})
+        assert payloads == [{"messages": expected, "user_id": "alice", "agent_id": "deployment-agent", "run_id": "thread-1"}]
+        assert messages_to_dict(messages) == original
+    finally:
+        manager.close()