fix(conversation): tolerate malformed event metadata (#7721)

* fix(conversation): tolerate malformed event metadata

* fix(gateway): guard malformed event metadata

* fix(gateway): guard malformed message kwargs

* fix(gateway): log malformed transcript metadata

* fix(gateway): ignore malformed message roles

backend/app/channels/manager.py

@@ -599,7 +599,7 @@ def _is_run_status_result(result: Any) -> bool:
     return isinstance(result, dict) and "messages" not in result and isinstance(result.get("status"), str)
 
 
-def _turn_of_message(values: Any, message_id: str) -> dict[str, Any] | None:
+def _turn_of_message(values: Any, message_id: str, *, thread_id: str | None = None) -> dict[str, Any] | None:
     """Thread *values* cut after the turn that answered human message *message_id*.
 
     The cut ends before the next visible human message, so the response, artifact
@@ -611,7 +611,7 @@ def _turn_of_message(values: Any, message_id: str) -> dict[str, Any] | None:
     for index, message in enumerate(messages):
         if isinstance(message, dict) and message.get("type") == "human" and message.get("id") == message_id:
             end = next(
-                (later for later in range(index + 1, len(messages)) if isinstance(messages[later], dict) and messages[later].get("type") == "human" and not _is_hidden_human_control_message(messages[later])),
+                (later for later in range(index + 1, len(messages)) if isinstance(messages[later], dict) and messages[later].get("type") == "human" and not _is_hidden_human_control_message(messages[later], thread_id=thread_id)),
                 len(messages),
             )
             return {**values, "messages": messages[:end]}
@@ -810,7 +810,7 @@ def _apply_explicit_model_choice(
         carrier["model_name"] = model_name
 
 
-def _extract_response_text(result: dict | list) -> str:
+def _extract_response_text(result: dict | list, *, thread_id: str | None = None) -> str:
     """Extract the last AI message text from a LangGraph runs.wait result.
 
     ``runs.wait`` returns the final state dict which contains a ``messages``
@@ -837,7 +837,7 @@ def _extract_response_text(result: dict | list) -> str:
 
         # Stop at the last human message — anything before it is a previous turn
         if msg_type == "human":
-            if _is_hidden_human_control_message(msg):
+            if _is_hidden_human_control_message(msg, thread_id=thread_id):
                 continue
             break
 
@@ -1146,7 +1146,7 @@ def _accumulate_stream_text(
     return buffers[message_id], message_id
 
 
-def _extract_artifacts(result: dict | list) -> list[str]:
+def _extract_artifacts(result: dict | list, *, thread_id: str | None = None) -> list[str]:
     """Extract artifact paths from the last AI response cycle only.
 
     Instead of reading the full accumulated ``artifacts`` state (which contains
@@ -1167,7 +1167,7 @@ def _extract_artifacts(result: dict | list) -> list[str]:
             continue
         # Stop at the last human message — anything before it is a previous turn
         if msg.get("type") == "human":
-            if _is_hidden_human_control_message(msg):
+            if _is_hidden_human_control_message(msg, thread_id=thread_id):
                 continue
             break
         # Look for AI messages with present_files tool calls
@@ -1181,13 +1181,20 @@ def _extract_artifacts(result: dict | list) -> list[str]:
     return artifacts
 
 
-def _is_hidden_human_control_message(msg: Mapping[str, Any]) -> bool:
+def _is_hidden_human_control_message(msg: Mapping[str, Any], *, thread_id: str | None = None) -> bool:
     """Distinguish internal control messages from real hidden user answers."""
     if msg.get("type") != "human":
         return False
 
     additional_kwargs = msg.get("additional_kwargs")
     if not isinstance(additional_kwargs, Mapping):
+        if additional_kwargs is not None:
+            logger.warning(
+                "Ignoring malformed message additional_kwargs for thread %s at row seq %s: expected a mapping, got %s",
+                thread_id or msg.get("thread_id", "<unknown>"),
+                msg.get("seq", "<unknown>"),
+                type(additional_kwargs).__name__,
+            )
         return False
 
     return additional_kwargs.get("hide_from_ui") is True and read_human_input_response(additional_kwargs) is None
@@ -3462,9 +3469,9 @@ class ChannelManager:
                 return
             result = await self._admitted_turn(client, msg, thread_id) or result
 
-        response_text = _extract_response_text(result)
+        response_text = _extract_response_text(result, thread_id=thread_id)
         pending_clarification = _has_current_turn_clarification(result)
-        artifacts = _extract_artifacts(result)
+        artifacts = _extract_artifacts(result, thread_id=thread_id)
 
         logger.info(
             "[Manager] agent response received: thread_id=%s, response_len=%d, artifacts=%d",
@@ -3517,7 +3524,7 @@ class ChannelManager:
         except Exception:
             logger.warning("[Manager] failed to read thread state for an admitted run: thread_id=%s", thread_id, exc_info=True)
             return None
-        return _turn_of_message(state.get("values") if isinstance(state, dict) else None, message_id)
+        return _turn_of_message(state.get("values") if isinstance(state, dict) else None, message_id, thread_id=thread_id)
 
     async def _handle_streaming_chat(
         self,
@@ -3594,7 +3601,7 @@ class ChannelManager:
                     # Clarification text is only in the values snapshot;
                     # publish it so the user sees the question mid-stream.
                     if _has_current_turn_clarification(data):
-                        clarification_text = _extract_response_text(data)
+                        clarification_text = _extract_response_text(data, thread_id=thread_id)
                         if clarification_text and clarification_text != latest_text:
                             latest_text = clarification_text
 
@@ -3644,9 +3651,9 @@ class ChannelManager:
             if replay_gap and last_values is None and not latest_text and stream_error is None:
                 last_values = await self._admitted_turn(client, msg, thread_id)
             result = last_values if last_values is not None else {"messages": [{"type": "ai", "content": latest_text}]}
-            response_text = _extract_response_text(result)
+            response_text = _extract_response_text(result, thread_id=thread_id)
             pending_clarification = _has_current_turn_clarification(result)
-            artifacts = _extract_artifacts(result)
+            artifacts = _extract_artifacts(result, thread_id=thread_id)
             # Reuse the storage owner resolved by _handle_chat so artifact delivery
             # matches the upload bucket and we avoid re-running _safe_user_id_for_run
             # (and its possible filesystem touch) on the streaming-error path.

backend/app/gateway/conversation_access.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import json
 import logging
 import re
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Mapping
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlsplit
 
@@ -57,16 +57,39 @@ def _source_id(reference: str, request_url: str) -> str:
     return validate_thread_id(match.group(1))
 
 
-def _visible_text(row: dict) -> tuple[str, str] | None:
+def _visible_text(row: dict, *, thread_id: str | None = None) -> tuple[str, str] | None:
     message = row.get("content")
-    if not isinstance(message, dict):
+    if not isinstance(message, Mapping):
         return None
     role = message.get("type") or message.get("role")
+    if not isinstance(role, str):
+        return None
     role = {"human": "user", "ai": "assistant"}.get(role, role)
-    extra = message.get("additional_kwargs") or {}
+    extra = message.get("additional_kwargs")
+    if extra is None:
+        extra = {}
+    elif not isinstance(extra, Mapping):
+        logger.warning(
+            "Ignoring malformed message additional_kwargs for thread %s at row seq %s: expected a mapping, got %s",
+            thread_id or "<unknown>",
+            row.get("seq", "<unknown>"),
+            type(extra).__name__,
+        )
+        extra = {}
     if role not in {"user", "assistant"} or extra.get("hide_from_ui") or message.get("name") == "summary":
         return None
-    if str((row.get("metadata") or {}).get("caller", "")).startswith(("middleware:", "subagent:")):
+    metadata = row.get("metadata")
+    if metadata is None:
+        metadata = {}
+    elif not isinstance(metadata, Mapping):
+        logger.warning(
+            "Ignoring malformed conversation metadata for thread %s at row seq %s: expected a mapping, got %s",
+            thread_id or "<unknown>",
+            row.get("seq", "<unknown>"),
+            type(metadata).__name__,
+        )
+        metadata = {}
+    if str(metadata.get("caller", "")).startswith(("middleware:", "subagent:")):
         return None
     content = message.get("content")
     if role == "user" and isinstance(extra.get("original_user_content"), str):
@@ -228,7 +251,7 @@ def _prepare_owner_bound_conversation_reader(
         found: dict[int, tuple[str, str, int]] = {}
 
         def include_target(row: dict) -> bool:
-            parsed = _visible_text(row) if row["seq"] == message_seq else None
+            parsed = _visible_text(row, thread_id=thread_id) if row["seq"] == message_seq else None
             if parsed is None:
                 return False
             role, text = parsed
@@ -273,7 +296,7 @@ def _prepare_owner_bound_conversation_reader(
         parsed_text: dict[int, tuple[str, str]] = {}
 
         def include_message(row: dict) -> bool:
-            parsed = _visible_text(row)
+            parsed = _visible_text(row, thread_id=thread_id)
             if parsed is None:
                 return False
             role, text = parsed

backend/app/gateway/conversation_reader.py

@@ -8,7 +8,7 @@ access from a model argument or an ambient HTTP request.
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -70,8 +70,19 @@ def _message_type(message: Any) -> str | None:
     return str(value) if value else None
 
 
-def _is_thread_history_hidden_message_row(row: dict[str, Any]) -> bool:
-    caller = str((row.get("metadata") or {}).get("caller", ""))
+def _is_thread_history_hidden_message_row(row: dict[str, Any], *, thread_id: str | None = None) -> bool:
+    metadata = row.get("metadata")
+    if metadata is None:
+        metadata = {}
+    elif not isinstance(metadata, Mapping):
+        logger.warning(
+            "Ignoring malformed conversation metadata for thread %s at row seq %s: expected a mapping, got %s",
+            thread_id or "<unknown>",
+            row.get("seq", "<unknown>"),
+            type(metadata).__name__,
+        )
+        metadata = {}
+    caller = str(metadata.get("caller", ""))
     return caller.startswith("middleware:") or (caller.startswith("subagent:") and _message_type(row.get("content")) == "ai")
 
 
@@ -110,7 +121,7 @@ async def scan_visible_thread_messages(
                 if before_seq is not None and row["seq"] >= before_seq:
                     reached_before_bound = True
                     break
-                if (not include_middleware and _is_thread_history_hidden_message_row(row)) or row.get("run_id") in hidden_run_ids or (message_filter is not None and not message_filter(row)):
+                if (not include_middleware and _is_thread_history_hidden_message_row(row, thread_id=thread_id)) or row.get("run_id") in hidden_run_ids or (message_filter is not None and not message_filter(row)):
                     continue
                 visible.append(row)
                 if len(visible) == needed:
@@ -137,7 +148,7 @@ async def scan_visible_thread_messages(
             break
         _validate_message_scan_rows(raw, thread_id=thread_id, scan_before=scan_before, scan_after=None)
         for row in reversed(raw):
-            if (not include_middleware and _is_thread_history_hidden_message_row(row)) or row.get("run_id") in hidden_run_ids or (message_filter is not None and not message_filter(row)):
+            if (not include_middleware and _is_thread_history_hidden_message_row(row, thread_id=thread_id)) or row.get("run_id") in hidden_run_ids or (message_filter is not None and not message_filter(row)):
                 continue
             visible_desc.append(row)
             if len(visible_desc) == needed:

backend/app/gateway/routers/thread_runs.py

@@ -16,6 +16,7 @@ import hashlib
 import logging
 import re
 import uuid
+from collections.abc import Mapping
 from copy import deepcopy
 from datetime import UTC, datetime
 from typing import Annotated, Any, Literal
@@ -208,7 +209,7 @@ def compute_run_durations(runs) -> dict[str, int]:
     return durations
 
 
-def stamp_turn_duration_on_last_ai(messages, run_durations: dict[str, int]) -> None:
+def stamp_turn_duration_on_last_ai(messages, run_durations: dict[str, int], *, thread_id: str | None = None) -> None:
     """Attach each run's elapsed seconds to that run's final visible AI message only.
 
     ``turn_duration`` is the run's wall-clock lifetime (``compute_run_durations``),
@@ -236,10 +237,37 @@ def stamp_turn_duration_on_last_ai(messages, run_durations: dict[str, int]) -> N
             continue
         content = msg.get("content")
         payload = content if isinstance(content, dict) else msg
-        metadata = msg.get("metadata") or {}
+        metadata = msg.get("metadata")
+        if metadata is None:
+            metadata = {}
+        elif not isinstance(metadata, Mapping):
+            logger.warning(
+                "Ignoring malformed event metadata for thread %s at row seq %s: expected a mapping, got %s",
+                sanitize_log_param(thread_id or "<unknown>"),
+                msg.get("seq", "<unknown>"),
+                type(metadata).__name__,
+            )
+            metadata = {}
         is_middleware = str(metadata.get("caller", "")).startswith("middleware:")
         if payload.get("type") == "ai" and not is_middleware:
-            payload.setdefault("additional_kwargs", {})["turn_duration"] = run_durations[rid]
+            raw_additional_kwargs = payload.get("additional_kwargs")
+            if raw_additional_kwargs is None:
+                additional_kwargs: dict[str, Any] = {}
+            elif isinstance(raw_additional_kwargs, Mapping):
+                # A Mapping is not necessarily mutable (for example, a
+                # MappingProxyType from a serializer), so copy it before
+                # attaching the duration.
+                additional_kwargs = dict(raw_additional_kwargs)
+            else:
+                logger.warning(
+                    "Ignoring malformed message additional_kwargs for thread %s at row seq %s: expected a mapping, got %s",
+                    sanitize_log_param(thread_id or "<unknown>"),
+                    msg.get("seq", "<unknown>"),
+                    type(raw_additional_kwargs).__name__,
+                )
+                additional_kwargs = {}
+            payload["additional_kwargs"] = additional_kwargs
+            additional_kwargs["turn_duration"] = run_durations[rid]
             stamped.add(rid)
 
 
@@ -1582,7 +1610,7 @@ async def list_thread_messages(
     run_durations = compute_run_durations(runs)
 
     if run_durations:
-        stamp_turn_duration_on_last_ai(messages, run_durations)
+        stamp_turn_duration_on_last_ai(messages, run_durations, thread_id=thread_id)
 
     return messages
 
@@ -1651,7 +1679,7 @@ async def _enrich_thread_message_page(
     # endpoints already use ``stamp_turn_duration_on_last_ai``; the page
     # endpoint was inlining the equivalent loop but stamping every AI row,
     # which #4163 fixed for the other paths and missed here.
-    stamp_turn_duration_on_last_ai(data, run_durations)
+    stamp_turn_duration_on_last_ai(data, run_durations, thread_id=thread_id)
     return data
 
 
@@ -1716,7 +1744,7 @@ async def list_run_messages(
         if record:
             durations = compute_run_durations([record])
             if durations:
-                stamp_turn_duration_on_last_ai(data, durations)
+                stamp_turn_duration_on_last_ai(data, durations, thread_id=thread_id)
 
     return {"data": data, "has_more": has_more}
 

backend/app/gateway/routers/threads.py

@@ -2002,6 +2002,7 @@ async def get_thread_history(
                                 stamp_turn_duration_on_last_ai(
                                     serialized_msgs,
                                     checkpoint_run_durations,
+                                    thread_id=thread_id,
                                 )
                                 raise
 
@@ -2052,6 +2053,7 @@ async def get_thread_history(
                         stamp_turn_duration_on_last_ai(
                             serialized_msgs,
                             resolved_run_durations,
+                            thread_id=thread_id,
                         )
 
                     except Exception:

backend/tests/test_conversation_access.py

@@ -148,6 +148,31 @@ def test_reader_pages_visible_text_and_keeps_source_unchanged():
     asyncio.run(exercise())
 
 
+def test_reader_skips_malformed_optional_metadata_without_losing_the_page(caplog):
+    async def exercise():
+        prepare, events, threads, manager, _ = _setup()
+        await threads.create("source", user_id="alice")
+        await _put(events, "valid before")
+        malformed = await events.put(
+            thread_id="source",
+            run_id="run-2",
+            category="message",
+            event_type="llm.ai.response",
+            content={"type": "ai", "id": "malformed", "content": "valid after", "additional_kwargs": "legacy"},
+            metadata=["legacy metadata"],
+        )
+        reader, _ = prepare(["source"])
+        with caplog.at_level("WARNING", logger="app.gateway.conversation_access"):
+            page = json.loads(await reader(thread_id="source", cursor=None, limit=20))
+        assert page["status"] == "ok"
+        assert [message["text"] for message in page["messages"]] == ["valid before", "valid after"]
+        assert malformed is not None
+        assert any("thread source" in record.message and "row seq 2" in record.message for record in caplog.records)
+        assert any("additional_kwargs" in record.message and "thread source" in record.message and "row seq 2" in record.message for record in caplog.records)
+
+    asyncio.run(exercise())
+
+
 def test_wrong_owner_missing_and_unlisted_targets_are_denied_before_content_read():
     async def exercise():
         prepare, events, threads, _, _ = _setup()

backend/tests/test_conversation_access_edge_cases.py

@@ -39,9 +39,9 @@ def test_visible_text_is_parsed_once_per_scan_and_refreshed_on_the_next_read(mon
 
     calls = Counter()
 
-    def track_projection(row):
+    def track_projection(row, **kwargs):
         calls[row["seq"]] += 1
-        return _visible_text(row)
+        return _visible_text(row, **kwargs)
 
     monkeypatch.setattr(conversation_access, "_visible_text", track_projection)
 
@@ -97,6 +97,13 @@ def test_only_visible_text_blocks_cross_the_reader():
     assert _visible_text(row) == ("assistant", "visible answer")
 
 
+@pytest.mark.parametrize("malformed_role", [["ai"], {"type": "ai"}])
+def test_visible_text_skips_unhashable_message_roles(malformed_role):
+    row = {"seq": 1, "content": {"type": malformed_role, "content": "ignored"}}
+
+    assert _visible_text(row) is None
+
+
 @pytest.mark.parametrize("path", ["/workspace/chats/source", "/workspace/agents/researcher/chats/source"])
 def test_reference_accepts_both_frontend_conversation_routes(path):
     prepare, _, _, _, _ = _setup()

backend/tests/test_conversation_reader.py

@@ -6,7 +6,7 @@ import asyncio
 from copy import deepcopy
 from unittest.mock import AsyncMock
 
-from app.gateway.conversation_reader import read_visible_message_page
+from app.gateway.conversation_reader import _is_thread_history_hidden_message_row, read_visible_message_page
 from deerflow.runtime.events.store.memory import MemoryRunEventStore
 from deerflow.runtime.runs.manager import EditReplayVisibility
 
@@ -134,3 +134,16 @@ def test_reader_reports_exhaustion_when_only_filtered_messages_remain():
         )
 
     assert asyncio.run(exercise()) == ([], False)
+
+
+def test_reader_logs_and_keeps_rows_with_malformed_metadata(caplog):
+    row = {
+        "seq": 4,
+        "metadata": [],
+        "content": {"type": "ai", "content": "visible", "additional_kwargs": {}},
+    }
+
+    with caplog.at_level("WARNING", logger="app.gateway.conversation_reader"):
+        assert _is_thread_history_hidden_message_row(row, thread_id="source-thread") is False
+
+    assert any("thread source-thread" in record.message and "row seq 4" in record.message for record in caplog.records)

backend/tests/test_thread_run_messages_pagination.py

@@ -308,6 +308,41 @@ def test_list_run_messages_injects_turn_duration_only_on_last_ai():
     assert data[2]["content"]["additional_kwargs"]["turn_duration"] == 5
 
 
+def test_list_run_messages_tolerates_malformed_metadata_and_logs(caplog):
+    """Event-store metadata is untrusted and must not break the messages endpoint."""
+    from deerflow.runtime import RunRecord
+
+    mock_run = RunRecord(
+        run_id="run-1",
+        thread_id="thread-1",
+        assistant_id=None,
+        status="success",
+        on_disconnect="cancel",
+        created_at="2026-06-20T10:00:00Z",
+        updated_at="2026-06-20T10:00:05Z",
+    )
+    rows = [
+        {
+            "seq": 7,
+            "run_id": "run-1",
+            "content": {"type": "ai", "text": "Response"},
+            "metadata": ["legacy metadata"],
+        }
+    ]
+    event_store = _make_event_store(rows)
+    run_manager = AsyncMock()
+    run_manager.get.return_value = mock_run
+    app = _make_app(event_store=event_store, run_manager=run_manager)
+
+    with caplog.at_level("WARNING", logger="app.gateway.routers.thread_runs"):
+        with TestClient(app) as client:
+            response = client.get("/api/threads/thread-1/runs/run-1/messages")
+
+    assert response.status_code == 200
+    assert response.json()["data"][0]["content"]["additional_kwargs"]["turn_duration"] == 5
+    assert any("thread thread-1" in record.message and "row seq 7" in record.message and "list" in record.message for record in caplog.records)
+
+
 def test_list_run_messages_turn_duration_skips_middleware_tail():
     """The duration badge lands on the assistant's final answer, not on a
     trailing middleware-caller message (e.g. title generation)."""

backend/tests/test_threads_router.py

@@ -2703,6 +2703,17 @@ def test_get_thread_history_injects_turn_duration_once_per_run() -> None:
     assert replayed[4]["additional_kwargs"]["turn_duration"] == 9
 
 
+def test_stamp_turn_duration_recovers_from_non_mapping_additional_kwargs(caplog) -> None:
+    from app.gateway.routers.thread_runs import stamp_turn_duration_on_last_ai
+
+    messages = [{"seq": 7, "run_id": "run-1", "type": "ai", "additional_kwargs": "legacy"}]
+    with caplog.at_level("WARNING", logger="app.gateway.routers.thread_runs"):
+        stamp_turn_duration_on_last_ai(messages, {"run-1": 4}, thread_id="thread-1")
+
+    assert messages[0]["additional_kwargs"] == {"turn_duration": 4}
+    assert any("thread thread-1" in record.message and "row seq 7" in record.message for record in caplog.records)
+
+
 # ── branch threads from completed assistant turns ─────────────────────────────