fix: handle id-less values messages and redact stream payload logs

Author: zouyonghe

astrbot/core/agent/runners/deerflow/deerflow_agent_runner.py

@@ -1,4 +1,6 @@
 import asyncio
+import hashlib
+import json
 import sys
 import typing as T
 from collections import deque
@@ -49,6 +51,8 @@ class _StreamState:
         task_failures: list[str] = field(default_factory=list)
         seen_message_ids: set[str] = field(default_factory=set)
         seen_message_order: deque[str] = field(default_factory=deque)
+        # Fallback tracking for backends that omit message ids in values events.
+        no_id_message_fingerprints: dict[int, str] = field(default_factory=dict)
         baseline_initialized: bool = False
         run_values_messages: list[dict[str, T.Any]] = field(default_factory=list)
         timed_out: bool = False
@@ -267,16 +271,38 @@ def _extract_new_messages_from_values(
         state: _StreamState,
     ) -> list[dict[str, T.Any]]:
         new_messages: list[dict[str, T.Any]] = []
-        for msg in values_messages:
+        no_id_indexes_seen: set[int] = set()
+        for idx, msg in enumerate(values_messages):
             if not isinstance(msg, dict):
                 continue
             msg_id = get_message_id(msg)
-            if not msg_id or msg_id in state.seen_message_ids:
+            if msg_id:
+                if msg_id in state.seen_message_ids:
+                    continue
+                self._remember_seen_message_id(state, msg_id)
+                new_messages.append(msg)
                 continue
-            self._remember_seen_message_id(state, msg_id)
+
+            no_id_indexes_seen.add(idx)
+            msg_fingerprint = self._fingerprint_message(msg)
+            if state.no_id_message_fingerprints.get(idx) == msg_fingerprint:
+                continue
+            state.no_id_message_fingerprints[idx] = msg_fingerprint
             new_messages.append(msg)
+
+        # Keep no-id index state aligned with latest values payload shape.
+        for idx in list(state.no_id_message_fingerprints.keys()):
+            if idx not in no_id_indexes_seen:
+                state.no_id_message_fingerprints.pop(idx, None)
         return new_messages
 
+    def _fingerprint_message(self, message: dict[str, T.Any]) -> str:
+        try:
+            raw = json.dumps(message, sort_keys=True, ensure_ascii=False, default=str)
+        except (TypeError, ValueError):
+            raw = repr(message)
+        return hashlib.sha1(raw.encode("utf-8", errors="ignore")).hexdigest()
+
     def _remember_seen_message_id(self, state: _StreamState, msg_id: str) -> None:
         if not msg_id or msg_id in state.seen_message_ids:
             return
@@ -422,9 +448,14 @@ def _handle_values_event(
 
         if not state.baseline_initialized:
             state.baseline_initialized = True
-            for msg in values_messages:
+            for idx, msg in enumerate(values_messages):
+                if not isinstance(msg, dict):
+                    continue
                 msg_id = get_message_id(msg)
-                self._remember_seen_message_id(state, msg_id)
+                if msg_id:
+                    self._remember_seen_message_id(state, msg_id)
+                    continue
+                state.no_id_message_fingerprints[idx] = self._fingerprint_message(msg)
             return responses
 
         new_messages = self._extract_new_messages_from_values(

astrbot/core/agent/runners/deerflow/deerflow_api_client.py

@@ -149,7 +149,20 @@ async def stream_run(
     ) -> AsyncGenerator[dict[str, Any], None]:
         session = self._get_session()
         url = f"{self.api_base}/api/langgraph/threads/{thread_id}/runs/stream"
-        logger.debug(f"deerflow stream_run payload: {payload}")
+        input_payload = payload.get("input")
+        message_count = 0
+        if isinstance(input_payload, dict) and isinstance(
+            input_payload.get("messages"), list
+        ):
+            message_count = len(input_payload["messages"])
+        # Log only a minimal summary to avoid exposing sensitive user content.
+        logger.debug(
+            "deerflow stream_run payload summary: thread_id=%s, keys=%s, message_count=%d, stream_mode=%s",
+            thread_id,
+            list(payload.keys()),
+            message_count,
+            payload.get("stream_mode"),
+        )
         # For long-running SSE streams, avoid aiohttp total timeout.
         # Use socket read timeout so active heartbeats/chunks can keep the stream alive.
         stream_timeout = ClientTimeout(