修复: 上下文流式与工具签名域清债
- tool_signature 归一化补参,合法换参不再误触探索熔断 - SSE 超限截头改对齐事件边界保尾,流不再永久破损 - content_filter 纳入终态判定,无 DONE 流也收敛 - sanitize 占位补头前缀短路只吞 orphan 的盲区消除 - intent 关键词加词边界,history 不再误判 - 恢复重放后重置话题锚点,URL 图 token 地板不被稀释
This commit is contained in:
@@ -17,7 +17,8 @@
|
||||
//! data: 前缀提取)
|
||||
//! - 多字节续接:事件体只在定位到完整分隔符后才解码,跨 chunk 的多字节字符在
|
||||
//! buffer 中天然拼接复原(分隔符为 ASCII,不落在多字节序列中间)
|
||||
//! - 容错:解析失败时跳过该事件继续,不中断流;buffer 设 `BUF_MAX` 上限防 OOM
|
||||
//! - 容错:解析失败时跳过该事件继续,不中断流;buffer 设 `BUF_MAX` 上限防 OOM,
|
||||
//! 超限时按事件边界对齐保尾裁剪(丢最旧的超限事件整体),不产出半截破损事件
|
||||
//! - 返回 Vec<String>(每个元素是一个事件 data 字段拼接内容)
|
||||
|
||||
use futures::Stream;
|
||||
@@ -35,6 +36,9 @@ const BUF_MAX: usize = 1 << 20; // 1 MiB
|
||||
pub struct SseStream<S> {
|
||||
inner: S,
|
||||
buffer: Vec<u8>,
|
||||
/// 超限裁剪置位:无边界的超限事件已整体丢弃,但其尾部字节仍在后续 chunk 中,
|
||||
/// 到达收尾 boundary 前的数据须静默吞掉,避免残缺事件体被当完整事件产出
|
||||
skip_partial: bool,
|
||||
}
|
||||
|
||||
impl<S> SseStream<S>
|
||||
@@ -45,6 +49,7 @@ where
|
||||
Self {
|
||||
inner,
|
||||
buffer: Vec::with_capacity(8192),
|
||||
skip_partial: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -62,6 +67,17 @@ where
|
||||
/// 字符在 buffer 中天然拼接复原(分隔符为 ASCII,不会落在多字节序列中间)。
|
||||
/// 解码用 `String::from_utf8`(失败再降级 lossy),避免把合法多字节误判为残缺。
|
||||
fn parse_events(&mut self) -> Vec<SseEvent> {
|
||||
// 超限裁剪收尾:被整体丢弃的残缺事件,其后续字节以首个 boundary 收尾,
|
||||
// 此处静默吞掉不产出,之后的正常事件恢复正常解析
|
||||
if self.skip_partial {
|
||||
match Self::find_event_boundary(&self.buffer) {
|
||||
Some((pos, len)) => {
|
||||
self.buffer.drain(..pos + len);
|
||||
self.skip_partial = false;
|
||||
}
|
||||
None => return Vec::new(),
|
||||
}
|
||||
}
|
||||
let mut events = Vec::new();
|
||||
loop {
|
||||
// 同时识别三种分隔符,取最早出现的位置与该分隔符的字节长度
|
||||
@@ -149,10 +165,23 @@ where
|
||||
Poll::Ready(Some(Ok(chunk))) => {
|
||||
self.buffer.extend_from_slice(&chunk);
|
||||
// 防御:缓冲区不应无限增长。正常情况下 parse_events 会持续消费,
|
||||
// 仅当上游持续不发分隔符(畸形流)时才触发,此处裁掉头部旧数据避免 OOM。
|
||||
// 超限只发生在单个事件体超过 BUF_MAX 的场景(畸形上游不发分隔符,
|
||||
// 或超大单事件)。按字节盲截头部会把半截事件体留给后续解析,产出
|
||||
// 破损 JSON 且无法复原;改为对齐事件边界的保尾裁剪:
|
||||
// - 已有完整分隔符 → 最旧的一个事件(超限源头)整体丢弃,保留的
|
||||
// 数据从干净的事件起点开始,后续事件解析不受损;
|
||||
// - 全程无分隔符 → 整体丢弃并置 skip_partial,该事件的收尾字节
|
||||
// 到达时静默吞掉。
|
||||
if self.buffer.len() > BUF_MAX {
|
||||
let drop_n = self.buffer.len() - BUF_MAX;
|
||||
self.buffer.drain(..drop_n);
|
||||
match Self::find_event_boundary(&self.buffer) {
|
||||
Some((pos, len)) => {
|
||||
self.buffer.drain(..pos + len);
|
||||
}
|
||||
None => {
|
||||
self.buffer.clear();
|
||||
self.skip_partial = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
@@ -160,8 +189,9 @@ where
|
||||
return Poll::Ready(Some(Err(format!("SSE 流读取错误: {}", e))));
|
||||
}
|
||||
Poll::Ready(None) => {
|
||||
// 流结束,处理 buffer 中的剩余数据(可能没有 \n\n 结束的最后一段)
|
||||
if !self.buffer.is_empty() {
|
||||
// 流结束,处理 buffer 中的剩余数据(可能没有 \n\n 结束的最后一段);
|
||||
// skip_partial 置位时剩余数据是被丢弃残缺事件的尾段,不得溢出为事件
|
||||
if !self.buffer.is_empty() && !self.skip_partial {
|
||||
// 解码:优先严格 UTF-8(EOF 无下一 chunk 可拼接,
|
||||
// 残缺多字节尾部只能尽力而为,降级 lossy 不崩)
|
||||
let mut remaining = match String::from_utf8(self.buffer.clone()) {
|
||||
@@ -185,3 +215,58 @@ where
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use futures::stream;
|
||||
|
||||
/// 构造只含 Ok chunk 的内存流,跑完整个 SseStream,返回拍平的事件序列
|
||||
fn run_chunks(chunks: Vec<String>) -> Vec<SseEvent> {
|
||||
use futures::StreamExt;
|
||||
let items: Vec<Result<bytes::Bytes, reqwest::Error>> = chunks
|
||||
.into_iter()
|
||||
.map(|c| Ok(bytes::Bytes::from(c)))
|
||||
.collect();
|
||||
let mut s = SseStream::new(stream::iter(items));
|
||||
futures::executor::block_on(async {
|
||||
let mut out = Vec::new();
|
||||
while let Some(item) = s.next().await {
|
||||
out.extend(item.expect("测试流不应产生错误"));
|
||||
}
|
||||
out
|
||||
})
|
||||
}
|
||||
|
||||
/// 回归:正常多事件流照常解析
|
||||
#[test]
|
||||
fn normal_events_parse() {
|
||||
let events = run_chunks(vec![
|
||||
"data: {\"a\":1}\n\n".into(),
|
||||
"data: {\"b\":2}\n\n".into(),
|
||||
]);
|
||||
assert_eq!(events, vec!["{\"a\":1}".to_string(), "{\"b\":2}".to_string()]);
|
||||
}
|
||||
|
||||
/// 超限单事件(末尾带完整分隔符)整体丢弃:不产出半截破损事件,后续正常事件照常解析。
|
||||
/// 旧按字节截头实现会把残留尾部 "data: TAIL" 行当伪事件产出
|
||||
#[test]
|
||||
fn oversize_event_with_boundary_dropped_whole() {
|
||||
let giant = format!("data: {}\ndata: TAIL\n\n", "x".repeat(BUF_MAX));
|
||||
let events = run_chunks(vec![giant, "data: {\"ok\":1}\n\n".into()]);
|
||||
assert_eq!(events, vec!["{\"ok\":1}".to_string()]);
|
||||
}
|
||||
|
||||
/// 验收:超限注入(无边界大块)丢弃后,残缺事件收尾静默吞掉,后续正常事件仍可解析。
|
||||
/// 旧按字节截头实现会把 STALE-TAIL 残段当伪事件产出
|
||||
#[test]
|
||||
fn oversize_boundaryless_blob_then_normal_events_still_parse() {
|
||||
let events = run_chunks(vec![
|
||||
"x".repeat(BUF_MAX + 100),
|
||||
"data: STALE-TAIL\n\n".into(),
|
||||
"data: {\"ok\":1}\n\n".into(),
|
||||
"data: [DONE]\n\n".into(),
|
||||
]);
|
||||
assert_eq!(events, vec!["{\"ok\":1}".to_string(), "[DONE]".to_string()]);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user