fix(channels): handle streamed run errors before dedupe cleanup (#6272)

* fix(channels): handle streamed run error frames

* fix(channels): log streamed run error details

README.md

@@ -874,6 +874,7 @@ Notes:
 - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation.
 - Inbound work is bounded to `inbound_queue_maxsize` pending messages plus `max_concurrency` active workers. When capacity is exhausted, socket/polling providers drop new messages before sending DeerFlow's working acknowledgment and emit a rate-limited warning. Buzz leaves its replay cursor unchanged and reconnects for relay replay; GitHub webhooks return `503`, marking the delivery failed for manual/API redelivery. Shutdown closes admission immediately, keeps channel transports available while accepted messages drain for up to `shutdown_grace_period_seconds`, then cancels and awaits active handlers before closing provider resources; the Gateway's outer timeout can cancel an incomplete shutdown without detaching those resources.
 - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches.
+- Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis.
 
 Set the corresponding API keys in your `.env` file:
 

backend/app/channels/AGENTS.md

@@ -8,7 +8,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 - `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels)
 - `store.py` - JSON persistence for `channel_name:chat_id[:topic_id]` → `thread_id`. Guard every `_data` access with `_lock`; `list_entries()` snapshots under the lock and formats after release so concurrent writers cannot resize the dict during iteration.
 - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
-  A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply.
+  Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection.
   **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `<durable_context_data>` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `<memory>` fact block and, in another run, a verbatim echo of the user's own message as the assistant's reply. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`).
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): used by the scheduled-task notification outbox to push markdown to a bound external identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running) so unsupported providers fail visibly in the outbox. Connectivity failures must raise `ChannelUnavailable` so the delivery worker parks without consuming retries; platform rejections of a well-formed send remain ordinary exceptions. **Only WeCom implements proactive push today** (`wecom.py` via `WSClient.send_message`); the SDK raises `RuntimeError` from that call for a dead socket and for a platform rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding via `resolve_connections` (`ChannelConnectionRepository.list_connections`, owner-scoped, ahead of the channel-liveness check): a target no longer `connected` for the row's owner is finalized with `mark_failed(terminal=True)`; a lookup failure parks. Terminal outbox rows (`sent`/`failed`) ignore later status writes. Scheduler side (`app/scheduler/`): `handle_run_completion` enqueues only after `complete_run` recorded a success or failed outcome; manual triggers, interrupts and recovery-finalized occurrences stay silent. Gateway lifespan wires the enqueue side with the scheduler, then starts `NotificationDeliveryWorker` after the channel service (the scheduler starts first), detaching the enqueue side if delivery cannot start; the worker stops before the channel service.
 - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` without consuming retry budget.

backend/app/channels/manager.py

@@ -2631,6 +2631,17 @@ class ChannelManager:
                     accumulated_text, current_message_id = _accumulate_stream_text(streamed_buffers, current_message_id, data)
                     if accumulated_text:
                         latest_text = accumulated_text
+                elif event == "error":
+                    error_data = data if isinstance(data, dict) else {}
+                    logger.warning(
+                        "[Manager] stream error frame: thread_id=%s, error=%s: %s",
+                        thread_id,
+                        error_data.get("name", "Error"),
+                        error_data.get("message", "unknown"),
+                    )
+                    # The v1 SDK yields backend errors as frames, not exceptions.
+                    # Keep draining the stream and use the existing failure path.
+                    stream_error = RuntimeError("Agent run failed.")
                 elif event == "values" and isinstance(data, (dict, list)):
                     last_values = data
                     # Clarification text is only in the values snapshot;

backend/tests/test_channel_stream_errors.py

@@ -0,0 +1,107 @@
+"""Error SSE frames must enter the IM channel's existing failure cleanup."""
+
+import logging
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from _router_auth_helpers import make_authed_test_app
+from langgraph_sdk.client import LangGraphClient
+
+from app.channels.manager import ChannelManager
+from app.channels.message_bus import InboundMessage, MessageBus
+from app.channels.store import ChannelStore
+from app.gateway.routers import thread_runs
+from deerflow.runtime import DisconnectMode, RunRecord, RunStatus
+from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "has_error, error_payload",
+    [
+        (False, None),
+        (True, {"name": "RuntimeError", "message": "Model failed."}),
+        (True, None),
+        (True, {"name": "ConflictError", "message": "Thread already running a task."}),
+    ],
+    ids=["success", "error-event", "empty-error-event", "busy-error-event"],
+)
+@pytest.mark.parametrize("has_partial", [False, True], ids=["empty", "partial-text"])
+@pytest.mark.parametrize("has_values", [False, True], ids=["message-stream", "with-values"])
+async def test_channel_consumes_error_sse_and_releases_dedupe_after_final_publish(monkeypatch, tmp_path, caplog, has_error, error_payload, has_partial, has_values):
+    caplog.set_level(logging.WARNING, logger="app.channels.manager")
+    bridge = MemoryStreamBridge()
+    record = RunRecord(
+        run_id="stream-run",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.error if has_error else RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        error="Model failed." if has_error else None,
+    )
+
+    async def start_run(*args, **kwargs):
+        if has_partial:
+            await bridge.publish(record.run_id, "messages-tuple", [{"type": "AIMessageChunk", "id": "answer", "content": "Partial"}, {"langgraph_node": "agent"}])
+        if has_values:
+            await bridge.publish(record.run_id, "values", {"messages": [{"type": "ai", "content": "Partial"}] if has_partial else []})
+        if has_error:
+            await bridge.publish(record.run_id, "error", error_payload)
+        await bridge.publish_end(record.run_id)
+        return record
+
+    monkeypatch.setattr(thread_runs, "start_run", start_run)
+    app = make_authed_test_app()
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = bridge
+    app.state.run_manager = SimpleNamespace(get=AsyncMock(return_value=record))
+    bus = MessageBus()
+    manager = ChannelManager(bus=bus, store=ChannelStore(path=tmp_path / "channels.json"))
+    inbound = InboundMessage(channel_name="feishu", chat_id="chat-1", user_id="user-1", text="Hello", metadata={"message_id": "message-1"})
+    assert await manager._is_duplicate_inbound(inbound) is False
+    outbound = []
+    deduped_during_final = []
+
+    async def capture(message):
+        outbound.append(message)
+        if message.is_final:
+            deduped_during_final.append(await manager._is_duplicate_inbound(inbound))
+
+    bus.subscribe_outbound(capture)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test/api") as http:
+        await manager._handle_streaming_chat(
+            LangGraphClient(http),
+            inbound,
+            "thread-1",
+            "lead_agent",
+            {},
+            {},
+            {"role": "user", "content": "Hello"},
+        )
+
+    final = outbound[-1]
+    assert final.is_final
+    if has_partial:
+        # Preserve the same partial-text behavior as a transport exception.
+        assert final.text == "Partial"
+    elif has_error:
+        assert final.text == "An error occurred while processing your request. Please try again."
+    else:
+        assert final.text == "(No response from agent)"
+    assert deduped_during_final == [True]
+    assert await manager._is_duplicate_inbound(inbound) is (not has_error)
+
+    error_logs = [record for record in caplog.records if record.name == "app.channels.manager" and "stream error frame:" in record.getMessage()]
+    assert len(error_logs) == int(has_error)
+    if has_error:
+        assert error_logs[0].levelno == logging.WARNING
+        logged = error_logs[0].getMessage()
+        assert "thread_id=thread-1" in logged
+        if isinstance(error_payload, dict):
+            assert error_payload["name"] in logged
+            assert error_payload["message"] in logged
+        else:
+            assert "Error" in logged
+            assert "unknown" in logged