修复: 协议层与体验缺陷六件

- temperature 按模型适配,仅收 1 的模型不再 400
- 流式 error 帧不再被 choices 必填整帧丢弃
- 只读缓存写后失效
- git diff 中文截断撞 UTF-8 边界 panic
- 非视觉模型纯图片直接拒绝
- 全局快捷键补输入焦点守卫
This commit is contained in:
lxy
2026-10-04 16:27:31 +08:00
parent 01d231c39d
commit 7aa0da32f2
28 changed files with 491 additions and 52 deletions
+12
View File
@@ -138,6 +138,14 @@ pub struct ModelConfig {
#[serde(default = "default_context_window")]
pub context_window: usize,
/// 是否支持 temperature 参数(默认 true)。
///
/// 部分模型仅接受固定温度值(如商汤 sensenova 系列仅允许 1.0),透传其它温度(标题 0.3/
/// 压缩 0.2 等)→ 上游 HTTP 400,实测 9 次。false 时协议转换层省略 temperature 字段
/// (交由上游默认),避免无效往返;true 保持原样透传(零回归)。
#[serde(default = "default_true")]
pub temperature_supported: bool,
// ── 探测元数据(只读,由 ModelProbe 填充,Phase 2)──
#[serde(default, skip_serializing_if = "Option::is_none")]
pub probe_source: Option<ProbeSource>,
@@ -173,6 +181,9 @@ impl ModelConfig {
intelligence: IntelligenceTier::Standard,
weight: default_weight(),
context_window: default_context_window(),
// 默认支持 temperature(老格式升级/未知模型保持原样透传,零回归);
// 明确不支持的模型由 presets 表 temperature_supported=false 标注。
temperature_supported: true,
probe_source: None,
}
}
@@ -339,6 +350,7 @@ mod tests {
intelligence: IntelligenceTier::Plus,
weight: 70,
context_window: 128_000,
temperature_supported: true,
probe_source: Some(ProbeSource::PresetTable),
};
let json = serde_json::to_string(&mc).unwrap();
+1
View File
@@ -472,6 +472,7 @@ mod tests {
usage: None,
error: None,
reasoning_content: Some("thought".to_string()),
truncated_by_max_tokens: false,
};
let json = serde_json::to_string(&chunk).unwrap();
assert!(json.contains("reasoning_content"));
+5
View File
@@ -350,6 +350,11 @@ pub struct StreamChunk {
/// 非空表示流中途出错,不应视为正常完成(finished 路径),由 stream_llm 转 AiError。
#[serde(skip)]
pub error: Option<String>,
/// 本流因 provider 输出上限截断收尾(openai finish_reason=length /
/// anthropic stop_reason=max_tokens)。finished 仍为 true(正常终止非断连),但文本/
/// tool_calls 可能不完整;消费方跨帧 OR 累积该标志,轮产出据此标记截断,不冒充正常完成。
#[serde(default)]
pub truncated_by_max_tokens: bool,
/// DeepSeek thinking 模式推理内容增量
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
+2 -1
View File
@@ -133,7 +133,8 @@
"modalities": ["text", "vision"],
"capabilities": ["tool_use"],
"weight": 85,
"context_window": 262144
"context_window": 262144,
"temperature_supported": false
},
{
"model_id": "deepseek-v4-flash",
+27 -1
View File
@@ -262,12 +262,23 @@ impl AnthropicCompatProvider {
.collect()
});
// 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。
// 判定走 model_probe(命中 presets 表;未命中默认支持,保持原样透传零回归)。
let temperature = if req.temperature.is_some()
&& !crate::model_probe::probe(&model).temperature_supported
{
tracing::debug!(model = %model, "模型不支持 temperature,已省略该参数");
None
} else {
req.temperature
};
AnthropicRequest {
model,
messages,
max_tokens: req.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS),
system,
temperature: req.temperature,
temperature,
stream: req.stream,
tools,
tool_choice: req.tool_choice,
@@ -927,6 +938,21 @@ mod tests {
assert_eq!(a.total_tokens, 15);
}
/// stop_reason=max_tokens(message_delta 帧携带)必须置截断标志;end_turn 不置。
/// 终态 finished 在后续 message_stop 帧,消费方跨帧 OR 累积标志。
#[test]
fn anthropic_max_tokens_stop_reason_sets_truncation_flag() {
let mut acc: Option<TokenUsage> = None;
let data = r#"{"type":"message_delta","delta":{"stop_reason":"max_tokens"},"usage":{"output_tokens":9}}"#;
let c = apply_anthropic_event(data, &mut acc);
assert!(!c.finished, "message_delta 非终态帧");
assert!(c.truncated_by_max_tokens, "max_tokens 必须置截断标志");
let mut acc2: Option<TokenUsage> = None;
let c2 = apply_anthropic_event(&message_delta(5), &mut acc2);
assert!(!c2.truncated_by_max_tokens, "end_turn 不置截断标志");
}
/// message_delta 的 output_tokens 必须是累计覆盖(非累加):连续两个 delta 5 和 8,结果应是 8 不是 13
#[test]
fn anthropic_message_delta_output_is_cumulative_not_incremental() {
+16 -10
View File
@@ -137,7 +137,7 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
let v: serde_json::Value = match serde_json::from_str(data) {
Ok(v) => v,
Err(_) => {
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false }
}
};
let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
@@ -169,7 +169,7 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
reasoning_tokens: 0,
});
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: usage_accum.clone(), error: None, reasoning_content: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: usage_accum.clone(), error: None, reasoning_content: None, truncated_by_max_tokens: false }
}
// 消息增量:output_tokens 是累计值(非增量),直接覆盖 completion + 重算 total
"message_delta" => {
@@ -179,9 +179,12 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
acc.completion_tokens = out as u32;
acc.total_tokens = acc.prompt_tokens.saturating_add(acc.completion_tokens);
}
// stop_reason=max_tokens(message_delta 帧携带):输出被 max_tokens 截断。
// 终态 finished 在后续 message_stop 帧,截断标志挂本帧,消费方跨帧 OR 累积。
let truncated_by_max_tokens = v.get("delta").and_then(|d| d.get("stop_reason")).and_then(|s| s.as_str()) == Some("max_tokens");
// 加固:usage 挂到本帧(而非仅 message_stop 带出),中途断连/端点不发 message_stop
// 时仍能拿到真实 usage,对称 openai_helpers 的修复。
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: usage_accum.clone(), error: None, reasoning_content: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: usage_accum.clone(), error: None, reasoning_content: None, truncated_by_max_tokens }
}
// 文本增量
"content_block_delta" => {
@@ -190,16 +193,16 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
// 不混入正文 delta。本侧请求未启用 thinking,此分支只为不丢端点侧自行产出的思考。
if delta.get("type").and_then(|t| t.as_str()) == Some("thinking_delta") {
let text = delta.get("thinking").and_then(|t| t.as_str()).unwrap_or("").to_string();
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: Some(text) };
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: Some(text), truncated_by_max_tokens: false };
}
// signature_delta:thinking 块签名字段,仅回传校验用。DevFlow 从不把 thinking 回传
// anthropic 历史(丢弃 signature 后伪造回传必 400),签名无处消费,安静忽略。
if delta.get("type").and_then(|t| t.as_str()) == Some("signature_delta") {
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None };
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false };
}
if delta.get("type").and_then(|t| t.as_str()) == Some("text_delta") {
let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("").to_string();
return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None };
return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false };
}
// 工具入参增量
if delta.get("type").and_then(|t| t.as_str()) == Some("input_json_delta") {
@@ -217,10 +220,11 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
usage: None,
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
};
}
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false }
}
// 工具块开始:带 id + name
"content_block_start" => {
@@ -253,10 +257,11 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
usage: None,
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
};
}
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false }
}
// 消息结束:带出累积 usage
"message_stop" => StreamChunk {
@@ -266,15 +271,16 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
usage: usage_accum.take(),
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
},
// 错误事件:流中途出错。不走 finished 完成路径(避免残缺响应被当正常完成入库),
// 改由 stream_llm 识别 error 非空 → 发 AiError + 丢弃残缺(与 OpenAI 路径 Err 一致)。
"error" => {
let msg = v.get("error").and_then(|e| e.get("message")).and_then(|m| m.as_str()).unwrap_or("stream error").to_string();
error!(%msg, "Anthropic 流式错误事件");
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None, truncated_by_max_tokens: false }
}
// content_block_stop / ping 等不产出 chunk
_ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None },
_ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None, truncated_by_max_tokens: false },
}
}
+23 -8
View File
@@ -841,12 +841,12 @@ fn phase1_push_only_keeps_clean_no_dirty_flag() {
#[test]
fn phase1_advance_persisted_count_sets_baseline_and_clears_dirty() {
// advance_persisted_count(new_count):推进基线 + 清 needs_full_rewrite(DB 已与内存一致)
// advance_persisted_count(new_count, full):推进基线;全量重写才清 needs_full_rewrite
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("x"));
mgr.push(ChatMessage::user("y"));
// 模拟 save 成功后推进
mgr.advance_persisted_count(2);
// 模拟 save(全量重写)成功后推进
mgr.advance_persisted_count(2, true);
assert_eq!(mgr.persisted_msg_count(), 2);
assert!(!mgr.needs_full_rewrite());
}
@@ -855,11 +855,26 @@ fn phase1_advance_persisted_count_sets_baseline_and_clears_dirty() {
fn phase1_advance_persisted_count_max_no_regression() {
// 防回退:advance 传更小值不回退基线(防误传 / 并发致基线丢失 → 下次增量漏写)
let mut mgr = ContextManager::new(cfg(100_000));
mgr.advance_persisted_count(5);
mgr.advance_persisted_count(3); // 误传更小
mgr.advance_persisted_count(5, true);
mgr.advance_persisted_count(3, true); // 误传更小
assert_eq!(mgr.persisted_msg_count(), 5, "基线不可回退");
}
#[test]
fn phase1_advance_incremental_keeps_dirty() {
// 增量/无变化路径(full=false)不得清 dirty:写库期间新置的 dirty(如压缩链 mark)
// 与本次 append 写库无关,误清会致改动滞留内存永不落库
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("a"));
mgr.advance_persisted_count(1, false); // 增量 save 成功
assert!(!mgr.needs_full_rewrite());
mgr.mark_needs_full_rewrite(); // 模拟写库期间压缩 mark
mgr.advance_persisted_count(1, false); // 又一次增量 save
assert!(mgr.needs_full_rewrite(), "增量路径 advance 不得清 dirty");
mgr.advance_persisted_count(1, true); // 直到全量重写才清
assert!(!mgr.needs_full_rewrite());
}
#[test]
fn phase1_compress_old_messages_sets_dirty() {
// compress 改既有消息 status(active→compressed) → 必须全量重写收敛
@@ -881,7 +896,7 @@ fn phase1_compress_no_op_does_not_set_dirty() {
mgr.push(ChatMessage::user("a"));
let _ = mgr.compress_old_messages(1);
assert!(mgr.needs_full_rewrite(), "首次压缩应置 dirty");
mgr.advance_persisted_count(1); // 模拟 save 后清 dirty
mgr.advance_persisted_count(1, true); // 模拟 save(全量)后清 dirty
// 二次压缩同范围:全已 compressed,无新标 → 不应再置 dirty
let compressed = mgr.compress_old_messages(1);
assert!(compressed.is_empty());
@@ -974,7 +989,7 @@ fn phase1_clear_resets_baseline_and_sets_dirty() {
// clear:基线重置 0 + 置 dirty(下次 save 全量重写空列表清 DB)
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("a"));
mgr.advance_persisted_count(1);
mgr.advance_persisted_count(1, true);
assert_eq!(mgr.persisted_msg_count(), 1);
mgr.clear();
assert_eq!(mgr.persisted_msg_count(), 0, "clear 应重置基线到 0");
@@ -1021,7 +1036,7 @@ fn phase1_append_after_dirty_stays_dirty_until_save() {
mgr.push(ChatMessage::user("b")); // 追加
assert!(mgr.needs_full_rewrite(), "dirty 后 push 不应清 dirty");
// 模拟 save(全量重写)成功后推进 + 清
mgr.advance_persisted_count(2);
mgr.advance_persisted_count(2, true);
assert!(!mgr.needs_full_rewrite());
}
+7 -3
View File
@@ -528,11 +528,15 @@ impl ContextManager {
}
/// msg-split-phase1:推进已落库消息条数(insert_batch/replace_conversation 成功后调用)。
/// 同时清 needs_full_rewrite(DB 已与内存一致)。供 save_conversation 在写库成功后调用。
pub fn advance_persisted_count(&mut self, new_count: usize) {
/// `full_rewrite_applied`=本次写库是否走了全量重写(快照时刻 needs_full_rewrite 快照)。
/// 仅全量重写覆盖了既有行才清 needs_full_rewrite;增量/无变化路径写库期间新置的 dirty
/// (如压缩链 mark)与本次 append 写库无关,误清会致改动滞留内存永不落库。供 save_conversation 写库成功后调用。
pub fn advance_persisted_count(&mut self, new_count: usize, full_rewrite_applied: bool) {
// 防回退:若调用方误传更小值(理论上不应发生),取 max 保不丢基线(下次 save 仍能收敛)。
self.persisted_msg_count = self.persisted_msg_count.max(new_count);
self.needs_full_rewrite = false;
if full_rewrite_applied {
self.needs_full_rewrite = false;
}
}
/// msg-split-phase1:是否需要全量重写(改过既有消息 / 缩短 / 中段插入 / 清空)。
+18
View File
@@ -159,6 +159,24 @@ mod tests {
assert!(m.modalities.contains(&Modality::Text));
// context_window 应为预设的 256K
assert_eq!(m.context_window, 262144);
// P1 温度:商汤 sensenova 仅接受 temperature=1.0,透传其它值(标题 0.3/压缩 0.2)
// → 上游 HTTP 400(实测 9 次)。预设表须标 temperature_supported=false,
// 协议转换层据此省略该字段。
assert!(
!m.temperature_supported,
"sensenova 应标 temperature_supported=false(仅接受 1.0)"
);
}
/// 温度支持判定:仅 sensenova 系列为 false,其余模型默认 true(零回归)。
#[test]
fn probe_temperature_supported_defaults_true_except_sensenova() {
assert!(!probe("sensenova-6.7-flash-lite").temperature_supported);
// 启发式路径(非预设表):sensenova 前缀仍判 false
assert!(!probe("sensenova-unknown-model").temperature_supported);
// 常规模型保持 true(原样透传)
assert!(probe("deepseek-v4-flash").temperature_supported);
assert!(probe("glm-4v").temperature_supported);
}
#[test]
+10
View File
@@ -74,6 +74,7 @@ pub(super) fn heuristic_infer(model_id: &str) -> ModelConfig {
intelligence,
weight: 50,
context_window: 8192,
temperature_supported: temperature_supported_by_name(&name),
probe_source: None,
};
}
@@ -106,6 +107,7 @@ pub(super) fn heuristic_infer(model_id: &str) -> ModelConfig {
intelligence,
weight: 50,
context_window: 8192,
temperature_supported: temperature_supported_by_name(&name),
probe_source: None,
}
}
@@ -120,3 +122,11 @@ pub(super) fn has_vision_token(name: &str) -> bool {
|| name.contains("vl")
|| name.split(['-', '.', '_']).any(|seg| seg.ends_with('v'))
}
/// 温度参数支持判定(命名约定式):商汤 sensenova 系列仅接受 temperature=1.0,
/// 透传其它温度值(标题 0.3/压缩 0.2 等)→ 上游 HTTP 400(实测 9 次)。
/// 返回 false 时协议转换层省略 temperature 字段(交上游默认),避免无效往返;
/// 其余模型默认 true(原样透传,零回归)。
pub(super) fn temperature_supported_by_name(name: &str) -> bool {
!name.starts_with("sensenova")
}
+12 -1
View File
@@ -209,10 +209,21 @@ impl OpenAICompatProvider {
.collect()
});
// 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。
// 判定走 model_probe(命中 presets 表;未命中默认支持,保持原样透传零回归)。
let temperature = if req.temperature.is_some()
&& !crate::model_probe::probe(&model).temperature_supported
{
tracing::debug!(model = %model, "模型不支持 temperature,已省略该参数");
None
} else {
req.temperature
};
OpenAiRequest {
model,
messages,
temperature: req.temperature,
temperature,
max_tokens: req.max_tokens,
stream: req.stream,
tools,
+36
View File
@@ -274,6 +274,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
usage: usage_accum.take(),
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
};
}
@@ -296,6 +297,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
usage: None,
error: Some(msg),
reasoning_content: None,
truncated_by_max_tokens: false,
};
}
// 提取 usage(带 include_usage 时末段 chunk 携带,覆盖累积)
@@ -334,6 +336,9 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
choice.finish_reason.as_deref(),
Some("stop") | Some("tool_calls") | Some("length") | Some("content_filter")
);
// length = 输出被 max_tokens 截断:仍按正常终止收敛(finished=true),
// 但截断标志随 chunk 带出,消费方据此把该轮产出标记截断,不冒充完整。
let truncated_by_max_tokens = choice.finish_reason.as_deref() == Some("length");
let tool_calls = choice.delta.tool_calls.map(|tcs| {
// 缺 index 的增量缺省 0(单工具调用语义等价);但同帧多 call 且 index
@@ -379,6 +384,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
usage: usage_accum.clone(),
error: None,
reasoning_content: choice.delta.reasoning_content,
truncated_by_max_tokens,
}
} else {
// choices 为空 = usage-only chunk,不输出文本(usage 已累积),usage 一并带出
@@ -389,6 +395,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
usage: usage_accum.clone(),
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
}
}
}
@@ -401,6 +408,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
usage: None,
error: None,
reasoning_content: None,
truncated_by_max_tokens: false,
}
}
}
@@ -503,4 +511,32 @@ mod tests {
);
assert!(!chunk2.finished, "无终态 finish_reason 的普通 delta 不应终结");
}
/// finish_reason=length:max_tokens 截断仍按正常终止收敛(finished=true),
/// 但截断标志必须随 chunk 带出(零信号会让腰斩报告冒充完整完成);
/// stop 等正常终态不置标志。
#[test]
fn finish_reason_length_carries_truncation_flag() {
let mut accum = None;
let chunk = apply_openai_sse(
r#"{"choices":[{"delta":{"content":"报告写到一半"},"finish_reason":"length"}]}"#,
&mut accum,
);
assert!(chunk.finished, "length 属正常终止");
assert!(chunk.truncated_by_max_tokens, "length 必须带截断标志");
let mut accum2 = None;
let chunk2 = apply_openai_sse(
r#"{"choices":[{"delta":{"content":"ok"},"finish_reason":"stop"}]}"#,
&mut accum2,
);
assert!(chunk2.finished && !chunk2.truncated_by_max_tokens, "stop 不置截断标志");
let mut accum3 = None;
let chunk3 = apply_openai_sse(
r#"{"choices":[{"delta":{"content":"x"},"finish_reason":null}]}"#,
&mut accum3,
);
assert!(!chunk3.truncated_by_max_tokens, "普通 delta 不置截断标志");
}
}
+95 -7
View File
@@ -5,7 +5,7 @@
//! 但重试能在不侵入上层的前提下吸收瞬态错误,对低频后台调用零风险。
//!
//! 策略:
//! - 最多 3 次尝试(初次 + 2 次重试),指数退避 1s → 2s → 4s + ±20% jitter
//! - 最多 3 次尝试(初次 + 2 次重试),指数退避 1s → 2s → 4s + ±50% jitter
//! - 总挂钟预算 30s(含 sleep + 各次请求耗时),超预算直接放弃重试返回最后错误
//! - 仅对可重试错误重试: timeout / connect / 5xx / 429; 不重试 4xx(非429) / 参数错 / 鉴权错
//! - 不引入 reqwest-retry / tower: complete 路径简单,手写 loop 零依赖、可调试、不与 SSE/双层 Semaphore 耦合
@@ -34,6 +34,11 @@ const BASE_BACKOFF_SECS: u64 = 1;
/// 与单请求 60s timeout 配合: 最坏 3 次 × 60s = 180s,30s 预算会在此之前主动止损。
const MAX_TOTAL_BUDGET: Duration = Duration::from_secs(30);
/// 单次退避基准上限(毫秒),与 [`MAX_TOTAL_BUDGET`] 同口径: 超预算的退避没有意义,
/// attempt 走深时不再指数膨胀。与总预算的一致性由测试 `backoff_cap_matches_budget` 守住。
/// (同构先例: df-tunnel 的 backoff_delay 也以 RECONNECT_MAX 封顶。)
const MAX_BACKOFF_MS: u64 = 30_000;
/// jitter 上限(相对 base 的 ±比例)。避免重试风暴对齐。
/// CR-XX: 范围扩到 ±50%(由 gen_range(-0.5..0.5) × JITTER_RATIO=1.0 合成)。
const JITTER_RATIO: f64 = 1.0;
@@ -65,20 +70,45 @@ pub fn is_status_retryable(status: u16) -> bool {
status == 429 || (500..600).contains(&status)
}
/// 退避倍率 `2^(attempt-1)`,位移溢出安全。
///
/// 替代裸 `1u64 << (attempt - 1)`: attempt > 64 时位移越界(debug panic / release 环绕)、
/// attempt = 0 时下溢 panic。现 attempt = 0 视作 1(倍率 1),attempt - 1 ≥ 64 时
/// checked_shl 返回 None,倍率封顶 u64::MAX,最终时长由 [`MAX_BACKOFF_MS`] 兜底。
/// 全输入无 panic,序列单调不减。
fn backoff_multiplier(attempt: u32) -> u64 {
match 1u64.checked_shl(attempt.saturating_sub(1)) {
Some(m) => m,
None => u64::MAX,
}
}
/// 退避基准毫秒(未加 jitter),`BASE_BACKOFF_SECS * 2^(attempt-1) * 1000`,封顶 [`MAX_BACKOFF_MS`]。
///
/// 乘法全程 saturating: 倍率封顶 u64::MAX 后 `* 1000` 也会二次溢出,不能裸乘。
/// 单一事实源——backoff_delay 与单测的 base 断言都走这里,不再各写一份公式。
fn base_backoff_ms(attempt: u32) -> u64 {
BASE_BACKOFF_SECS
.saturating_mul(backoff_multiplier(attempt))
.saturating_mul(1000)
.min(MAX_BACKOFF_MS)
}
/// 指数退避 + jitter: 返回第 `attempt`(1-based)次重试前应 sleep 的时长。
/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s,各 ±50% jitter。
///
/// jitter 用 `rand::thread_rng().gen_range(-0.5..0.5)` 生成 ±50% 比例,避免多客户端同步重试风暴。
/// 以毫秒粒度计算后向下取整(避免秒级截断把 0.9s 砍成 0)。
/// 全输入安全: attempt 任意 u32(含 0 / >64)不 panic,结果 ≤ [`MAX_BACKOFF_MS`]。
///
/// 暴露 pub 供 src-tauri/agentic.rs 流前重试复用
/// ("复用 retry.rs backoff_delay 退避 1s→2s→4s+jitter"),避免重写退避逻辑。
pub fn backoff_delay(attempt: u32) -> Duration {
let base_ms = BASE_BACKOFF_SECS.saturating_mul(1u64 << (attempt - 1)) * 1000;
let base_ms = base_backoff_ms(attempt) as f64;
// ±50% jitter,相对 base 时长的浮动比例
let jitter_ratio = rand::thread_rng().gen_range(-0.5..0.5); // [-0.5, 0.5)
let factor = 1.0 + jitter_ratio * JITTER_RATIO;
let ms = (base_ms as f64 * factor).max(0.0) as u64;
let ms = (base_ms * factor).clamp(0.0, MAX_BACKOFF_MS as f64) as u64;
Duration::from_millis(ms)
}
@@ -160,12 +190,12 @@ mod tests {
#[test]
fn backoff_is_monotonic_within_jitter() {
// base 序列: 1, 2, 4 秒; jitter ±20% 后仍在 [0.8,4.8] 范围
// base 序列: 1, 2, 4 秒; jitter ±50%(JITTER_RATIO=1.0) 后仍在 [0.5x, 1.5x] 范围
for attempt in 1..=3 {
let d = backoff_delay(attempt);
let base_ms = (BASE_BACKOFF_SECS * (1u64 << (attempt - 1))) * 1000;
let lo = (base_ms as f64 * (1.0 - JITTER_RATIO)) as u64;
let hi = (base_ms as f64 * (1.0 + JITTER_RATIO)) as u64;
let base_ms = base_backoff_ms(attempt);
let lo = (base_ms as f64 * (1.0 - JITTER_RATIO / 2.0)) as u64;
let hi = (base_ms as f64 * (1.0 + JITTER_RATIO / 2.0)) as u64;
assert!(
d.as_millis() as u64 >= lo && d.as_millis() as u64 <= hi,
"attempt={} delay={:?} 应在 [{},{}]ms",
@@ -177,6 +207,64 @@ mod tests {
}
}
/// 倍率溢出安全: attempt=0 不下溢,≥65 封顶 u64::MAX,序列单调不减
#[test]
fn backoff_multiplier_extremes_safe() {
assert_eq!(backoff_multiplier(0), 1, "attempt=0 视作 1,不下溢 panic");
assert_eq!(backoff_multiplier(1), 1);
assert_eq!(backoff_multiplier(2), 2);
assert_eq!(backoff_multiplier(3), 4);
assert_eq!(backoff_multiplier(63), 1u64 << 62);
assert_eq!(backoff_multiplier(64), 1u64 << 63, "u64 可表达的最大位移");
assert_eq!(backoff_multiplier(65), u64::MAX, "checked_shl 越界封顶");
assert_eq!(backoff_multiplier(100), u64::MAX);
assert_eq!(backoff_multiplier(u32::MAX), u64::MAX);
let seq: Vec<u64> = [0u32, 1, 2, 3, 10, 63, 64, 65, 100, u32::MAX]
.iter()
.map(|&a| backoff_multiplier(a))
.collect();
assert!(seq.windows(2).all(|w| w[0] <= w[1]), "倍率应单调不减: {:?}", seq);
}
/// 基准毫秒单调不减、封顶值正确(2^5×1s=32s 已超 30s 预算,从 attempt=6 起封顶)
#[test]
fn base_backoff_ms_monotonic_and_capped() {
let mut prev = 0u64;
for attempt in [0u32, 1, 2, 3, 4, 5, 6, 10, 63, 64, 65, 100, u32::MAX] {
let b = base_backoff_ms(attempt);
assert!(b >= prev, "attempt={} base_ms={} 应 ≥ 前值 {}", attempt, b, prev);
assert!(b <= MAX_BACKOFF_MS, "attempt={} base_ms={} 应 ≤ {}", attempt, b, MAX_BACKOFF_MS);
prev = b;
}
assert_eq!(base_backoff_ms(1), 1000);
assert_eq!(base_backoff_ms(2), 2000);
assert_eq!(base_backoff_ms(3), 4000);
assert_eq!(base_backoff_ms(5), 16_000, "未触及封顶");
assert_eq!(base_backoff_ms(6), MAX_BACKOFF_MS, "32s 超预算封顶 30s");
assert_eq!(base_backoff_ms(u32::MAX), MAX_BACKOFF_MS);
}
/// 极端 attempt 下 backoff_delay 不 panic(修复: 1u64<<(attempt-1) 曾 debug panic/release 环绕)
#[test]
fn backoff_delay_extreme_attempts_no_panic_and_capped() {
for attempt in [0u32, 1, 10, 63, 64, 65, 100, u32::MAX] {
let ms = backoff_delay(attempt).as_millis() as u64;
assert!(
ms <= MAX_BACKOFF_MS,
"attempt={} 实际 {}ms 应 ≤ {}ms(含 jitter 后仍不超)",
attempt,
ms,
MAX_BACKOFF_MS
);
}
}
/// 封顶常量与总预算同口径(改预算时两处同步)
#[test]
fn backoff_cap_matches_budget() {
assert_eq!(MAX_BACKOFF_MS, MAX_TOTAL_BUDGET.as_millis() as u64);
}
#[test]
fn status_classification() {
assert!(is_status_retryable(429));
+1
View File
@@ -163,6 +163,7 @@ mod tests {
intelligence: IntelligenceTier::Standard,
weight: 50,
context_window: 8192,
temperature_supported: true,
probe_source: None,
}
}