From b455d79a318d397c7678cd802a1ab629cef5d6a3 Mon Sep 17 00:00:00 2001 From: codex-layer-stabilizer Date: Fri, 16 Oct 2026 10:24:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(proxy):=20=E5=8E=9F=E7=94=9F=E6=96=B9?= =?UTF-8?q?=E8=A8=80=E7=A9=BA=E7=BB=88=E5=B8=A7=E6=8A=A4=E6=A0=8F=E2=80=94?= =?UTF-8?q?=E2=80=94=E6=8A=8A=E5=AE=88=E9=97=A8=E6=8C=AA=E5=88=B0=E7=9C=9F?= =?UTF-8?q?=E6=AD=A3=E6=9C=89=E6=B5=81=E9=87=8F=E7=9A=84=E9=80=8F=E4=BC=A0?= =?UTF-8?q?=E6=94=AF=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本机 dogfood 全程走 Responses 原生方言，护栏此前只住在 Chat 转换层， 对这条支路的静默终帧零覆盖。现在 NativeResponsesGuard 在事件边界上 判静默、扣终帧、同上游重发，重发轮条目按位移续排、终帧合并、身份回填， 客户端始终只看到单条 created 与单个 completed。 判据与红线： - 静默判定与 Chat 方言共用 native_response_is_silent 本体；refusal 与 非空字符串 content 算有效输出，合规拒绝不再被误判重发； - 扣帧后后到字节全部并入暂存区，任何收口路径原样放行，绝不静默断流； - 上游自报 failed/incomplete/error 不拦不改；观察者永吃原始字节， 重开一轮时收口证据同步重开，防止「重发轮半途断流」被误判成已完成； - 巨帧或等不到边界的字节整段直通并永久失能，护栏永不无界缓冲。 可观测与文案： - 被丢轮的 reasoning/total token 在扣帧时入账，retry/exhausted 事件带 droppedUsage；stream_completed 带 attempts，监看能看见真实上游轮次； - 决策日志先于重发落盘；重发轮补记 stream_open 的口径改为 attempts 内嵌在配对事件里，不制造无终态的孤儿 open； - 推动文案跟随会话语言（CJK 判中/英），去掉「会话被自动结束」的元叙述， 尾部带识别标记并在注入前剥除历史同类项，禁止复述本条提示； - 轮次标记豁免推动项：重发轮仍钉在真实用户指令上（与非重试轮逐字节 同形），剔除推动项后非 user 收尾则维持不钉。 兼容口径：护栏开启后原生透传由「chunk 原样」变为「事件边界对齐下发」。 SSE 事件本以空行为完整边界，客户端按事件消费，不依赖 TCP chunk 切分。 夹具说明：两处原生 mock 的终帧原本没有 output 条目，正是护栏的触发形态， 已补可见正文并在注释里点明理由（这些测试测稳定层与 tools 回填，不应 顺带点燃护栏）。 测试：cargo test --all-targets 全绿（lib 212 + 集成 220，0 失败）； 新增原生护栏单元对拍 10 条与端到端 2 条（单创建帧/合并终帧/id 回填/ 位移/额度用尽仍出门/推动项只进重发体）；pinning 与 streaming 冒烟 ALL PASS。 --- src/main.rs | 157 +++- src/protocol_proxy.rs | 991 +++++++++++++++++++++++++- tests/native_empty_frame_guard.rs | 392 ++++++++++ tests/responses_dialect_stabilizer.rs | 4 +- tests/tools_cache_persistence.rs | 4 +- 5 files changed, 1501 insertions(+), 47 deletions(-) create mode 100644 tests/native_empty_frame_guard.rs diff --git a/src/main.rs b/src/main.rs index aa7c3d6..0697da4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -50,7 +50,7 @@ use codex_layer::protocol_proxy::{ open_models_proxy_request, open_responses_proxy_request_for_path_with_session_headers, responses_json_to_sse_events, responses_error_from_upstream, wrap_non_stream_response_as_compaction, upstream_conversion_error_response, - EMPTY_FRAME_MAX_RETRIES, with_empty_frame_nudge, + EMPTY_FRAME_MAX_RETRIES, NativeResponsesGuard, with_empty_frame_nudge, }; use codex_layer::stability::{self, StabilityAggregateOptions}; use codex_layer::stability_ui; @@ -837,15 +837,12 @@ async fn responses_proxy( events::layer::STREAM_OPEN, json!({ "path": path, "wireApi": format!("{:?}", upstream.wire_api), "compaction": compaction_round || upstream.compaction, "requestId": request_id.clone() }), ); - // 空终帧护栏的重发通道：四个条件齐了才开——Chat 方言（终帧判定与重发 - // 在那里才有意义）、非压缩轮（含降级轮，压缩轮有自己的收口账）、客户端 - // 请求解析得动（重发要靠它重建请求体）。 + // 空终帧护栏的重发通道（issue #83：双方言同权）：三个条件齐了才开—— + // 非压缩轮（含降级轮，压缩轮有自己的收口账）、客户端请求解析得动 + // （重发要靠它重建请求体）。静默判定与扣帧改写分别住在 Chat 转换层与 + // 原生透传护栏里，判据共用同一本体。 let retry_opener: Option = - if upstream.wire_api == UpstreamWireApi::ChatCompletions - && !compaction_round - && !upstream.compaction - && request_json.is_some() - { + if !compaction_round && !upstream.compaction && request_json.is_some() { let user_agent = user_agent.map(str::to_string); let beta_features = beta_features.map(str::to_string); let path = path.to_string(); @@ -1174,7 +1171,9 @@ fn stream_from_upstream( return; } if wire_api == UpstreamWireApi::Responses { - // 原生 Responses 透传支路：字节原样转发，观察者只旁路读，不产出任何字节。 + // 原生 Responses 透传支路。护栏关闭时与旧实现逐字节一致：字节原样 + // 转发，观察者只旁路读，不产出任何字节。护栏开启时按事件边界对齐 + // 下发（兼容口径见 `NativeResponsesGuard` 文档），静默终帧扣帧重发。 // 守卫要解决的是「上游在完成事件前断流」：原样转发时客户端只看到连接关闭， // 分不清正常结束还是半途而废，所以收口时由本层补一帧显式 response.failed。 let mut observer = NativeResponsesSseObserver::default(); @@ -1182,6 +1181,17 @@ fn stream_from_upstream( // 超限后停止转发：继续下发只会让客户端拿到一段注定要按失败收口的残缺流。 let mut over_limit = false; let mut drained = false; + // 空终帧护栏（issue #83）：压缩轮不开——装配层的重发通道本就不为 + // 压缩轮装配，这里显式再钉一道，两条防线不许只靠一条。 + let guard_enabled = retry_opener.is_some() && !native_compaction; + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(guard_enabled); + let mut retries_used = 0usize; + // 排空截止用装箱 future：重试轮重开上游流后要换新的截止， + // 已命中的截止不可能被重试轮再踩一次。 + let mut drain_wait: std::pin::Pin< + std::boxed::Box + Send + '_>, + > = Box::pin(drain.wait_deadline()); loop { let next = tokio::select! { next = chunks.next() => next, @@ -1193,6 +1203,7 @@ fn stream_from_upstream( let Some(chunk) = next else { break }; match chunk { Ok(bytes) => { + // 红线 4：观察者永远吃上游原始字节，护栏改写不影响补帧判定。 observer.push_bytes(&bytes); if native_compaction { tally.push_bytes(&bytes); @@ -1204,14 +1215,107 @@ fn stream_from_upstream( if over_limit { continue; } - if tx.send(Ok(bytes)).is_err() { + // 护栏关闭时保持旧行为：每个 chunk 无条件原样下发 + // （含空 chunk，逐字节一致的测试红线依赖这一点）。 + let forwarded = if guard_enabled { + guard.push(&bytes) + } else { + if tx.send(Ok(bytes)).is_err() { + return; + } + Vec::new() + }; + if !forwarded.is_empty() && tx.send(Ok(Bytes::from(forwarded))).is_err() { // 客户端已经断开：没有接收方，补帧无意义，直接收工。 return; } + if !guard.deferred() { + continue; + } + // 静默终帧被扣下：走重试决策。语义与 Chat 分支同款—— + // 重开成功就换新的上游流继续泵（第二轮条目按位移续排）； + // 额度用尽或重开失败则把扣下的帧原样放行后断泵， + // 走既有收口诊断。绝不造成「无 completed 也无 failed」 + // 的静默断流。 + if retries_used >= EMPTY_FRAME_MAX_RETRIES { + log_event( + events::layer::EMPTY_FRAME_EXHAUSTED, + json!({ "requestId": request_id, "attempts": retries_used, "droppedUsage": guard.dropped_usage() }), + ); + let released = guard.release_deferred(); + if !released.is_empty() && tx.send(Ok(Bytes::from(released))).is_err() { + return; + } + break; + } + let opener = retry_opener + .as_ref() + .expect("护栏开关打开时重发通道必在"); + // 第一次重试同体重采样（前缀全命中，纯赌温度）；第二次 + // 带推动文案。推动项拼不进 input（形态不认识）就原样 + // 重发，绝不自造请求结构。 + let same_body = request_json + .clone() + .expect("护栏开关打开时客户端请求必解析得动"); + let nudged = retries_used >= 1; + let retry_body = if nudged { + with_empty_frame_nudge(&same_body).unwrap_or(same_body) + } else { + same_body + }; + retries_used += 1; + // 决策日志先落：此刻重发还没成功，但「扣帧 + 已选请求体」 + // 这个事实现场是排障最需要的——重发失败时这里是唯一现场。 + // droppedUsage 带的是截至此刻被丢轮的累计账（扣帧时入账）。 + log_event( + events::layer::EMPTY_FRAME_RETRY, + json!({ "requestId": request_id, "attempt": retries_used, "nudged": nudged, "droppedUsage": guard.dropped_usage() }), + ); + match opener(retry_body).await { + Ok(opened) if opened.is_success() => { + // 扣帧暂存区的条目垫入下一轮基底，换上游流继续泵： + // 第二轮的重复身份事件由护栏吞掉，条目按位移续排 + // 在同一条客户端事件流里（Chat 分支同款机器）。 + guard.begin_retry(); + // 观察者的收口证据同步重开：上一轮 completed 已被 + // 扣帧丢弃（永不门），留着它会把「重发轮半途断流」 + // 误判成已完成——客户端将收不到任何终帧，红线所在。 + observer.begin_retry(); + chunks = opened.response.bytes_stream(); + drain_wait = Box::pin(drain.wait_deadline()); + continue; + } + other => { + // 重发通道自己失败：扣下的帧放行、照常收口； + // 日志要能分清「重发没开门」还是「又静默」。 + let reason = match &other { + Ok(opened) => { + format!("重发上游返回状态码 {}", opened.status_code) + } + Err(error) => error.to_string(), + }; + log_event( + events::layer::EMPTY_FRAME_EXHAUSTED, + json!({ "requestId": request_id, "attempts": retries_used, "reason": reason, "droppedUsage": guard.dropped_usage() }), + ); + let released = guard.release_deferred(); + if !released.is_empty() { + let _ = tx.send(Ok(Bytes::from(released))); + } + break; + } + } } Err(error) => { - // 读流错误：先补帧，再按既有行为把错误交给 HTTP 层中止 body； - // 这个顺序保证客户端至少还能拿到一个显式终止信号。 + // 读流错误：扣着的帧先放行（客户端至少先拿到「上游 + // 说完了」），再补帧，再按既有行为把错误交给 HTTP 层 + // 中止 body；这个顺序保证客户端至少还能拿到显式终止信号。 + if guard.deferred() { + let released = guard.release_deferred(); + if !released.is_empty() { + let _ = tx.send(Ok(Bytes::from(released))); + } + } let error_text = error.to_string(); let reason = match close_native_responses_stream(observer, "stream_error", &tx) { Some(reason) => format!("{reason}（传输错误：{error_text}）"), @@ -1232,6 +1336,14 @@ fn stream_from_upstream( } } } + if drained && guard.deferred() { + // 关停排空截止命中且还扣着终帧：先放行「上游说完了」， + // 再走显式失败收口。语义与 Chat 分支同款。 + let released = guard.release_deferred(); + if !released.is_empty() { + let _ = tx.send(Ok(Bytes::from(released))); + } + } if native_compaction && !drained { // 压缩轮流式收口：三条判据决定「已透传的字节算不算成功」。 // 成功与「上游自报失败」都不补帧（字节已经逐字节透传完了）； @@ -1264,6 +1376,12 @@ fn stream_from_upstream( } } } + // EOF 兜底：未成帧残留照发；万一还扣着帧（正常路径走不到这里）， + // 一并放行——红线高于一切，客户端永远拿得到终帧。 + let tail = guard.release_deferred(); + if !tail.is_empty() { + let _ = tx.send(Ok(Bytes::from(tail))); + } // HTTP EOF 不是 Responses 协议的完成事件，`[DONE]` 也不算： // 没观察到完成事件就一律补失败帧（与 Chat 泵的收口口径一致）。 // 排空截止点命中时同样走这条收口，只是 error_type 与措辞区分开。 @@ -1283,7 +1401,7 @@ fn stream_from_upstream( None => { log_event( events::layer::STREAM_COMPLETED, - json!({ "model": model, "requestId": request_id }), + json!({ "model": model, "requestId": request_id, "attempts": retries_used + 1 }), ); } } @@ -1336,7 +1454,7 @@ fn stream_from_upstream( if retries_used >= EMPTY_FRAME_MAX_RETRIES { log_event( events::layer::EMPTY_FRAME_EXHAUSTED, - json!({ "requestId": request_id, "attempts": retries_used }), + json!({ "requestId": request_id, "attempts": retries_used, "droppedUsage": converter.dropped_usage() }), ); let released = converter.release_deferred_terminal(); if !released.is_empty() @@ -1362,9 +1480,12 @@ fn stream_from_upstream( same_body }; retries_used += 1; + // 决策日志先落：此刻重发还没成功，但「扣帧 + 已选请求体」是 + // 排障最需要的现场——重发失败时这里是唯一现场。droppedUsage + // 带的是截至此刻被丢轮的累计账（扣帧时入账）。 log_event( events::layer::EMPTY_FRAME_RETRY, - json!({ "requestId": request_id, "attempt": retries_used, "nudged": nudged }), + json!({ "requestId": request_id, "attempt": retries_used, "nudged": nudged, "droppedUsage": converter.dropped_usage() }), ); match opener(retry_body).await { Ok(opened) if opened.is_success() => { @@ -1384,7 +1505,7 @@ fn stream_from_upstream( }; log_event( events::layer::EMPTY_FRAME_EXHAUSTED, - json!({ "requestId": request_id, "attempts": retries_used, "reason": reason }), + json!({ "requestId": request_id, "attempts": retries_used, "reason": reason, "droppedUsage": converter.dropped_usage() }), ); let released = converter.release_deferred_terminal(); if !released.is_empty() { @@ -1457,7 +1578,7 @@ fn stream_from_upstream( } else { log_event( events::layer::STREAM_COMPLETED, - json!({ "model": model, "requestId": request_id }), + json!({ "model": model, "requestId": request_id, "attempts": retries_used + 1 }), ); } }); diff --git a/src/protocol_proxy.rs b/src/protocol_proxy.rs index 1950f5e..4cd718e 100644 --- a/src/protocol_proxy.rs +++ b/src/protocol_proxy.rs @@ -1426,6 +1426,16 @@ impl NativeResponsesSseObserver { } } + /// 护栏重开一轮：收口证据只认当前轮的事件。上一轮的 completed 已被扣帧 + /// 丢弃（永不门），留着它会把「重发轮半途断流」误判成已完成，客户端将 + /// 收不到任何终帧——红线所在。未读完的事件前缀照常保留，跨重开边界的 + /// 残帧续读，不重复计入。 + pub fn begin_retry(&mut self) { + self.completed = false; + self.failure = None; + self.unclassified_large_event = false; + } + /// 上游是否已自带终止失败事件（failed / incomplete / error）。 /// 装配点用它判定「还要不要向客户端补一帧」：上游自己已经宣告失败的流，客户端手里 /// 已有显式终止信号，再补一帧只是重复噪音；而什么都没发过的流必须由本层补， @@ -1470,6 +1480,476 @@ pub fn native_stream_failure_frame(message: &str, error_type: &str) -> String { output } +/// 原生透传护栏的单事件可判定上限：超过即整轮退化为原样透传、不再判定。 +/// 上限与观察侧前缀上限同源——reasoning 摘要、图片 Base64 这类大 payload +/// 可能整段落在一个事件里，判定器不许为此无界缓冲。 +const NATIVE_GUARD_EVENT_LIMIT: usize = 2 * 1024 * 1024; + +/// 原生 Responses 透传支路的空终帧护栏：在**事件边界**上缓冲上游字节，判定 +/// 本轮 `response.completed` 是否静默（判据与 Chat 方言共用 +/// `native_response_is_silent` 本体）。静默则把终帧扣下（连同其后已到字节）， +/// 装配层拿到「扣帧」信号后走重发决策；重发轮的条目事件按位移改写、终帧合并 +/// 改写，续排在同一条客户端事件流里，响应身份全程回填为首轮 id。 +/// +/// 兼容口径（PR 正文同步声明）：护栏开启后，原生透传从「chunk 原样」变为 +/// 「事件边界对齐下发」。SSE 事件本就以空行为完整边界，客户端解析器按事件 +/// 消费，不依赖 TCP chunk 切分——这是本层既有「事件完整性」假设的显式化。 +/// +/// 四条红线： +/// 1. 扣帧状态下后到的字节全部并入暂存区，`release_deferred` 原样吐回， +/// 绝不静默断流； +/// 2. 判定读不出结构一律按非静默放行（宁可不护，不可误护）； +/// 3. 上游自报 failed / incomplete / error 的事件不拦不改，原样放行； +/// 4. 装配层的观察者始终看上游**原始字节**，补帧判定不受改写影响。 +#[derive(Default)] +pub struct NativeResponsesGuard { + enabled: bool, + /// 超限退化位：置位后本流永久直通（护栏失能，收口机器照常）。 + disabled_for_stream: bool, + /// 尚未处理的原始字节（事件边界对齐前的残留）。 + buffer: Vec, + /// 边界扫描的水位线：同一残段不反复重扫，流式吞吐不随缓冲长度退化。 + scan_from: usize, + /// 已进行的轮次（0 为首轮）。begin_retry 递增；首轮与改写轮的行为分岔点。 + round: usize, + /// 首轮 created/in_progress 事件里的 response id：改写轮一律回填成它， + /// 客户端全程只认一个响应身份。 + first_response_id: Option, + /// 被丢弃轮的输出条目：终帧合并时垫在前面，保证客户端最终看到的 output + /// 数组与已流式下发过的条目一致（位移量即其长度）。 + carried_items: Vec, + /// 扣下的终帧及其后到达的全部字节。`Some` 即「装配层该走重发决策了」。 + deferred: Option>, + /// 被丢弃轮的 usage 聚合，供诊断事件记账：白烧的 reasoning token 恰是 + /// 本问题最该被盯住的数字。 + dropped_usage: Option, +} + +impl NativeResponsesGuard { + /// 开关由装配层控制（它才知道重发通道、请求体与本轮是不是压缩轮）， + /// 护栏只照开关行事——与 Chat 方言的 `set_guard_enabled` 同一分工。 + pub fn set_enabled(&mut self, enabled: bool) { + self.enabled = enabled; + } + + /// 是否正扣着终帧。true 时装配层必须走重发决策：重开、或放行扣帧收口。 + pub fn deferred(&self) -> bool { + self.deferred.is_some() + } + + /// 是否已进入改写轮（至少丢弃过一个终帧）。仅用于诊断表述。 + pub fn rewriting(&self) -> bool { + self.round > 0 + } + + /// 被丢弃轮的 usage 聚合（无则 None）。 + pub fn dropped_usage(&self) -> Option<&Value> { + self.dropped_usage.as_ref() + } + + /// 放行扣帧字节：暂存内容与缓冲残留原样返回，一个字节不改。 + /// 放弃重试、重发失败、关停排空都必须走这里收口——客户端宁可收到一帧 + /// 静默 completed，也绝不收到「无终帧」的断流。 + pub fn release_deferred(&mut self) -> Vec { + let mut out = self.deferred.take().unwrap_or_default(); + out.append(&mut self.buffer); + self.scan_from = 0; + out + } + + /// 丢弃扣下的终帧，进入下一轮：轮次递增、暂存清空。被丢轮的条目与 usage + /// 已在扣帧判定时入账（见 `handle_event` 的 Completed 分支），这里不许 + /// 再记一遍。 + pub fn begin_retry(&mut self) { + if self.deferred.take().is_none() { + return; + } + self.round += 1; + self.scan_from = 0; + } + + /// 上游本轮的位移量：已垫入的条目数。溢出时返回 None，改写处宁缺毋滥。 + fn index_offset(&self) -> Option { + u32::try_from(self.carried_items.len()).ok() + } + + /// 喂入上游字节，返回此刻可以下发给客户端的字节。 + pub fn push(&mut self, bytes: &[u8]) -> Vec { + if !self.enabled || self.disabled_for_stream { + return bytes.to_vec(); + } + // 红线 1：扣帧状态下后到字节并入暂存区，等待放行或重开裁决。 + if let Some(deferred) = self.deferred.as_mut() { + deferred.extend_from_slice(bytes); + self.buffer.clear(); + self.scan_from = 0; + return Vec::new(); + } + self.buffer.extend_from_slice(bytes); + if self.buffer.len() > NATIVE_GUARD_EVENT_LIMIT { + // 迟迟等不到事件边界的巨量字节：整段原样放行并永久失能。 + // 观察者照常旁路读原始字节，收口补帧机器不受影响。 + self.disabled_for_stream = true; + self.scan_from = 0; + return std::mem::take(&mut self.buffer); + } + let mut output = Vec::new(); + while let Some(boundary) = self.take_boundary() { + let raw = &self.buffer[..boundary]; + let Some(event) = classify_native_event(raw) else { + // 读不出任何已知形态（含非 UTF-8）：原样放行，绝不猜测。 + output.extend_from_slice(&self.drain_event(boundary)); + continue; + }; + match self.handle_event(&event, boundary) { + EventAction::Forward => { + output.extend_from_slice(&self.drain_event(boundary)); + } + EventAction::Drop => { + self.drain_event(boundary); + } + EventAction::Rewrite { bytes: rebuilt } => { + self.drain_event(boundary); + output.extend_from_slice(&rebuilt); + } + EventAction::Defer { bytes } => { + // 扣帧：终帧（原始或合并改写形态）连同其后的全部残留进 + // 暂存区，本轮处理终止。 + let rest = self.buffer.split_off(boundary); + let mut deferred = match bytes { + Some(bytes) => { + self.buffer.clear(); + self.scan_from = 0; + bytes + } + None => std::mem::replace(&mut self.buffer, Vec::new()), + }; + deferred.extend_from_slice(&rest); + self.deferred = Some(deferred); + self.scan_from = 0; + return output; + } + } + } + output + } + + /// 从当前缓冲里找下一个事件边界，返回含分隔符的帧长；同步推进扫描水位。 + fn take_boundary(&mut self) -> Option { + let start = self.scan_from.min(self.buffer.len()); + let mut lf = None; + let mut crlf = None; + if let Some(pos) = self.buffer[start..].windows(2).position(|w| w == b"\n\n") { + lf = Some(start + pos + 2); + } + if let Some(pos) = self.buffer[start..].windows(4).position(|w| w == b"\r\n\r\n") { + crlf = Some(start + pos + 4); + } + let boundary = match (lf, crlf) { + (Some(a), Some(b)) => a.min(b), + (Some(a), None) => a, + (None, Some(b)) => b, + (None, None) => { + // 无边界：水位退到尾部 3 字节，防跨 push 的 `\n\n` 被漏看。 + self.scan_from = self.buffer.len().saturating_sub(3); + return None; + } + }; + self.scan_from = boundary; + Some(boundary) + } + + /// 从缓冲头部摘走一个完整帧（含分隔符）并返回其字节。 + fn drain_event(&mut self, boundary: usize) -> Vec { + let rest = self.buffer.split_off(boundary); + let event = std::mem::replace(&mut self.buffer, rest); + self.scan_from = 0; + event + } + + fn handle_event(&mut self, event: &NativeEventInfo, boundary: usize) -> EventAction { + let oversized = boundary > NATIVE_GUARD_EVENT_LIMIT; + match event.kind { + NativeEventKind::Created | NativeEventKind::InProgress => { + if oversized { + return EventAction::Forward; + } + if self.round == 0 { + if self.first_response_id.is_none() { + self.first_response_id = event.response_id().map(str::to_string); + } + EventAction::Forward + } else { + // 改写轮的身份事件（含重复 created）：客户端已见过身份，丢弃。 + if self.first_response_id.is_none() { + self.first_response_id = event.response_id().map(str::to_string); + } + EventAction::Drop + } + } + NativeEventKind::Completed => { + if oversized { + // 巨帧无法判定：放行等于本轮不护，但绝不误伤成断流。 + return EventAction::Forward; + } + let silent = event + .payload + .as_ref() + .and_then(|payload| payload.get("response")) + .is_some_and(|resp| native_response_is_silent(resp)); + if self.round == 0 { + if silent { + // 扣帧即入账：装配层在重发决策点就要拿到被丢轮的账， + // 条目同步垫入基底（首轮无基底，合并帧与原始帧同形）。 + self.record_dropped_usage(event_response_field(event, "usage")); + if let Some(items) = + event_response_field(event, "output").and_then(Value::as_array) + { + self.carried_items.extend(items.iter().cloned()); + } + EventAction::Defer { bytes: None } + } else { + EventAction::Forward + } + } else { + let Some(payload) = event.payload.clone() else { + return EventAction::Forward; + }; + let mut merged = payload; + rewrite_completed_payload( + &mut merged, + &self.carried_items, + self.first_response_id.as_deref(), + ); + let still_silent = merged + .get("response") + .is_some_and(|resp| native_response_is_silent(resp)); + if still_silent { + // 扣帧即入账（usage 取原始 payload：合并帧会把已入账轮 + // 算两遍）；本轮条目垫入基底，供下一轮继续续排。 + self.record_dropped_usage(event_response_field(event, "usage")); + if let Some(items) = + event_response_field(event, "output").and_then(Value::as_array) + { + self.carried_items.extend(items.iter().cloned()); + } + // 暂存区收合并改写帧：放弃重试时放行的也是它——客户端 + // 已按位移收过本轮增量事件，原始轮帧会对不上账。 + EventAction::Defer { + bytes: Some(rebuild_sse_event(event, &merged)), + } + } else { + EventAction::Rewrite { + bytes: rebuild_sse_event(event, &merged), + } + } + } + } + // 红线 3：上游自报失败不拦不改，装配层的补帧判定照常接管。 + NativeEventKind::TerminalFailure => EventAction::Forward, + NativeEventKind::Done => { + // [DONE] 永不会先于终帧出门：事件按到达顺序处理，同缓冲时它 + // 必在 completed 之后；若本轮判成静默扣帧，它落进暂存区随终帧 + // 一起走（重试时被丢轮连带丢弃，放行时原序吐回）。 + EventAction::Forward + } + NativeEventKind::Other => { + if oversized || self.round == 0 { + return EventAction::Forward; + } + match shift_event_output_indices(event, self.index_offset()) { + Some(shifted) => EventAction::Rewrite { + bytes: rebuild_sse_event(event, &shifted), + }, + None => EventAction::Forward, + } + } + } + } + + /// 被丢弃轮 usage 的累加：只加 reasoning/total 两口径。丢掉的轮不产出 + /// 交付，但白烧的 token 必须可见。 + fn record_dropped_usage(&mut self, usage: Option<&Value>) { + accumulate_dropped_usage(&mut self.dropped_usage, usage); + } +} + +/// 被丢弃轮 usage 的累加本体：只加 reasoning/total 两口径。丢掉的轮不产出 +/// 交付，但白烧的 token 必须可见；两套方言的护栏共用这一条记账。 +fn accumulate_dropped_usage(accumulated: &mut Option, usage: Option<&Value>) { + let Some(usage) = usage.filter(|usage| usage.is_object()) else { + return; + }; + let accumulated = accumulated.get_or_insert_with(|| json!({})); + for key in ["reasoning_tokens", "total_tokens"] { + let add = usage + .get("output_tokens_details") + .and_then(|details| details.get(key)) + .or_else(|| usage.get(key)) + .and_then(Value::as_u64); + let Some(add) = add else { continue }; + let current = accumulated + .get(key) + .and_then(Value::as_u64) + .unwrap_or(0); + accumulated[key] = json!(current.saturating_add(add)); + } +} + +/// 一个已对齐事件的读后形态：event 行、解析出的 payload、事件类别、 +/// 帧尾分隔符形态（重建时保持原样，CRLF 与 LF 不互换）。 +struct NativeEventInfo { + event_name: Option, + payload: Option, + kind: NativeEventKind, + ends_with_crlf: bool, +} + +enum NativeEventKind { + Created, + InProgress, + Completed, + TerminalFailure, + Done, + Other, +} + +impl NativeEventInfo { + /// 事件 payload 里的 response id（`response.id` 优先，顶层 `id` 兜底）。 + fn response_id(&self) -> Option<&str> { + let payload = self.payload.as_ref()?; + payload + .get("response") + .and_then(|resp| resp.get("id")) + .and_then(Value::as_str) + .or_else(|| payload.get("id").and_then(Value::as_str)) + } +} + +enum EventAction { + Forward, + Drop, + Rewrite { bytes: Vec }, + /// `bytes` 为 `Some` 时用改写后的终帧入暂存区（改写轮放行必须拿到合并帧）， + /// `None` 时按事件原始字节扣帧（首轮的原始字节与合并帧同形）。 + Defer { bytes: Option> }, +} + +/// 把一个原始事件块读成 `NativeEventInfo`。返回 `None` 表示连 UTF-8 都不是， +/// 调用方必须原样透传（护栏对不可读字节零动作）。 +fn classify_native_event(raw: &[u8]) -> Option { + let text = std::str::from_utf8(raw).ok()?; + let event_name = text + .lines() + .find_map(|line| strip_sse_field(line, "event")) + .map(str::trim) + .map(str::to_string); + let data = text + .lines() + .filter_map(|line| strip_sse_field(line, "data")) + .collect::>() + .join("\n"); + let payload = if data.is_empty() { + None + } else { + serde_json::from_str::(&data).ok() + }; + // 事件类别判定与观察侧同口径：payload 顶层 type 优先，event 行兜底。 + let type_name = payload + .as_ref() + .and_then(|parsed| parsed.get("type").and_then(Value::as_str)) + .or(event_name.as_deref()); + let kind = if !data.is_empty() && data.trim() == "[DONE]" { + NativeEventKind::Done + } else { + match type_name { + Some("response.created") => NativeEventKind::Created, + Some("response.in_progress") => NativeEventKind::InProgress, + Some("response.completed") => NativeEventKind::Completed, + Some("response.failed") | Some("response.incomplete") | Some("error") => { + NativeEventKind::TerminalFailure + } + _ => NativeEventKind::Other, + } + }; + Some(NativeEventInfo { + event_name, + payload, + kind, + ends_with_crlf: raw.ends_with(b"\r\n\r\n"), + }) +} + +/// 读事件 payload 里 `response` 对象的一个字段（护栏只在 completed 判定处用）。 +fn event_response_field<'a>(event: &'a NativeEventInfo, field: &str) -> Option<&'a Value> { + event + .payload + .as_ref() + .and_then(|payload| payload.get("response")) + .and_then(|resp| resp.get(field)) +} + +/// 改写轮的终帧合并：本轮 output 条目按全局位移垫到 carried 之后， +/// response.id 回填为首轮身份。位移写进每个条目的 `output_index`， +/// 与已流式下发过的条目序号衔接。 +fn rewrite_completed_payload(payload: &mut Value, carried: &[Value], first_id: Option<&str>) { + let Some(response) = payload.get_mut("response") else { + return; + }; + if let Some(id) = first_id { + response["id"] = json!(id); + } + let own: Vec = response + .get("output") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let mut merged = carried.to_vec(); + for (pos, item) in own.into_iter().enumerate() { + let mut item = item; + // 只在条目自带 output_index 时位移：数组位置本身就是全局序号， + // 上游没写的字段不许替它发明。 + if item.get("output_index").is_some() + && let Some(index) = shifted_index(carried.len(), pos) + { + item["output_index"] = json!(index); + } + merged.push(item); + } + response["output"] = json!(merged); +} + +/// carried 长度 + 轮内序号 → 位移后的全局 index（u32 溢出时不写，宁缺毋滥）。 +fn shifted_index(carried_len: usize, pos: usize) -> Option { + u32::try_from(carried_len.saturating_add(pos)).ok() +} + +/// 改写轮的增量事件位移：顶层 `output_index` 抬到位移量。payload 缺失、 +/// 没有顶层 index、或位移量溢出——一律返回 None 走原样放行。 +fn shift_event_output_indices(event: &NativeEventInfo, offset: Option) -> Option { + let offset = offset?; + let mut shifted = event.payload.clone()?; + let top = shifted.get_mut("output_index")?; + *top = json!(offset); + Some(shifted) +} + +/// 用改写后的 payload 重建事件块，保持原始的事件边界字节形态（CRLF 或 LF）。 +fn rebuild_sse_event(event: &NativeEventInfo, payload: &Value) -> Vec { + let separator = if event.ends_with_crlf { "\r\n\r\n" } else { "\n\n" }; + let mut out = String::new(); + if let Some(name) = &event.event_name { + out.push_str("event: "); + out.push_str(name); + out.push('\n'); + } + out.push_str("data: "); + out.push_str(&serde_json::to_string(payload).unwrap_or_default()); + out.push_str(separator); + out.into_bytes() +} + + /// Responses 非流式 JSON 重组为 SSE 事件文本。 /// /// 为什么必须有这一步：`upstream_compaction_item_count` 是 SSE 感知的计数器， @@ -2703,6 +3183,12 @@ impl ChatSseToResponsesConverter { self.state.begin_retry_attempt(); } + /// 被丢弃轮的 usage 聚合（无则 None）。诊断事件 `empty_frame_retry` 靠它 + /// 记账：丢掉的轮不产出交付，但白烧的 token 必须可见。 + pub fn dropped_usage(&self) -> Option<&Value> { + self.state.dropped_usage.as_ref() + } + pub fn push_bytes(&mut self, bytes: &[u8]) -> Vec { append_utf8_safe(&mut self.buffer, &mut self.utf8_remainder, bytes); let mut output = String::new(); @@ -2784,6 +3270,12 @@ impl ChatSseToResponsesConverter { return; } if self.state.empty_final_frame() { + // 白烧的 reasoning token 在扣帧这一刻就进诊断账：retry 事件在重发 + // 决策点出门（早于重开成功），账必须先行，否则事件只能带 null。 + accumulate_dropped_usage( + &mut self.state.dropped_usage, + self.state.latest_usage.as_ref(), + ); self.deferred_terminal = Some(scratch); } else { output.push_str(&scratch); @@ -2795,10 +3287,40 @@ impl ChatSseToResponsesConverter { /// 上限存在的意义是让「模型反复静默」的病态循环有代价上界，而不是无限重放。 pub const EMPTY_FRAME_MAX_RETRIES: usize = 2; -/// 第二次重试的推动文案：同体重采样（第一次重试）可能再次静默，第二次在 -/// `input` 尾部显式要求「可见正文或工具调用」。只拼进重发请求，客户端历史里 -/// 不会出现这条消息。 -pub const EMPTY_FRAME_NUDGE_TEXT: &str = "系统提示：你上一段输出只包含思考内容，既无正文回复也无工具调用，会话因此被自动结束。请现在立即行动：要么给出面向用户的正文，要么发起必要的工具调用；不允许再次只输出思考内容。"; +/// 第二次重试推动文案（中文形态）。两条约束来自审查定档： +/// - 不得含「会话被自动结束」这类元叙述——那是把中继内部机制塞进模型上下文； +/// - 必须显式禁止复述本条提示——推动项以 user 消息注入，防被当成用户指令回显。 +const EMPTY_FRAME_NUDGE_TEXT_ZH: &str = + "请立即行动：直接给出面向用户的正文，或发起必要的工具调用。不要只输出思考内容，也不要提及或复述本条提示。"; +const EMPTY_FRAME_NUDGE_TEXT_EN: &str = + "Act now: produce the visible answer for the user or call the necessary tools. Do not emit thinking only, and do not mention or repeat this note."; + +/// 本层推动项的识别标记：推动文案的尾部含此标记，注入前会先剥除历史同类项， +/// 防止被污染的会话历史里推动消息跨轮累积。 +const EMPTY_FRAME_NUDGE_MARK: &str = "[relay-empty-frame-nudge]"; + +/// 推动文案语言选择：从请求尾部向前扫一段，有 CJK 即判中文会话，否则英文。 +/// 只看尾部是因为推动文案只与「用户最近的话」同屏；语言不会在一条会话里 +/// 隔 2K 字符突变，扫描预算定在尾部 2048 字符。 +fn nudge_text_for(request_json: &Value) -> &'static str { + let is_cjk = |c: char| { + matches!( + c as u32, + 0x3040..=0x30FF | 0x3400..=0x4DBF | 0x4E00..=0x9FFF | 0xF900..=0xFAFF + ) + }; + if request_json + .to_string() + .chars() + .rev() + .take(2048) + .any(is_cjk) + { + EMPTY_FRAME_NUDGE_TEXT_ZH + } else { + EMPTY_FRAME_NUDGE_TEXT_EN + } +} /// 空终帧护栏的判据本体（纯函数）：finalize 后的「completed」终帧是不是 /// 静默帧——没有任何工具调用项、也没有非空可见正文，只有 reasoning（或什么都没有）。 @@ -2825,43 +3347,82 @@ pub fn silent_final_frame(items: &[Value], finish_reason: Option<&str>) -> bool }) } -/// message 项是否携带非空可见正文。读不出 content 数组时按「无正文」处理： +/// message 项是否携带非空可见内容。读不出 content 数组时按「无正文」处理： /// 宁可多重试一次，也不能把读得出的静默帧漏判成有效输出。 +/// `refusal` 部件算可见内容（非空即有效输出）：合规的拒绝绝不能被判成静默帧 +/// 重发——那等于拿推动文案顶回一次拒绝，可能顶穿安全边界。 fn message_item_has_visible_text(item: &Value) -> bool { + // 字符串形态的 content：非空即有可见文本（上游形态漂移的防御分支）。 + if let Some(text) = item.get("content").and_then(Value::as_str) { + return !text.trim().is_empty(); + } item.get("content") .and_then(Value::as_array) .is_some_and(|parts| { parts.iter().any(|part| { - part.get("type") - .and_then(Value::as_str) - == Some("output_text") - && part - .get("text") - .and_then(Value::as_str) - .is_some_and(|text| !text.trim().is_empty()) + let text = match part.get("type").and_then(Value::as_str) { + Some("output_text") => part.get("text").and_then(Value::as_str), + Some("refusal") => part.get("refusal").and_then(Value::as_str), + _ => None, + }; + text.is_some_and(|text| !text.trim().is_empty()) }) }) } -/// 第二次重试的推动请求体：克隆客户端原请求，往 `input` 尾部追加一条用户 -/// 消息（形态与既有压缩摘要指令同款）。`input` 不是数组时返回 None——没有 -/// 可追加的位置，调用方退化为同体重采样，绝不自造 `input` 结构。 +/// 原生方言的静默终帧判据：终态必须是 completed，且 output 条目列表静默 +/// （与 Chat 方言共用 `silent_final_frame` 这一条判据本体，两处不宽窄不一）。 +pub fn native_response_is_silent(response: &Value) -> bool { + response.get("status").and_then(Value::as_str) == Some("completed") + && silent_final_frame( + response + .get("output") + .and_then(Value::as_array) + .map(Vec::as_slice) + .unwrap_or(&[]), + None, + ) +} + +/// 第二次重试的推动请求体：克隆客户端原请求，先剥除历史里本层此前注入的同类 +/// 推动项（被污染的会话历史不得让推动消息跨轮累积），再往 `input` 尾部追加 +/// 一条用户消息（形态与既有压缩摘要指令同款，文案语言跟随会话）。 +/// `input` 不是数组时返回 None——没有可追加的位置，调用方退化为同体重采样， +/// 绝不自造 `input` 结构。 pub fn with_empty_frame_nudge(request_json: &Value) -> Option { if !request_json.get("input").is_some_and(Value::is_array) { return None; } let mut nudged = request_json.clone(); - nudged["input"] - .as_array_mut() - .expect("input 已判过是数组") - .push(json!({ + { + let items = nudged["input"].as_array_mut().expect("input 已判过是数组"); + items.retain(|item| !is_relay_nudge_item(item)); + items.push(json!({ "type": "message", "role": "user", - "content": [{ "type": "input_text", "text": EMPTY_FRAME_NUDGE_TEXT }], + "content": [{ "type": "input_text", "text": format!("{} {}", nudge_text_for(request_json), EMPTY_FRAME_NUDGE_MARK) }], })); + } Some(nudged) } +/// 是否为本层注入的推动项：user 消息且任一文本部件含识别标记。 +fn is_relay_nudge_item(item: &Value) -> bool { + if item.get("role").and_then(Value::as_str) != Some("user") { + return false; + } + let has_mark = |text: &str| text.contains(EMPTY_FRAME_NUDGE_MARK); + match item.get("content") { + Some(Value::String(text)) => has_mark(text), + Some(Value::Array(parts)) => parts.iter().any(|part| { + part.get("text") + .and_then(Value::as_str) + .is_some_and(has_mark) + }), + _ => false, + } +} + pub fn is_responses_proxy_path(path: &str) -> bool { let path = path.split_once('?').map_or(path, |(path, _)| path); matches!( @@ -4198,6 +4759,8 @@ struct ChatSseState { tools: BTreeMap, output_items: Vec<(u32, Value)>, latest_usage: Option, + /// 被护栏丢弃的重试轮 usage 聚合（诊断记账，不进交付） + dropped_usage: Option, finish_reason: Option, tool_context: CodexToolContext, original_request: Option, @@ -4219,6 +4782,7 @@ impl Default for ChatSseState { tools: BTreeMap::new(), output_items: Vec::new(), latest_usage: None, + dropped_usage: None, finish_reason: None, tool_context: CodexToolContext::default(), original_request: None, @@ -6576,12 +7140,17 @@ fn mark_current_user_turn(messages: &mut Vec) { // 等待工具结果后的继续请求末尾是 tool/assistant 角色，此时最后一条 user // 是历史旧问题，钉上会把历史问题误导成「当前问题」。 // 非 user 收尾：解包已完成，直接返回。 - match messages - .last() - .and_then(|message| message.get("role").and_then(Value::as_str)) - { - Some("user") => wrap_current_turn_mark(messages.last_mut().unwrap()), - _ => return, + // 护栏推动项豁免：空终帧重发的请求以本层注入的推动消息收尾。轮次标记 + // 属于「真实用户指令」，绝不钉到推动项上——那会把中继提示当成当前指令。 + // 豁免后回到原始请求的收尾形态判定：推动项之前若以 user 收尾就钉那条 + // （与非重试轮逐字节同形，前缀缓存不受重发影响）；是非 user 收尾 + // （等工具结果的续轮）就维持「不以 user 收尾不钉」的既有语义。 + let mut tail = messages.len(); + while tail > 0 && is_relay_nudge_item(&messages[tail - 1]) { + tail -= 1; + } + if tail > 0 && messages[tail - 1].get("role").and_then(Value::as_str) == Some("user") { + wrap_current_turn_mark(&mut messages[tail - 1]); } } @@ -10284,7 +10853,16 @@ mod empty_frame_guard_tests { let messages = converted["messages"].as_array().unwrap(); let last = messages.last().unwrap(); assert_eq!(last["role"], json!("user")); - assert_eq!(last["content"].as_str().unwrap(), EMPTY_FRAME_NUDGE_TEXT); + // 会话含中文 → 选中文文案；识别标记必须随文案落地（历史剥除靠它）。 + let text = last["content"].as_str().unwrap(); + assert!( + text.contains(EMPTY_FRAME_NUDGE_TEXT_ZH), + "中文会话必须选中文文案：{text}" + ); + assert!( + text.contains(EMPTY_FRAME_NUDGE_MARK), + "推动项必须带识别标记：{text}" + ); } #[test] @@ -10408,3 +10986,362 @@ mod empty_frame_guard_tests { assert!(released.contains("data: [DONE]")); } } + +#[cfg(test)] +mod native_empty_frame_guard_tests { + use super::*; + + fn sse_event(name: &str, payload: Value) -> String { + format!("event: {name}\ndata: {payload}\n\n") + } + + fn created(id: &str) -> String { + sse_event( + "response.created", + json!({"type": "response.created", "response": {"id": id, "status": "in_progress"}}), + ) + } + + fn completed(id: &str, response_extra: Value) -> String { + let mut response = json!({"id": id, "status": "completed"}); + for (key, value) in response_extra.as_object().expect("附加字段必须是对象") { + response[key] = value.clone(); + } + sse_event("response.completed", json!({"type": "response.completed", "response": response})) + } + + fn reasoning_item() -> Value { + json!({"type": "reasoning", "id": "r1"}) + } + + fn text_item(text: &str) -> Value { + json!({"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}) + } + + /// 把一个 SSE 文本流切成下发 chunk（护栏必须自己按事件边界对齐， + /// chunk 怎么切都不许影响判定）。 + fn chunks_of(stream: &str) -> Vec> { + stream.as_bytes().chunks(7).map(<[u8]>::to_vec).collect() + } + + fn push_all(guard: &mut NativeResponsesGuard, stream: &str) -> Vec { + let mut out = Vec::new(); + for chunk in chunks_of(stream) { + out.extend_from_slice(&guard.push(&chunk)); + } + out + } + + /// 从一段输出里定位 `response.completed` 事件并解析其 payload。 + /// 增量事件与终帧混在同一段里，按事件名取帧，别拿错 data 行。 + fn parse_completed(haystack: &[u8]) -> Value { + let text = String::from_utf8(haystack.to_vec()).unwrap(); + for block in split_sse_events(&text) { + let is_completed = block + .lines() + .find_map(|line| strip_sse_field(line, "event")) + .map(str::trim) + == Some("response.completed"); + if is_completed { + return parse_sse_event_data(&block).expect("终帧必须是合法 JSON"); + } + } + panic!("输出里必须有一个 response.completed 事件：{text}") + } + + #[test] + fn native_silence_judgement_covers_all_shapes() { + // reasoning-only / 空 output / 缺 output：都是「客户端不会为它续轮」的静默收尾。 + assert!(native_response_is_silent( + &json!({"status": "completed", "output": [reasoning_item()]}) + )); + assert!(native_response_is_silent( + &json!({"status": "completed", "output": []}) + )); + assert!(native_response_is_silent(&json!({"status": "completed"}))); + // 空白正文按无正文处理（trim 判空与 Chat 方言同口径）。 + assert!(native_response_is_silent(&json!({ + "status": "completed", + "output": [text_item(" ")] + }))); + // 可见正文 / 非空字符串 content / 工具调用项：一律有效输出。 + assert!(!native_response_is_silent(&json!({ + "status": "completed", + "output": [text_item("答案")] + }))); + assert!(!native_response_is_silent(&json!({ + "status": "completed", + "output": [{"type": "message", "content": "纯字符串正文"}] + }))); + assert!(!native_response_is_silent(&json!({ + "status": "completed", + "output": [{"type": "function_call", "name": "exec_command"}] + }))); + // refusal 是合规拒绝，不是静默：误判会拿推动文案顶穿安全边界。 + assert!(!native_response_is_silent(&json!({ + "status": "completed", + "output": [{"type": "message", "content": [{"type": "refusal", "refusal": "不能协助该请求"}]}] + }))); + // 非 completed 终态（截断/失败）不由护栏接管。 + assert!(!native_response_is_silent( + &json!({"status": "incomplete", "output": []}) + )); + assert!(!native_response_is_silent( + &json!({"status": "failed", "output": []}) + )); + } + + #[test] + fn guard_disabled_is_byte_identical() { + // 红线：开关不开时逐字节不变——静默终帧也即刻下发。 + let mut guard = NativeResponsesGuard::default(); + let stream = format!("{}{}", created("rA"), completed("rA", json!({"output": []}))); + let mut out = Vec::new(); + for chunk in chunks_of(&stream) { + out.extend_from_slice(&guard.push(&chunk)); + } + assert_eq!(out, stream.as_bytes()); + assert!(!guard.deferred()); + } + + #[test] + fn silent_completed_is_deferred_and_released_byte_identical() { + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let first = format!( + "{}{}", + created("rA"), + sse_event( + "response.output_item.done", + json!({"type": "response.output_item.done", "output_index": 0, "item": reasoning_item()}) + ) + ); + let out = push_all(&mut guard, &first); + assert_eq!(out, first.as_bytes(), "非终帧事件必须原样即刻下发"); + + let tail = format!("{}data: [DONE]\n\n", completed("rA", json!({"output": [reasoning_item()]}))); + let out = push_all(&mut guard, &tail); + assert!(out.is_empty(), "静默终帧及其后的 [DONE] 一并扣下"); + assert!(guard.deferred()); + + let released = guard.release_deferred(); + assert_eq!(released, tail.as_bytes(), "放行必须逐字节原样，绝不静默断流"); + assert!(!guard.deferred()); + } + + #[test] + fn non_silent_completed_passes_through_untouched() { + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let stream = format!( + "{}{}", + created("rA"), + completed("rA", json!({"output": [text_item("答案")]})) + ); + let out = push_all(&mut guard, &stream); + assert_eq!(out, stream.as_bytes()); + assert!(!guard.deferred()); + assert!(!guard.rewriting()); + } + + #[test] + fn retry_round_merges_items_shifts_indices_and_backfills_id() { + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let round1 = format!( + "{}{}{}", + created("resp_first"), + sse_event( + "response.output_item.done", + json!({"type": "response.output_item.done", "output_index": 0, "item": reasoning_item()}) + ), + completed("resp_first", json!({ + "output": [reasoning_item()], + "usage": {"output_tokens_details": {"reasoning_tokens": 120}, "total_tokens": 5000} + })) + ); + let _ = push_all(&mut guard, &round1); + assert!(guard.deferred()); + assert_eq!( + guard.dropped_usage(), + Some(&json!({"reasoning_tokens": 120, "total_tokens": 5000})), + "白烧的 reasoning token 必须在扣帧时入账" + ); + + guard.begin_retry(); + assert!(guard.rewriting()); + + let round2_item = "event: response.output_item.added\r\ndata: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"m2\"}}\r\n\r\n"; + let round2 = format!( + "{}{}{}", + created("resp_second"), + round2_item, + completed("resp_second", json!({ + "output": [{"type": "message", "id": "m2", "output_index": 0, "content": [{"type": "output_text", "text": "第二轮正文"}]}], + "usage": {"output_tokens_details": {"reasoning_tokens": 30}, "total_tokens": 900} + })) + ); + let out = push_all(&mut guard, &round2); + let out_text = String::from_utf8(out).unwrap(); + + // 第二轮的 created 不许出门（客户端已认得这个响应身份）。 + assert!(!out_text.contains("response.created"), "{out_text}"); + // 增量事件位移保留 CRLF 形态，output_index 抬到位移量。 + assert!(out_text.contains("\"output_index\":1"), "{out_text}"); + assert!(out_text.contains("\r\n\r\n"), "事件边界字节形态必须保持"); + // 终帧合并：id 回填首轮，output 前垫第一轮条目、本轮条目接位。 + let merged = parse_completed(out_text.as_bytes()); + let response = &merged["response"]; + assert_eq!(response["id"], json!("resp_first")); + let output = response["output"].as_array().unwrap(); + assert_eq!(output.len(), 2); + assert_eq!(output[0]["type"], json!("reasoning")); + assert_eq!(output[1]["id"], json!("m2")); + assert_eq!(output[1]["output_index"], json!(1)); + assert!(!guard.deferred()); + } + + #[test] + fn back_to_back_silence_defers_merged_frame_and_accumulates_usage() { + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let round1 = format!( + "{}{}", + created("rA"), + completed("rA", json!({ + "output": [reasoning_item()], + "usage": {"output_tokens_details": {"reasoning_tokens": 100}, "total_tokens": 4000} + })) + ); + let _ = push_all(&mut guard, &round1); + guard.begin_retry(); + + let round2 = completed("rB", json!({ + "output": [{"type": "reasoning", "id": "r2"}], + "usage": {"output_tokens_details": {"reasoning_tokens": 50}, "total_tokens": 1500} + })); + let out = push_all(&mut guard, &round2); + assert!(out.is_empty()); + assert!(guard.deferred()); + assert_eq!( + guard.dropped_usage(), + Some(&json!({"reasoning_tokens": 150, "total_tokens": 5500})), + "连续静默的账必须累加" + ); + // 放弃重试时放行的是合并帧：客户端已按位移收过条目，轮帧对不上账。 + let released = guard.release_deferred(); + let merged = parse_completed(&released); + assert_eq!(merged["response"]["id"], json!("rA")); + let output = merged["response"]["output"].as_array().unwrap(); + assert_eq!(output.len(), 2); + assert_eq!(output[1]["id"], json!("r2")); + } + + #[test] + fn upstream_failure_event_is_never_rewritten() { + // 红线：上游自报失败不拦不改，收口机器照常接管。 + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let round1 = format!( + "{}{}", + created("rA"), + completed("rA", json!({"output": [reasoning_item()]})) + ); + let _ = push_all(&mut guard, &round1); + guard.begin_retry(); + let failure = sse_event( + "response.failed", + json!({"type": "response.failed", "response": {"id": "rB", "status": "failed"}}), + ); + let out = push_all(&mut guard, &failure); + assert_eq!(out, failure.as_bytes()); + assert!(!guard.deferred()); + } + + #[test] + fn oversized_burst_without_boundary_degrades_to_passthrough() { + let mut guard = NativeResponsesGuard::default(); + guard.set_enabled(true); + let burst = vec![b'x'; NATIVE_GUARD_EVENT_LIMIT + 1]; + let out = guard.push(&burst); + assert_eq!(out, burst, "等不到边界的巨量字节必须原样放行"); + // 退化后整流直通：后续事件不再判定。 + let stream = completed("rA", json!({"output": []})); + let out = guard.push(stream.as_bytes()); + assert_eq!(out, stream.as_bytes()); + assert!(!guard.deferred()); + } + + #[test] + fn nudge_language_follows_request_and_mark_strips_history() { + let zh = json!({"input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "继续盯住这个任务"}]}]}); + let nudged = with_empty_frame_nudge(&zh).unwrap(); + let items = nudged["input"].as_array().unwrap(); + let tail_text = items.last().unwrap()["content"][0]["text"].as_str().unwrap(); + assert!(tail_text.contains(EMPTY_FRAME_NUDGE_TEXT_ZH), "{tail_text}"); + assert!(tail_text.contains(EMPTY_FRAME_NUDGE_MARK), "{tail_text}"); + + let en = json!({"input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "keep going on this task"}]}]}); + let nudged = with_empty_frame_nudge(&en).unwrap(); + let tail_text = nudged["input"].as_array().unwrap().last().unwrap()["content"][0]["text"] + .as_str() + .unwrap(); + assert!(tail_text.contains(EMPTY_FRAME_NUDGE_TEXT_EN), "{tail_text}"); + + // 历史里已注入过的推动项必须剥除，不跨轮累积。 + let polluted = json!({"input": [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "真问题"}]}, + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": format!("{} {}", EMPTY_FRAME_NUDGE_TEXT_ZH, EMPTY_FRAME_NUDGE_MARK)}]}, + ]}); + let nudged = with_empty_frame_nudge(&polluted).unwrap(); + let items = nudged["input"].as_array().unwrap(); + assert_eq!(items.len(), 2, "剥一条、追一条：总数不涨"); + assert_eq!(items[0]["content"][0]["text"], json!("真问题")); + assert!( + items[1]["content"][0]["text"].as_str().unwrap().contains(EMPTY_FRAME_NUDGE_MARK) + ); + } + + #[test] + fn turn_mark_skips_relay_nudge_item() { + // 推动项收尾 + 前面是真实 user 指令：标记钉在真实指令上（与非重试轮同形）。 + let nudge = json!({"role": "user", "content": [{"type": "text", "text": format!("{} {}", EMPTY_FRAME_NUDGE_TEXT_ZH, EMPTY_FRAME_NUDGE_MARK)}]}); + let mut messages = vec![ + json!({"role": "system", "content": [{"type": "text", "text": "前言"}]}), + json!({"role": "user", "content": [{"type": "text", "text": "真问题"}]}), + nudge.clone(), + ]; + mark_current_user_turn(&mut messages); + assert!( + messages[1]["content"][0]["text"].as_str().unwrap().starts_with(CURRENT_TURN_OPEN), + "标记必须钉在真实用户指令上" + ); + assert!( + !messages[2]["content"][0]["text"].as_str().unwrap().contains(CURRENT_TURN_OPEN), + "推动项绝不携带轮次标记" + ); + + // 推动项收尾 + 前面是工具结果（等工具后续轮）：维持「不以 user 收尾不钉」。 + let mut messages = vec![ + json!({"role": "user", "content": [{"type": "text", "text": "旧问题"}]}), + json!({"role": "tool", "content": [{"type": "text", "text": "工具结果"}]}), + nudge, + ]; + mark_current_user_turn(&mut messages); + for message in &messages { + let texts: Vec<&str> = message["content"] + .as_array() + .map(|parts| { + parts + .iter() + .filter_map(|part| part["text"].as_str()) + .collect() + }) + .unwrap_or_default(); + assert!( + texts.iter().all(|text| !text.contains(CURRENT_TURN_OPEN)), + "非 user 收尾（剔除推动项后）不得钉标记" + ); + } + } +} diff --git a/tests/native_empty_frame_guard.rs b/tests/native_empty_frame_guard.rs new file mode 100644 index 0000000..e19053d --- /dev/null +++ b/tests/native_empty_frame_guard.rs @@ -0,0 +1,392 @@ +//! 原生 Responses 方言下空终帧护栏的集成测试（issue #83 的 P0 本体）。 +//! +//! 本机 dogfood 全程是原生方言（MLX 直接回 Responses SSE），护栏必须住在这条 +//! 透传支路上。断言的是**客户端看到什么**与**上游实际收到什么**： +//! 1) 首轮静默 completed（只有 reasoning）→ 同体重发一轮，客户端拿到单条 +//! created + 单个合并终帧（首轮条目垫底、本轮条目接位、id 回填首轮）， +//! 全程无 failed、无重复身份事件； +//! 2) 连续静默打满额度 → 仍下发一个 completed 终帧（合并形态），绝不静默断流； +//! 第二次重发带推动文案，且推动项只出现在重发体里，不进客户端历史。 + +use serde_json::{Value, json}; +use std::io::Read; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +struct ChildGuard(Child); + +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn reserve_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +/// 把一次上游回包整成原始 HTTP 报文。 +fn raw_response(status_line: &str, content_type: &str, body: &str) -> Vec { + format!( + "HTTP/1.1 {status_line}\r\ncontent-type: {content_type}\r\nconnection: close\r\ncontent-length: {}\r\n\r\n{body}", + body.len() + ) + .into_bytes() +} + +fn sse_response(body: &str) -> Vec { + raw_response("200 OK", "text/event-stream", body) +} + +/// 按脚本回包的假上游：每次请求按顺序取一条响应原样写回，并抓取请求体。 +async fn mock_upstream( + responses: Vec>, +) -> (std::net::SocketAddr, tokio::task::JoinHandle>) { + let upstream = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = upstream.local_addr().unwrap(); + let handle = tokio::spawn(async move { + let mut captured = Vec::new(); + for response in responses { + let (mut stream, _) = upstream.accept().await.unwrap(); + let mut buffer = Vec::new(); + let mut chunk = [0u8; 8192]; + let (header_end, content_length) = loop { + let read = stream.read(&mut chunk).await.unwrap(); + assert!(read > 0); + buffer.extend_from_slice(&chunk[..read]); + let Some(end) = buffer.windows(4).position(|w| w == b"\r\n\r\n") else { + continue; + }; + let headers = String::from_utf8_lossy(&buffer[..end]).to_lowercase(); + let content_length = headers + .lines() + .find_map(|line| { + line.split_once(':') + .filter(|(name, _)| name.trim() == "content-length") + .and_then(|(_, value)| value.trim().parse::().ok()) + }) + .unwrap_or(0); + break (end + 4, content_length); + }; + while buffer.len() < header_end + content_length { + let read = stream.read(&mut chunk).await.unwrap(); + assert!(read > 0); + buffer.extend_from_slice(&chunk[..read]); + } + let body = String::from_utf8_lossy(&buffer[header_end..header_end + content_length]) + .to_string(); + captured.push(serde_json::from_str(&body).expect("发往上游的应是合法 JSON")); + stream.write_all(&response).await.unwrap(); + } + captured + }); + (addr, handle) +} + +async fn spawn_proxy( + upstream_addr: std::net::SocketAddr, + log_path: &std::path::Path, +) -> (u16, ChildGuard) { + let proxy_port = reserve_port(); + let config_path = log_path.with_extension("toml"); + std::fs::write( + &config_path, + format!( + r#" +[server] +port = {proxy_port} + +[upstream] +base_url = "http://{upstream_addr}/v1" +api_key = "sk-e2e-test" +protocol = "responses" + +[stabilizer] +prompt_prefix = true + +[log] +path = "{}" +"#, + log_path.display() + ), + ) + .unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_codex-layer")) + .env("CODEX_LAYER_CONFIG", &config_path) + .env("CODEX_LAYER_HOME", log_path.parent().unwrap()) + .env( + "CODEX_LAYER_TOOLS_CACHE_PATH", + log_path.with_extension("tools-cache.json"), + ) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let guard = ChildGuard(child); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + match TcpStream::connect(("127.0.0.1", proxy_port)).await { + Ok(_) => break, + Err(_) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(100)).await + } + Err(error) => panic!("codex-layer 未在 5 秒内就绪: {error}"), + } + } + (proxy_port, guard) +} + +type Harness = ( + u16, + ChildGuard, + tempfile::TempDir, + std::path::PathBuf, + tokio::task::JoinHandle>, +); + +async fn harness(responses: Vec>) -> Harness { + let (upstream_addr, capture) = mock_upstream(responses).await; + let dir = tempfile::tempdir().unwrap(); + let log_path = dir.path().join("codex-layer.log"); + std::fs::write(&log_path, "").unwrap(); + let (port, guard) = spawn_proxy(upstream_addr, &log_path).await; + (port, guard, dir, log_path, capture) +} + +async fn post(port: u16, body: &Value) -> String { + let payload = body.to_string(); + let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + stream + .write_all( + format!( + "POST /v1/responses HTTP/1.1\r\nhost: x\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + payload.len(), + payload + ) + .as_bytes(), + ) + .await + .unwrap(); + let mut proxied = String::new(); + stream.read_to_string(&mut proxied).await.unwrap(); + proxied +} + +fn normal_request(key: &str) -> Value { + json!({ + "model": "m", + "instructions": "core instructions", + "prompt_cache_key": key, + "stream": true, + "tools": [{"type": "function", "name": "exec_command", "parameters": {}}], + "input": [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "继续盯住这个任务"}]}, + ], + }) +} + +/// 静默轮：只有 reasoning 条目的 completed（本机 MLX 的病灶形态）。 +fn silent_stream(id: &str, item_id: &str, reasoning_tokens: u64) -> Vec { + sse_response(&format!( + concat!( + "event: response.created\ndata: {{\"type\":\"response.created\",\"response\":{{\"id\":\"{id}\",\"status\":\"in_progress\"}}}}\n\n", + "event: response.output_item.done\ndata: {{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{{\"id\":\"{item_id}\",\"type\":\"reasoning\"}}}}\n\n", + "event: response.completed\ndata: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"{id}\",\"status\":\"completed\",\"output\":[{{\"id\":\"{item_id}\",\"type\":\"reasoning\"}}],\"usage\":{{\"output_tokens_details\":{{\"reasoning_tokens\":{rt}}},\"total_tokens\":9000}}}}}}\n\n", + ), + id = id, + item_id = item_id, + rt = reasoning_tokens + )) +} + +/// 交付轮：带可见正文的 completed。 +fn delivered_stream(id: &str, item_id: &str, text: &str) -> Vec { + sse_response(&format!( + concat!( + "event: response.created\ndata: {{\"type\":\"response.created\",\"response\":{{\"id\":\"{id}\",\"status\":\"in_progress\"}}}}\n\n", + "event: response.output_item.done\ndata: {{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{{\"id\":\"{item_id}\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{{\"type\":\"output_text\",\"text\":\"{text}\"}}]}}}}\n\n", + "event: response.completed\ndata: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"{id}\",\"status\":\"completed\",\"output\":[{{\"id\":\"{item_id}\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{{\"type\":\"output_text\",\"text\":\"{text}\"}}]}}]}}}}\n\n", + ), + id = id, + item_id = item_id, + text = text + )) +} + +/// 数子串出现次数。 +fn count_needle(haystack: &str, needle: &str) -> usize { + haystack.matches(needle).count() +} + +fn read_log_records(log_path: &std::path::Path) -> Vec { + let mut raw = String::new(); + std::fs::File::open(log_path) + .expect("诊断日志必须存在") + .read_to_string(&mut raw) + .unwrap(); + raw.lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .collect() +} + +#[tokio::test] +async fn silent_first_round_retries_once_and_client_gets_merged_single_stream() { + let (port, _guard, _dir, log_path, capture) = harness(vec![ + silent_stream("resp_native_1", "reason_1", 4200), + delivered_stream("resp_native_2", "msg_2", "第二轮正文"), + ]) + .await; + let key = format!("native-guard-{}", std::process::id()); + + let seen = post(port, &normal_request(&key)).await; + assert!(seen.contains("200 OK"), "静默轮经重发后应放行, got={seen}"); + // 单条身份事件：第二轮的 created 必须被护栏吞掉。 + assert_eq!( + count_needle(&seen, "event: response.created"), + 1, + "客户端只能看到一条 created, got={seen}" + ); + // 单个终帧，且必须是合并帧：首轮 reasoning 垫底、本轮正文接位、id 回填首轮。 + assert_eq!( + count_needle(&seen, "event: response.completed"), + 1, + "客户端只能看到一个 completed, got={seen}" + ); + assert!(seen.contains("第二轮正文"), "本轮正文必须出门, got={seen}"); + assert!( + seen.contains("reason_1"), + "首轮条目必须留在终帧里, got={seen}" + ); + assert!( + seen.contains("resp_native_1"), + "终帧 id 必须回填首轮身份, got={seen}" + ); + assert!( + !seen.contains("resp_native_2"), + "第二轮的响应身份不许出门, got={seen}" + ); + assert!( + seen.contains("\"output_index\":1"), + "本轮条目必须按位移接位, got={seen}" + ); + assert!( + !seen.contains("response.failed"), + "重发成功不得报失败, got={seen}" + ); + + // 上游实收：两轮同体重采样（第一次重试赌温度，不动请求体）。 + let sent = tokio::time::timeout(Duration::from_secs(5), capture) + .await + .expect("上游应在 5 秒内收完两次请求") + .unwrap(); + assert_eq!(sent.len(), 2, "静默一轮只允许重发一次, sent={sent:?}"); + assert_eq!( + serde_json::to_string(&sent[0]).unwrap(), + serde_json::to_string(&sent[1]).unwrap(), + "第一次重试必须同体（前缀全命中），不发推动文案" + ); + + // 诊断账：retry 事件带被丢轮的 reasoning token，收口事件带真实上游轮次。 + let records = read_log_records(&log_path); + let retry = records + .iter() + .find(|record| record["event"] == json!("codex_layer.empty_frame_retry")) + .expect("必须记 empty_frame_retry"); + assert_eq!(retry["detail"]["attempt"], json!(1)); + assert_eq!(retry["detail"]["nudged"], json!(false)); + assert_eq!( + retry["detail"]["droppedUsage"]["reasoning_tokens"], + json!(4200), + "白烧的 reasoning token 必须可见, detail={}", + retry["detail"] + ); + let completed_round = records + .iter() + .find(|record| { + record["event"] == json!("codex_layer.stream_completed") + && record["detail"]["attempts"].is_number() + }) + .expect("原生支路的 stream_completed 必须带 attempts"); + assert_eq!(completed_round["detail"]["attempts"], json!(2)); +} + +#[tokio::test] +async fn exhausted_silence_still_delivers_merged_completed_and_nudges_second_retry() { + let (port, _guard, _dir, log_path, capture) = harness(vec![ + silent_stream("resp_native_1", "reason_1", 100), + silent_stream("resp_native_2", "reason_2", 200), + silent_stream("resp_native_3", "reason_3", 300), + ]) + .await; + let key = format!("native-exhausted-{}", std::process::id()); + + let seen = post(port, &normal_request(&key)).await; + assert!(seen.contains("200 OK"), "额度用尽也应放行, got={seen}"); + assert_eq!(count_needle(&seen, "event: response.created"), 1); + assert_eq!( + count_needle(&seen, "event: response.completed"), + 1, + "红线：绝不静默断流，最后一帧必须是合并 completed, got={seen}" + ); + assert!( + !seen.contains("response.failed"), + "静默不是失败，不得补失败帧, got={seen}" + ); + // 合并帧把三轮 reasoning 条目全部垫齐，与已流式下发的条目对得上账。 + for item_id in ["reason_1", "reason_2", "reason_3"] { + assert!(seen.contains(item_id), "终帧必须包含 {item_id}, got={seen}"); + } + assert!( + seen.contains("resp_native_1"), + "id 必须回填首轮, got={seen}" + ); + + let sent = tokio::time::timeout(Duration::from_secs(5), capture) + .await + .expect("上游应在 5 秒内收完三次请求") + .unwrap(); + assert_eq!( + sent.len(), + 3, + "额度 2 次重发 + 首轮 = 三次上游, sent={sent:?}" + ); + // 第一次同体，第二次带推动文案；推动项只进重发体。 + let body_of = |req: &Value| req["input"].as_array().unwrap().clone(); + assert_eq!(body_of(&sent[0]), body_of(&sent[1]), "第一次重试必须同体"); + let third_items = body_of(&sent[2]); + assert_eq!( + third_items.len(), + 2, + "第二次重试的 input 必须多一条推动项, items={third_items:?}" + ); + let nudge_text = third_items[1]["content"][0]["text"].as_str().unwrap(); + assert!( + nudge_text.contains("[relay-empty-frame-nudge]"), + "推动项必须带识别标记, text={nudge_text}" + ); + assert!( + !nudge_text.contains(""), + "推动项不得携带轮次标记, text={nudge_text}" + ); + + let records = read_log_records(&log_path); + let exhausted = records + .iter() + .find(|record| record["event"] == json!("codex_layer.empty_frame_exhausted")) + .expect("额度用尽必须记 empty_frame_exhausted"); + assert_eq!(exhausted["detail"]["attempts"], json!(2)); + assert_eq!( + exhausted["detail"]["droppedUsage"]["reasoning_tokens"], + json!(600), + "两轮被丢的 reasoning 账要累加, detail={}", + exhausted["detail"] + ); +} diff --git a/tests/responses_dialect_stabilizer.rs b/tests/responses_dialect_stabilizer.rs index e66e143..74ff932 100644 --- a/tests/responses_dialect_stabilizer.rs +++ b/tests/responses_dialect_stabilizer.rs @@ -36,10 +36,12 @@ async fn mock_responses_upstream( let upstream = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = upstream.local_addr().unwrap(); let handle = tokio::spawn(async move { + // 终帧必须带可见正文条目：空 output 的 completed 正是空终帧护栏的 + // 触发形态（issue #83），本测试测的是稳定层可见性，不该顺手点燃护栏。 let sse = concat!( "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n", "event: response.created\n\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"r1\",\"status\":\"in_progress\"}}\n\n", - "event: response.completed\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"r1\",\"status\":\"completed\"}}\n\n", + "event: response.completed\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"r1\",\"status\":\"completed\",\"output\":[{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"ok\"}]}]}}\n\n", ); let mut captured = Vec::new(); for _ in 0..n { diff --git a/tests/tools_cache_persistence.rs b/tests/tools_cache_persistence.rs index 05a4631..5f0723e 100644 --- a/tests/tools_cache_persistence.rs +++ b/tests/tools_cache_persistence.rs @@ -38,10 +38,12 @@ async fn mock_responses_upstream(n: usize) -> (SocketAddr, tokio::task::JoinHand let upstream = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = upstream.local_addr().unwrap(); let handle = tokio::spawn(async move { + // 终帧必须带可见正文条目：空 output 的 completed 正是空终帧护栏的 + // 触发形态（issue #83），本测试测 tools 回填，不该顺手点燃护栏。 let sse = concat!( "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n", "event: response.created\n\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"r1\",\"status\":\"in_progress\"}}\n\n", - "event: response.completed\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"r1\",\"status\":\"completed\"}}\n\n", + "event: response.completed\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"r1\",\"status\":\"completed\",\"output\":[{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"ok\"}]}]}}\n\n", ); let mut captured = Vec::new(); for _ in 0..n {