mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
6c4c74a504
* fix(channels): share chat-to-thread bindings through the database ChannelManager kept its IM chat-to-thread bindings (channel_name:chat_id[:topic_id] -> thread_id) in a per-process JSON file, channels/store.json, loaded once and rewritten whole on every change. With several Gateway processes a binding created on one replica was invisible to the others (the next message for the same chat opened a second thread) and concurrent writers clobbered each other's file. - New channel_thread_bindings table (migration 0036_channel_thread_bindings) and harness package deerflow.persistence.channel_thread_bindings: key PK (the legacy composite string), channel_name, chat_id, nullable topic_id, thread_id, user_id, epoch created_at/updated_at; index (channel_name, chat_id). Dialect upsert keeps created_at; prefix remove and batched ON CONFLICT DO NOTHING import. - app.channels.store.ChannelStore is an async protocol: SqlChannelStore for sqlite/postgres, JsonChannelStore (file I/O via asyncio.to_thread) for database.backend: memory. resolve_channel_store picks at ChannelService construction; ChannelService.start() imports an existing store.json once into an empty table and renames it store.json.migrated (a concurrently importing peer neither duplicates nor clobbers; a populated table leaves the file alone). - Callers updated: manager, connection_identity.lookup_thread_id, Feishu (send() awaits the mapping writer; the synchronous lark callback bridges its two store calls to the Gateway loop with a bounded wait), Discord (channels_dir). - Head pins in the 0027_0028/0029/0032/0034/0035 migration tests point at the new head; docs (channels and migrations AGENTS.md, CONFIGURATION.md, README) and CHANGELOG updated. * docs(changelog): link the channel thread bindings entry to #6558 * fix(channels): reject only the Gateway loop in the Feishu store bridge lark-oapi 1.5.5 invokes the message callback synchronously from its async frame handler, on the private event loop that FeishuChannel._run_ws starts for the SDK thread. The bridge's guard treated any running loop as unsafe, so every real inbound message skipped the stored-topic lookup and the pending-mapping restore: a reply whose parent card was mapped but whose root was not fell back to the unknown root, and could consume an unrelated pending clarification. Only a wait issued from the Gateway loop itself (the loop that executes the submitted coroutine) is refused now; the SDK's own loop may block for the bounded wait, as that thread blocked on the JSON rewrite before. Tests drive _on_message from a worker thread running its own loop, the production shape, for both the lookup and the write path, and pin the Gateway-loop refusal. * fix(channels): defer unresolved Feishu lookups and harden the store.json import Review follow-ups on the chat-to-thread binding store (#6558): - Import: a legacy entry whose key, channel_name, chat_id, topic_id, thread_id or user_id exceeds its column width is skipped and counted with the malformed ones instead of failing the whole batch on PostgreSQL (SQLite silently accepted the same row, so the backends diverged on one file). The corrupt-file warning now says what each caller does: the JSON store starts fresh, the import is skipped and the file kept. - Feishu bridge: store calls run on the Gateway loop under a 3s asyncio.wait_for (the SDK loop also services ws keepalives; the old 10s bound could stall it per message), and the lark-thread wait is that budget plus 1s. A deadline or store error is StoreLookupUnresolved, never a miss: _on_message no longer routes such a reply onto its unknown root (which created a second thread and later overwrote the parent alias). The message joins a per-chat FIFO drained on the Gateway loop with backoff for up to 30s, then is dropped with a warning. _on_message is split into parse / resolve / dispatch so the deferred path dispatches the same way; later messages for the chat queue behind the deferred one. * fix(tests): expect follow_redirects in the Jina client kwargs after #5808 #5808 made JinaClient pass follow_redirects=True to httpx.AsyncClient and updated test_jina_client.py, but test_jina_response_limit.py and test_jina_retry_after.py still pin the exact client kwargs to proxy and trust_env, so every backend-unit-tests shard that lands on them fails on main (the backend unit-test workflow only runs on pull requests, which is why the push to main did not surface it). Update the two expectations to include the new keyword. (cherry picked from commit d5aa23156ce19ce36d95df03953442f83ae6d050) * Revert "fix(tests): expect follow_redirects in the Jina client kwargs after #5808" This reverts commit 5bee00bfb09c65a30f77fb4df47a1f82c47ecef7. * fix(channels): bound deferred Feishu routing by the inbound queue capacity Review follow-up on #6558: a store outage let every subsequent message for a chat join the deferred queue with no capacity check (the bus reservation was only taken after routing succeeded), and each queued message started a fresh 30s retry budget when it reached the head, so a burst could be retained for N x 30s and grow without bound. - Intake is reserved BEFORE routing for every store-backed message (bus.reserve_inbound is thread-safe and already used from SDK threads), so deferred and live messages share the one configured bound, channels.inbound_queue_maxsize; a message that cannot be admitted is dropped at once with a warning, never queued or routed as a miss. - The reservation travels with the message into dispatch (committed by _prepare_inbound), is released on the batch path (the flush reserves at publish time), on drop after the budget, on a failed deferral, on dispatch failure and for every leftover when the drain task is cancelled by stop(). - The retry budget runs from the message's arrival: a queued message whose budget is spent when it reaches the head is dropped without a lookup. - The bounded resolver returns the unresolved outcome instead of raising it through the tracked task, which logged an ERROR for an expected, handled store timeout. - Upstream's new blocking-io test constructs the store by its old class name; it now uses JsonChannelStore. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
108 lines
4.5 KiB
Python
108 lines
4.5 KiB
Python
"""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 JsonChannelStore
|
|
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=JsonChannelStore(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
|