Skip to content

fix: ensure message stream order #4487

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Soulter merged 1 commit into from
Jan 15, 2026
Merged

fix: ensure message stream order #4487

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions astrbot/core/platform/sources/webchat/webchat_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ async def send_by_session(
session: MessageSesion,
message_chain: MessageChain,
):
await WebChatMessageEvent._send(message_chain, session.session_id)
message_id = f"active_{str(uuid.uuid4())}"
await WebChatMessageEvent._send(message_id, message_chain, session.session_id)
await super().send_by_session(session, message_chain)

async def _get_message_history(
Expand Down Expand Up @@ -196,7 +197,7 @@ async def convert_message(self, data: tuple) -> AstrBotMessage:

abm.session_id = f"webchat!{username}!{cid}"

abm.message_id = str(uuid.uuid4())
abm.message_id = payload.get("message_id")
Copy link
Contributor

@sourcery-ai sourcery-ai bot Jan 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): 直接依赖客户端提供的 message_id，可能会引入 None / 无效 ID 以及信任问题。

之前这里始终是由服务端生成的 UUID，因此下游代码可以依赖 message_id 一定存在、非空且唯一。现在直接使用客户端提供的值，它可能缺失、为 None，或者是任意值，这会破坏原有假设并削弱对关联 ID 的信任度。请增加基础校验（存在性 / 类型 / 格式），并在客户端值缺失或无效时回退到服务端生成的 UUID。

Original comment in English

issue (bug_risk): Relying directly on client-provided message_id may introduce None/invalid IDs and trust issues.

Previously this was always a server-generated UUID, so downstream code could rely on message_id being present, non-empty, and unique. With the client value used directly, it can be missing, None, or arbitrary, which may break that assumption and weakens correlation ID trust. Please add basic validation (presence/type/format) and fall back to a server-generated UUID when the client value is absent or invalid.


# 处理消息段列表
message_parts = payload.get("message", [])
Expand Down
19 changes: 16 additions & 3 deletions astrbot/core/platform/sources/webchat/webchat_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@ def __init__(self, message_str, message_obj, platform_meta, session_id):

@staticmethod
async def _send(
message: MessageChain | None, session_id: str, streaming: bool = False
message_id: str,
message: MessageChain | None,
session_id: str,
streaming: bool = False,
) -> str | None:
cid = session_id.split("!")[-1]
web_chat_back_queue = webchat_queue_mgr.get_or_create_back_queue(cid)
Expand All @@ -31,6 +34,7 @@ async def _send(
"type": "end",
"data": "",
"streaming": False,
"message_id": message_id,
}, # end means this request is finished
)
return
Expand All @@ -45,6 +49,7 @@ async def _send(
"data": data,
"streaming": streaming,
"chain_type": message.type,
"message_id": message_id,
},
)
elif isinstance(comp, Json):
Expand All @@ -54,6 +59,7 @@ async def _send(
"data": json.dumps(comp.data, ensure_ascii=False),
"streaming": streaming,
"chain_type": message.type,
"message_id": message_id,
},
)
elif isinstance(comp, Image):
Expand All @@ -69,6 +75,7 @@ async def _send(
"type": "image",
"data": data,
"streaming": streaming,
"message_id": message_id,
},
)
elif isinstance(comp, Record):
Expand All @@ -84,6 +91,7 @@ async def _send(
"type": "record",
"data": data,
"streaming": streaming,
"message_id": message_id,
},
)
elif isinstance(comp, File):
Expand All @@ -100,6 +108,7 @@ async def _send(
"type": "file",
"data": data,
"streaming": streaming,
"message_id": message_id,
},
)
else:
Expand All @@ -108,14 +117,16 @@ async def _send(
return data

async def send(self, message: MessageChain | None):
await WebChatMessageEvent._send(message, session_id=self.session_id)
message_id = self.message_obj.message_id
await WebChatMessageEvent._send(message_id, message, session_id=self.session_id)
await super().send(MessageChain([]))

async def send_streaming(self, generator, use_fallback: bool = False):
final_data = ""
reasoning_content = ""
cid = self.session_id.split("!")[-1]
web_chat_back_queue = webchat_queue_mgr.get_or_create_back_queue(cid)
message_id = self.message_obj.message_id
async for chain in generator:
# if chain.type == "break" and final_data:
# # 分割符
Expand All @@ -130,7 +141,8 @@ async def send_streaming(self, generator, use_fallback: bool = False):
# continue

r = await WebChatMessageEvent._send(
chain,
message_id=message_id,
message=chain,
session_id=self.session_id,
streaming=True,
)
Expand All @@ -147,6 +159,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
"data": final_data,
"reasoning": reasoning_content,
"streaming": True,
"message_id": message_id,
},
)
await super().send_streaming(generator, use_fallback)
10 changes: 10 additions & 0 deletions astrbot/dashboard/routes/chat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,6 +296,8 @@ async def chat(self):
# 构建用户消息段（包含 path 用于传递给 adapter）
message_parts = await self._build_user_message_parts(message)

message_id = str(uuid.uuid4())
Copy link
Contributor

@sourcery-ai sourcery-ai bot Jan 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): 直接丢弃 message_id 不匹配的响应，可能导致流式请求出现隐蔽的卡住或只输出部分内容的问题。

如果 result["message_id"] != message_id，我们目前只是记录日志并 continue。如果后端持续返回错误的 ID（例如 bug、会话路由错误、代理混合了不同的流），客户端会看到连接一直打开但没有任何 token 返回，本质上是无明确错误提示的“卡死”。请考虑把这种情况视为硬错误（中断/终止流）或将错误显式反馈给调用方，而不是悄悄丢弃所有数据。

Original comment in English

issue (bug_risk): Dropping mismatched message_id responses might cause subtle stream hangs or partial outputs.

If result["message_id"] != message_id, we currently just log and continue. If the backend ever consistently returns a wrong ID (e.g., bug, misrouted session, proxy mixing streams), the client will see an open connection but no tokens, effectively a hang with no explicit error. Consider treating this as a hard failure (breaking/aborting the stream) or surfacing an error to the caller instead of silently dropping all data.


async def stream():
client_disconnected = False
accumulated_parts = []
Expand All @@ -319,6 +321,13 @@ async def stream():
if not result:
continue

if (
"message_id" in result
and result["message_id"] != message_id
):
logger.warning("webchat stream message_id mismatch")
continue

result_text = result["data"]
msg_type = result.get("type")
streaming = result.get("streaming", False)
Expand Down Expand Up @@ -456,6 +465,7 @@ async def stream():
"selected_provider": selected_provider,
"selected_model": selected_model,
"enable_streaming": enable_streaming,
"message_id": message_id,
},
),
)
Expand Down
Loading