docs(phase11): 记录 Phase 11 方案文档与实施偏差

This commit is contained in:
徐涛
2026-07-06 14:53:00 +08:00
parent 71abe881ed
commit b4e5c7d651
+647
View File
@@ -0,0 +1,647 @@
# Phase 11: 测试与检索补强
## 背景与目标
AG Core 当前(v0.2.0-rc.1)已完成 Phase 0-10,全量测试 254 个,clippy 0 警告,11 个离线示例全部 exit 0。功能性交付物覆盖了 LLM Cycle、Prompt、Tool、Memory、Agent Runtime、流式事件、ContextSlot 上下文管理。但有两个系统性的短板尚未补齐:
1. **检索抽象缺失**`memory` 模块只有 `MemoryRetriever`(基于 TextOverlap Dice 系数的关键词检索),缺少语义向量的检索抽象。`docs/roadmap.md` P1 中「VectorRetriever trait」一直未实现。
2. **测试覆盖缺口**Provider 的 roundtrip 测试停留在"基本响应 + 普通 401/500"层面,缺少结构化错误体解析、请求头验证、流式边界、工具调用端到端等关键场景的回归覆盖。多线程并发的 MemoryStore 测试只在 SqliteStore 有一个 10×10 场景,InMemoryStore 完全没有并发压力测试。
Phase 11 是 v0.2.0 正式版发布前的最后一个功能 Phase,三个 Step 的目标:
| Step | 内容 | 定位 |
|------|------|------|
| **11.1** | `VectorRetriever` trait + `InMemoryVectorRetriever` 引用实现 | P1 功能补全 |
| **11.2** | wiremock Provider roundtrip 测试(12 个场景) | 测试质量补强 |
| **11.3** | 并发测试补强(InMemoryStore + SqliteStore | 并发安全验证 |
最终目标:全量测试从 254 → 275+,为 v0.2.0 正式版建立更高的质量基线。
## 需求推演概要
### Step 11.1 — VectorRetriever trait
**核心需求**:定义一个与后端无关的语义检索抽象接口,包含 `index(id, embeddings)` 索引和 `search(query, k)` 检索两个方法。附带一个基于 `HashMap` 全量余弦相似度扫描的参考实现。
**边界识别**
- 只定义 trait,不绑定任何具体后端(pgvector / qdrant / lancedb 留给社区或下游)
- 不引入第三方向量数据库依赖
- 引用实现的 `search()` 不做索引加速(O(n) 全量扫描已足够验证 trait 契约)
- 不与 `MemoryStore` 耦合——`VectorRetriever` 是独立维度
- 不嵌入到 `AgentSession``ContextSlot`(Phase 11 不承担集成消费端)
**关键假设**
- `Vec<f32>` 作为 embedding 类型已足够(大部分 embedding 模型输出 f32 向量)
- 余弦相似度作为默认评分函数可覆盖主流场景
- InMemoryVectorRetriever 的 `Mutex<HashMap>` 在 ~10K 向量内性能可接受
### Step 11.2 — wiremock Provider roundtrip 测试
**核心需求**:补充 12 个 wiremock 测试,覆盖目前缺失的关键回归场景——结构化 JSON 错误体解析、请求头验证、429 限流头解析、流式边界、工具调用端到端。
**边界识别**
- 只做 HTTP mock 层验证,不做端到端 LLM 模型调用
- 每个测试自包含(启动自己的 MockServer),不抽共享 helper
- 测试集中在 OpenAI`GenericOpenaiProvider`)和 Anthropic(独立实现)两个核心 Provider 上
- DeepSeek/Qwen/Ollama 同属 OpenAI Compat,继承 `GenericOpenaiProvider` 的测试覆盖
**关键假设**
- wiremock 的 `body_partial_json` matcher 可用且稳定(当前 dev-dependencies 中已有 wiremock
- OpenAI 和 Anthropic 的结构化错误体格式在当前 SDK 版本中未变化
### Step 11.3 — 并发测试补强
**核心需求**:验证 `MemoryStore` 两种实现(InMemoryStore + SqliteStore)在多线程并发写和混合读写场景下的正确性。
**边界识别**
- 不测试 `KnowledgeStore` / `ConversationMemory` 的并发——它们的行为完全由 `MemoryStore` 决定,不引入新 race 条件
- 不测试 TTL 淘汰的并发正确性(TTL 淘汰使用 wall clock,非原子,不保证精确)
- 混合读写测试只验证"无 panic + 数量正确",不验证"读到的结果恰好与写顺序一致"(后者需要强一致快照,当前 Mutex 模型不提供)
**关键假设**
- `tokio::spawn` 100 个 task 同时写入 `Mutex<HashMap>`InMemoryStore)不会死锁
- SqliteStore 的 WAL 模式 + `busy_timeout=5000` 足够容忍 100 并发写
## 当前状态分析
### 测试覆盖率现状
| 维度 | 当前值 | Phase 11 目标 |
|------|--------|-------------|
| 全量测试 | 254 passed | 275+ passed |
| InMemoryStore 测试 | 6 个(save/get/list/upsert/eviction/TTL | +4 个并发 |
| SqliteStore 测试 | 9 个(含 1 个 10×10 并发) | +1 个 100 并发 |
| OpenAI wiremock 测试 | 4 个(basic/401/500/stream | +8 个 |
| Anthropic wiremock 测试 | 4 个(basic/401/529/stream | +4 个 |
| 请求头验证测试 | 0 个 | +2 个 |
| ToolUse 端到端 mock 测试 | 0 个 | +2 个(OpenAI + Anthropic |
### Provider 测试缺口
现有 wiremock 测试仅覆盖最基础的响应路径,以下关键场景缺失回归保护:
| 场景 | 缺失风险 |
|------|---------|
| OpenAI 请求体格式验证 | `body_partial_json` 未匹配,请求体结构变化无声 |
| Authorization header 验证 | header 注入被修改时不告警 |
| 结构化 401 JSON 错误体 | `error.message`/`error.code` 未消费,错误消息丢失 |
| 429 + `retry-after` 头 | `RateLimit.retry_after` 字段不准确 |
| ToolUse 端到端 mock | tool_flow 解析路径无回归 |
| 流式 last chunk usage-only | `{choices:[], usage:{...}}` 可能 panic |
### MemoryStore 并发测试缺口
| Store | 当前并发测试 | 覆盖度 | 风险 |
|-------|-------------|--------|------|
| InMemoryStore | 0 个 | 无 | `Mutex` 锁竞争、deadlock、写入丢失 |
| SqliteStore | 1 个(10 写者 × 10 次 = 100 条) | 中等 | `spawn_blocking` 线程池耗尽、WAL 锁等待超时 |
### 向量检索现状
`memory` 模块已有 `MemoryRetriever`(关键词检索)和 `retriever.rs` 中的 `RetrievalResult`/`ScoredItem` 类型。但语义向量检索维度完全空缺——无 trait、无引用实现、无测试。`docs/roadmap.md` 将 VectorRetriever 列为 P1,与 ContextSlotP1Phase 10)同级。
## 架构决策记录
| 决策 | 选择 | 放弃 | 理由 |
|------|------|------|------|
| 1. VectorRetriever trait 参数类型 | `Vec<f32>` 裸向量 | `Embedding` newtype | 包装类型增加可见复杂度但未提供运行时保护;大部分 embedding 模型输出 f32 向量;下游可自行包装 |
| 2. `search()` 返回类型 | `Vec<(String, f32)>` | `ScoredItem`/`RetrievalResult` 命名 struct | `(String, f32)` 是 (id, score) 的最小表达;Phase 3 的 `RetrievalResult` 绑定了 `KnowledgePage` 引用,不适合向量检索场景;tuple 在 consumer 侧模式匹配更简洁 |
| 3. 文件归属 | 新文件 `memory/vector.rs` | 合入 `memory/retriever.rs` | `retriever.rs` 已承载 302 行关键词检索代码,语义维度独立不应耦合;`vector.rs` 作为独立模块便于后期扩展(pgvector adapter 等) |
| 4. 是否附带引用实现 | `InMemoryVectorRetriever` | trait-only | trait-only 是纯推测代码,无 consumer 验证引用实现作为"编译期测试"验证 trait 方法签名可用 |
| 5. InMemoryVectorRetriever 余弦相似度实现方式 | 手动三行点积/范数 | `ndarray`/`approx` 等第三方依赖 | 余弦相似度数学固定,不需要外部依赖;1e-10 防零除;零新依赖原则 |
| 5a | 引用实现不做向量维度校验 | 运行时维度检查 | 维度校验是具体后端(pgvector等)的职责;引用实现面向测试/验证场景;调用方负责传入等长向量 |
| 6. 错误类型 | 复用 `MemoryError` 现有变体 | 新增 `VecRetrieval` 变体 | 向量检索与关键词检索语义等价于"检索";`RetrievalError` 变体已覆盖索引/评分异常场景 |
| 7. wiremock 测试组织 | 自包含(每个测试启动自己的 MockServer | 共享 helper 函数 | 沿用现有测试模式(openai.rs line 825+、anthropic.rs line 900+);自包含测试可独立运行、定位更直接 |
| 8. 请求头验证 | 做(`body_partial_json` + `header` matcher) | 跳过 | 回归防御价值高——Provider 请求体结构变化会直接导致请求被拒绝,头验证是低成本高收益的回归保护 |
| 9. 并发测试模式 | 100 并发写 + 混合读写(5 读 + 5 写)双模式 | 只做 100 并发写 | 两种模式互补:纯写入验证数据完整性和无 id 重复;混合读写验证读操作在并发写期间不 panic 且返回有效数据 |
| 10. 实施顺序 | 11.1 → 11.2 → 11.3 | 任意顺序 | 与 roadmap 原定的 Step 顺序一致;11.1 是纯新增可独立交付;11.2/11.3 是对既有代码的测试追加,可并行但不优先于 11.1 |
## 设计方案
### Step 11.1 — VectorRetriever trait + InMemoryVectorRetriever
#### 文件位置
- 新增:`src/memory/vector.rs`
- 修改:`src/memory.rs`+2 行:module 声明 + re-export
#### Trait 定义
```rust
/// 语义向量检索器抽象接口。
///
/// 下游可实现此 trait 以对接向量数据库(pgvector / qdrant / lancedb 等)。
/// 默认引用实现 [`InMemoryVectorRetriever`] 基于进程内 HashMap + 余弦相似度。
///
/// **稳定性**:实验性 APIv0.2.x),方法签名可能在 v0.3 中调整。
/// 若未来需要 `remove()` / `clear()` 等方法,将在此 trait 中追加(带默认实现)。
#[async_trait]
pub trait VectorRetriever: Send + Sync {
/// 将 `id` 对应的文本向量 `embeddings` 加入索引。
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError>;
/// 检索与 `query` 向量最相似的 `k` 条记录。
/// 返回 `Vec<(id, score)>`,按 score 降序排列,score ∈ [0.0, 1.0]。
async fn search(&self, query: Vec<f32>, k: usize) -> Result<Vec<(String, f32)>, MemoryError>;
}
```
#### InMemoryVectorRetriever 实现要点
```rust
pub struct InMemoryVectorRetriever {
vectors: Mutex<HashMap<String, Vec<f32>>>,
}
impl InMemoryVectorRetriever {
pub fn new() -> Self {
Self {
vectors: Mutex::new(HashMap::new()),
}
}
}
#[async_trait]
impl VectorRetriever for InMemoryVectorRetriever {
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError> {
let mut vectors = self.vectors.lock().map_err(|e| {
MemoryError::RetrievalError(format!("lock poisoned: {e}"))
})?;
vectors.insert(id, embeddings);
Ok(())
}
async fn search(&self, query: Vec<f32>, k: usize) -> Result<Vec<(String, f32)>, MemoryError> {
let vectors = self.vectors.lock().map_err(|e| {
MemoryError::RetrievalError(format!("lock poisoned: {e}"))
})?;
if vectors.is_empty() || k == 0 {
return Ok(Vec::new());
}
let query_norm = dot(&query, &query).sqrt();
if query_norm == 0.0 {
return Ok(Vec::new());
}
let mut scored: Vec<(String, f32)> = vectors
.iter()
.map(|(id, vec)| {
let dot_product = dot(&query, vec);
let vec_norm = dot(vec, vec).sqrt();
let similarity = dot_product / (query_norm * vec_norm + 1e-10);
(id.clone(), similarity)
})
.collect();
// 降序排列
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
scored.truncate(k);
Ok(scored)
}
}
/// 点积(手动循环,零依赖)。
///
/// 注意:`zip` 对不等长向量静默截断到较短者。引用实现不做维度校验,
/// 调用方应确保 `a` 和 `b` 等长——不等长时结果无意义但不 panic。
fn dot(a: &[f32], b: &[f32]) -> f32 {
a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
}
```
#### 边界与约束
- **无维度校验**:不同维度向量传入 `search()` 时点积不报错,但余弦相似度结果无意义。维度校验是具体后端(pgvector等)的职责,引用实现不做运行时检查。
- **零向量处理**:query 为零向量时直接返回空结果(`query_norm == 0.0`)。
- **1e-10 防零除**:避免空库或全零向量导致除零 panic。
#### 测试(4 个)
| # | 测试名 | 验证点 |
|---|--------|--------|
| 1 | `basic_index_and_search` | index 两条("rust" + "python"),用 "rustacean" 查询应排在首位 |
| 2 | `search_empty_store` | 空库返回空 Vec |
| 3 | `concurrent_index` | 10 个 task 各 index 1 条,总量 10、id 无重复 |
| 4 | `concurrent_index_and_search` | 10 个 writer + 5 个 searcher 并发 2 秒,`search()` 遍历期间 `index()` 写入锁竞争不 panic |
#### 修改 `src/memory.rs`
```rust
pub mod vector;
// 在高频 re-export 区追加
pub use vector::{InMemoryVectorRetriever, VectorRetriever};
```
### Step 11.2 — wiremock Provider roundtrip 测试
#### 测试清单
全部 12 个测试均遵循现有自包含模式:`MockServer::start()``Mock::given(...).and(...).respond_with(...)``provider.chat_blocking(...)` / `provider.chat_stream_inner(...)` → assert。
##### P07 个)
| # | 测试名 | 所属文件 | Mock 关键点 | 断言 |
|---|--------|---------|------------|------|
| 1 | `openai_request_body_format` | `openai.rs` | `body_partial_json` 匹配 `{"model": "gpt-4o", "messages": [{"role": "user"}]}` | 请求体结构正确,响应解析正常 |
| 2 | `openai_authorization_header` | `openai.rs` | `header("authorization", "Bearer sk-test")` | header 精确匹配,响应解析正常 |
| 3 | `openai_401_structured_error` | `openai.rs` | 返回 401 + `{"error": {"message": "Incorrect API key", "code": "invalid_api_key"}}` | `LlmError::Authentication(msg)` 且 message 包含 "Incorrect API key" |
| 4 | `anthropic_401_structured_error` | `anthropic.rs` | 返回 401 + `{"error": {"type": "authentication_error", "message": "Invalid API key provided"}}` | `LlmError::Authentication(msg)` 且 message 包含 "Invalid API key" |
| 5 | `openai_429_with_retry_after` | `openai.rs` | 返回 429 + `{"error": {"message": "Rate limit exceeded"}}` + `retry-after: 30` 头 | `LlmError::RateLimit { retry_after: Some(30s) }` |
| 6 | `openai_tool_use_response` | `openai.rs` | 返回包含 `tool_calls` 的响应(choices[0].message.tool_calls ≠ null | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
| 7 | `anthropic_tool_use_response` | `anthropic.rs` | 返回含 `type: "tool_use"` content block + `stop_reason: "tool_use"`Anthropic 独立 wire 格式) | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
##### P15 个)
| # | 测试名 | 所属文件 | Mock 关键点 | 断言 |
|---|--------|---------|------------|------|
| 8 | `anthropic_version_header` | `anthropic.rs` | `header("anthropic-version", "2023-06-01")` | header 精确匹配 |
| 9 | `openai_stream_usage_only_last_chunk` | `openai.rs` | 流式最后 chunk `{"choices":[],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}` | 不 panic`MessageComplete` 包含正确 usage |
| 10 | `anthropic_529_overloaded_structured` | `anthropic.rs` | 返回 529 + `{"error": {"type": "overloaded_error", "message": "Overloaded"}}` | `LlmError::RateLimit { retry_after: None }` |
| 11 | `openai_500_structured_error` | `openai.rs` | 返回 500 + `{"error": {"message": "Internal server error", "type": "server_error"}}` | `LlmError::Request { status: 500, body }` 且 body 包含 "Internal server error" |
| 12 | `openai_stream_mid_stream_error` | `openai.rs` | 流式前几个 chunk 正常,中途服务端断开连接(模拟网络中断/限流断开) | `LlmError::Request(_)` — 流式中断映射为请求错误 |
#### 测试模式说明
```rust
// 每个测试自包含,不抽共享 helper(沿用现有模式)
#[tokio::test]
async fn openai_authorization_header() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/chat/completions"))
.and(header("authorization", "Bearer sk-test"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"id": "chatcmpl-hdr",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "OK"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
})))
.mount(&server)
.await;
let provider = GenericOpenaiProvider::new_with_name(
server.uri(), "sk-test".into(), "gpt-4o".into(), "openai", 30,
);
let response = provider.chat_blocking(MessageRequest {
model: "gpt-4o".into(),
messages: vec![Message::user_text("hi")],
..Default::default()
}).await.unwrap();
assert_eq!(response.text(), "OK");
}
```
#### 新增依赖
dev-dependencies 中 wiremock 已就绪(当前 openai.rs / anthropic.rs 已在测试中使用),无需新增。
### Step 11.3 — 并发测试补强
#### 测试清单
| # | 测试名 | Store | 模式 | 验证标准 |
|---|--------|-------|------|---------|
| 1 | `concurrent_writers_max_pressure` | InMemoryStore | 100 task × 1 write | 总量 100, id 无重复 |
| 2 | `concurrent_writers_max_pressure` | SqliteStore | 100 task × 1 write | 总量 100, id 无重复 |
| 3 | `concurrent_mixed_read_write` | InMemoryStore | 预热 20 条, 5 读 + 5 写并发 2 秒 | 无 panic |
| 4 | `concurrent_mixed_read_write` | SqliteStore | 预热 20 条, 5 读 + 5 写并发 2 秒 | 无 panic |
| 5 | `concurrent_capacity_eviction` | InMemoryStore | 15 写者, max_items=10 | 最终 ≤ 10 |
#### 关键实现要点
**100 并发写模式**InMemoryStore + SqliteStore 各一):
```rust
#[tokio::test]
async fn concurrent_writers_max_pressure() {
let store = Arc::new(InMemoryStore::new());
let mut handles = Vec::new();
for i in 0..100 {
let s = Arc::clone(&store);
handles.push(tokio::spawn(async move {
let id = format!("concurrent_{i}");
s.save(make_item(&id)).await.unwrap();
}));
}
for h in handles {
h.await.unwrap();
}
let list = store.list(&MemoryFilter::default()).await.unwrap();
assert_eq!(list.len(), 100);
let mut ids: Vec<String> = list.iter().map(|v| v.id.clone()).collect();
ids.sort();
ids.dedup();
assert_eq!(ids.len(), 100);
}
```
**混合读写模式**InMemoryStore + SqliteStore 各一):
```rust
#[tokio::test]
async fn concurrent_mixed_read_write() {
let store = Arc::new(InMemoryStore::new());
// 预热
for i in 0..20 {
store.save(make_item(&format!("seed_{i}"))).await.unwrap();
}
let mut handles = Vec::new();
// 5 个写者
for w in 0..5 {
let s = Arc::clone(&store);
handles.push(tokio::spawn(async move {
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
let mut i = 0;
while tokio::time::Instant::now() < deadline {
let id = format!("writer{w}_item{i}");
s.save(make_item(&id)).await.unwrap();
i += 1;
}
}));
}
// 5 个读者
for r in 0..5 {
let s = Arc::clone(&store);
handles.push(tokio::spawn(async move {
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
while tokio::time::Instant::now() < deadline {
let _ = s.list(&MemoryFilter::default()).await.unwrap();
}
}));
}
for h in handles {
h.await.unwrap();
}
// 不 panic 即算通过
}
```
**容量淘汰并发模式**InMemoryStore):
```rust
#[tokio::test]
async fn concurrent_capacity_eviction() {
let eviction = EvictionConfig {
policy: EvictionPolicy::Capacity { max_items: 10 },
check_interval: 1,
};
let store = Arc::new(InMemoryStore::with_eviction(eviction));
let mut handles = Vec::new();
for i in 0..15 {
let s = Arc::clone(&store);
handles.push(tokio::spawn(async move {
s.save(make_item(&format!("item_{i}"))).await.unwrap();
}));
}
for h in handles {
h.await.unwrap();
}
let list = store.list(&MemoryFilter::default()).await.unwrap();
assert!(list.len() <= 10);
}
```
**测试归属**
- InMemoryStore 并发测试 → `src/memory/store/in_memory.rs``mod tests`
- SqliteStore 并发测试 → `src/memory/store/sqlite_store.rs``mod tests`
- 混合读写测试中的 `make_item` 辅助函数:直接复用各文件现有 `fn make_item`
## 已否决的方案
### 1. 砍掉 11.1PM 建议)
**内容**PM 在讨论中提出 VectorRetriever trait 无消费者,建议整体砍掉,等 Phase 12 或 v0.3 有人用时再做。
**否决理由**:引用实现作为 trait 契约的编译期验证手段——没有 consumer 不意味着 trait 签名不需要测试。同时社区贡献(pgvector adapter 等)需要稳定的 trait 边界。附带引用实现还可作为"如何在 agcore 中实现一个 VectorRetriever"的示例,降低社区参与门槛。代码量仅 ~80 行,维护成本可忽略。
### 2. trait-only VectorRetriever(无引用实现)
**内容**:只定义 `VectorRetriever` trait,不做 `InMemoryVectorRetriever`
**否决理由**trait-only 是纯推测代码——没有运行时验证,无法确认 trait 方法签名在实际调用链中是否可编译。理想情况下每个 trait 至少有一个引用实现来验证"这个 trait 确实可以被实现"。
### 3. 跳过请求头验证
**内容**:请求体格式和 Authorization header 验证是"过度保护"。
**否决理由**`body_partial_json` + `header` matcher 的回归防御价值高。Provider 适配层的最大风险是请求体结构无声变更(如 `ToolDef` IR 切换时漏改了序列化字段),头验证是低成本(每测试 ~5 行)高收益的回归保护。
### 4. 先发 v0.2.0 正式版再迭代
**内容**:当前 rc.1 已经包含所有 P0 功能,建议直接发正式版,Phase 11 推迟到 v0.2.1。
**否决理由**:测试补强是正式版的信号而非负担。Phase 11 的三个 Step 都是"如果现在不做,以后更不会做"的类型。在正式版前补齐测试基线,避免「发布了再补测试」的经典陷阱。
### 5. 只做 100 并发写,不做混合读写
**内容**:并发写验证数据完整性已足够。
**否决理由**:纯写入和混合读写暴露不同类型的 bug。纯写入验证"数据不丢、id 无重复";混合读写验证"读操作在并发写期间不 panic、返回有效数据"。两种模式互补缺失。
### 6. (实施后补充)openai_stream_mid_stream_error 的 mock 模式偏差
**实际实施**:返回 `200 + SSE content-type + 畸形 JSON payload``data: {not-valid-json}\n\ndata: [DONE]\n\n`),断言 `ChunkToEventStream` 产出 `StreamEvent::Error`
**方案原文**:返回"前几个 chunk 正常,中途服务端断开连接",断言 `LlmError::Request(_)`
**偏差原因**wiremock 0.6 标准 responder 的 `set_delay` / `set_body_string` 行为是「延迟响应 + 发完 body 后关闭连接」,无法精确模拟「send partial body then hang 保持连接」。`read_timeout` 配合 `set_delay` 触发的超时属于 send 阶段(`LlmError::Timeout`),不属于流式中断。
**采纳方案**:用畸形 JSON payload 替代——同样验证"流中途产生错误事件而不 panic"的回归保护意图,且在 wiremock 0.6 上 100% 可重现。断言改为 `StreamEvent::Error{message}` + "stream 最终结束",保持核心回归价值。
**影响**:测试意图(流阶段错误检测)完全保留;mock 行为从「TCP 断开」变为「畸形应用层数据」;断言从 `LlmError::Request` 改为 `StreamEvent::Error`(语义等价:客户端发现流异常)。
### 7. (实施后补充)openai.rs 的 429 retry-after 解析修复
**实际实施**`handle_error_response` 新增 `retry-after` header 解析逻辑(与 anthropic.rs 完全对齐)。
**方案原文**:方案测试 11.2.4 要求 `RateLimit { retry_after: Some(30s) }`,但 ADC 表中未显式列出此修复作为生产代码变更。
**修复原因**:原 `openai.rs:186-191` 的 429 分支固定 `retry_after: None`,与 `anthropic.rs:339-351` 已有的解析逻辑不一致。原代码注释甚至已写"仅读取 retry-after",但实际未实现——这是隐藏 bug。修复让 OpenAI 兼容层(DeepSeek/Qwen 等)的限流重试信息可用,与 Anthropic 行为统一。
**影响**:方案测试 11.2.4 从「不可通过的回归保护」变为「可验证的实际行为」。变更 5 行,与 anthropic 实现完全镜像。
## 实施计划与顺序
**实施顺序**11.1 → 11.2 → 11.3(与 roadmap 一致,每步可单独交付验证)。
| Step | 文件变更 | 测试增量 | 预估代码量 | 验证标准 |
|------|---------|---------|-----------|---------|
| 11.1 | +`src/memory/vector.rs`~80 行),~`src/memory.rs`+2 行) | +4 | ~85 行实现 + 70 行测试 | `cargo build --all-targets` 编译通过,4 个测试通过 |
| 11.2 | ~`src/llm/provider/openai.rs`+6 个测试),~`src/llm/provider/anthropic.rs`+2 个测试) | +127 P0 + 5 P1 | ~290 行(含 test mod 和 mock 数据) | `cargo test --all-targets` 全绿,wiremock 12 场景均绿 |
| 11.3 | ~`src/memory/store/in_memory.rs`+3 个测试),~`src/memory/store/sqlite_store.rs`+2 个测试) | +5 | ~120 行 | `cargo test --all-targets` 全绿 |
| **总计** | 6 个文件(1 新增 + 5 修改) | +21 | ~500 行 | 全量 254 → 275+`cargo clippy --all-targets -- -D warnings` 0 警告 |
### 验证通过标准
1. `cargo build --all-targets` —— 编译通过,无 warning
2. `cargo test --all-targets` —— 全部通过(254 + 21 = 275+
3. `cargo clippy --all-targets -- -D warnings` —— 0 警告
4. `cargo test --all-targets 2>&1 | grep -E "test result:"` —— 确认新增测试全部出现在执行列表中
5. 新增 wiremock 测试单独验证网络隔离(无需 API key,纯本地 mock
## 参考来源
- **讨论收口结论**Phase 11 讨论,含 PM/SA 双视角输入(2026-07-07
- **现有代码模式**
- `src/llm/provider/openai.rs` line 824-1034——wiremock 测试模式(`MockServer::start``Mock::given(...).and(...).respond_with(...)``provider.chat_blocking` → assert
- `src/llm/provider/anthropic.rs` line 899-1079——Anthropic provider wiremock 测试
- `src/memory/store/sqlite_store.rs` line 458-483——`concurrent_writers_no_data_loss` 10×10 并发模式
- `src/memory/store.rs`——`MemoryStore` trait 定义(`#[async_trait]` 风格)
- `src/memory/error.rs`——`MemoryError` 枚举(`#[non_exhaustive]` + `RetrievalError` 变体)
- `src/memory/retriever.rs`——现有检索模块(`RetrievalResult` / `ScoredItem`
- `src/memory.rs`——模块根与 re-export 模式
- **方案文档**`docs/roadmap.md` Phase 11 章节(line 516-528
- **编译器 pragma**`#[non_exhaustive]` —— 新增枚举变体需要此标记,公开结构体字段未来变化预留兼容空间
## 关键假设与风险
### 关键假设清单
| # | 假设 | 影响 | 推翻后的应对 |
|---|------|------|------------|
| 1 | wiremock `body_partial_json` matcher 在 wiremock 0.6+ 中可用 | Step 11.2 测试 #1 的实现方式 | 改用 `body_json`(精确匹配)或 `body_string`(部分串匹配) |
| 2 | `tokio::spawn` 100 task 并发写入 `Mutex<HashMap>` 无死锁 | Step 11.3 #1 InMemoryStore 并发 | 降低并发数(50)继续验证,或换 `tokio::sync::Mutex` |
| 3 | SqliteStore 的 `busy_timeout=5000` 能容忍 100 并发写 | Step 11.3 #2 SqliteStore 并发 | 增加 `busy_timeout`10s),或限制最大并发数 |
| 4 | 新增测试不使用 wiremock 以外的未列在 dev-dependencies 中的依赖 | Phase 11 零新增外部依赖 | 若需要额外 matcher,评估后加入 dev-dependencies |
| 5 | InMemoryVectorRetriever 的 O(n) 全量扫描在测试规模下 (<1000 向量) 性能可接受 | Step 11.1 测试通过 | 如果有竞态问题,改为 read/write lock`RwLock<HashMap>` |
| 6 | `MemoryError::RetrievalError` 变体足够覆盖向量检索的索引/评分失败场景 | Step 11.1 错误映射 | 如果不够,可增加新的 `MemoryError` 变体 |
| 7 | OpenAI 和 Anthropic 的结构化错误体格式在当前 SDK 版本中未变化 | Step 11.2 测试 #3/#4/#7/#10/#11(结构化错误解析断言) | 若 SDK 变更错误体格式,更新 mock body 和断言匹配新格式 |
| 8 | 调用方传入 `search()` 的向量与已索引向量维度一致(引用实现不做维度校验,`dot()` 对不等长向量静默截断) | Step 11.1 InMemoryVectorRetriever 正确性 | 若须维度校验,在 `index()` 时记录维度并在 `search()` 时断言;引用实现维持零校验 |
### 已识别的风险
| 风险 | 等级 | 缓解措施 |
|------|------|---------|
| wiremock `body_partial_json` matcher 行为在版本升级后变化 | 低 | 限定 wiremock 版本范围(当前已在 Cargo.lock 中锁定);P0 测试不依赖该 matcher |
| 100 并发写暴露 SqliteStore 的 `spawn_blocking` 线程池瓶颈 | 中 | 观察 CI 执行时间;如果超时,降低并发到 50 或增加 `max_blocking_threads` |
| InMemoryVectorRetriever 的 `Mutex` 锁争用导致测试 flaky | 低 | Mutex 不会死锁(单线程持有不 await),测试不依赖精确时序 |
| 新增 wiremock 测试与现有测试冲突(端口占用) | 低 | `MockServer::start()` 自动选择随机端口,不冲突 |
| `cargo test --all-targets` 执行时间增加 >30% | 低 | 预估 +21 个测试,增量约 8%254→275),其中 wiremock 测试有网络 IO 但延迟 <10ms/个 |
### 非阻塞已知项
- **Ollama Provider** 是 OpenAI Compatwiremock 测试继承 `GenericOpenaiProvider`,不单独新增
- **DeepSeek / Qwen Provider** 同样通过 `GenericOpenaiProvider` 实现,继承测试
- **Step 11.1 不消费到 AgentSession / ContextSlot**,留待 Phase 12 或 v0.3 做消费端集成
- **Phase 11 完成后**,全量测试预计 275+,`cargo test --all-targets` 执行时间预计 < 60s
---
## 实施计划(附录)
**实施顺序**11.1 → 11.2 ‖ 11.311.2 与 11.3 无文件冲突,可并行交付;11.2 优先因回归保护价值更高)。每步完成后运行 `cargo test --all-targets` + `cargo clippy --all-targets -- -D warnings` 验证无回归。
---
### Step 11.1 — VectorRetriever trait + InMemoryVectorRetriever~160 行,4 测试)
**前置**:无。与 Step 11.2/11.3 可并行开发但优先交付。
| 任务 | 描述 | 文件 | 前置 | 工作量 | 风险 | 验收条件 |
|------|------|------|------|--------|------|---------|
| **11.1.1** | 创建 `memory/vector.rs`:定义 `VectorRetriever` trait + `InMemoryVectorRetriever` struct + `dot()` 辅助函数 | `src/memory/vector.rs` | 无 | S | 低 | `cargo build --all-targets` 编译通过 |
| **11.1.2** | 实现 `InMemoryVectorRetriever``index()` — Mutex insert`search()` — 全量余弦扫描 + 降序排列 + k 截断 | `src/memory/vector.rs` | 11.1.1 | S | 低 | trait 实现编译通过;`Mutex::lock()` 使用 `map_err` 处理 poison,不 panic |
| **11.1.3** | 添加 4 个内联测试:`basic_index_and_search``search_empty_store``concurrent_index``concurrent_index_and_search` | `src/memory/vector.rs` (mod tests) | 11.1.2 | S | 低 | 4 测试全部通过 |
| **11.1.4** | 修改 `src/memory.rs`:加 `pub mod vector;` + `pub use vector::{VectorRetriever, InMemoryVectorRetriever};` | `src/memory.rs` | 11.1.1 | S | 低 | `cargo build --all-targets` 无 warning |
**Step 验证**
```
cargo test --all-targets # 254 + 4 = 258+ passed
cargo clippy --all-targets -- -D warnings # 0 warning
```
---
### Step 11.2 — wiremock Provider roundtrip 测试(12 测试,P0=7 + P1=5
**前置**:无。与 Step 11.1 无文件冲突,可并行。
**`openai.rs` 新增测试(8 个:P0=5 + P1=3**
| 任务 | 测试名 | 优先级 | 前置 | 工作量 | 风险 | Mock 模式 | 验收条件 |
|------|--------|--------|------|--------|------|----------|---------|
| **11.2.1** | `openai_request_body_format` | P0 | 无 | S | 低 | `body_partial_json` 匹配 model/messages | 请求体结构正确,响应解析正常 |
| **11.2.2** | `openai_authorization_header` | P0 | 无 | S | 低 | `header("authorization", "Bearer sk-test")` | header 精确匹配 |
| **11.2.3** | `openai_401_structured_error` | P0 | 无 | S | 低 | 401 + `{"error":{"message":"...","code":"invalid_api_key"}}` | `LlmError::Authentication` 含 "Incorrect API key" |
| **11.2.4** | `openai_429_with_retry_after` | P0 | 无 | S | 低 | 429 + `retry-after: 30` | `RateLimit { retry_after: Some(30s) }` |
| **11.2.5** | `openai_tool_use_response` | P0 | 无 | S | 低 | 响应含 `tool_calls` | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
| **11.2.6** | `openai_stream_usage_only_last_chunk` | P1 | 无 | S | 低 | 流式最后 chunk `{choices:[], usage:{...}}` | 不 panic`MessageComplete` 含正确 usage |
| **11.2.7** | `openai_500_structured_error` | P1 | 无 | S | 低 | 500 + `{"error":{"message":"server error"}}` | `Request { status: 500 }` 含 body |
| **11.2.8** | `openai_stream_mid_stream_error` | P1 | 无 | M | 中 | 前几个 chunk 正常后连接断开 | `LlmError::Request(_)` 流中断映射 |
**`anthropic.rs` 新增测试(4 个:P0=2 + P1=2**
| 任务 | 测试名 | 优先级 | 前置 | 工作量 | 风险 | Mock 模式 | 验收条件 |
|------|--------|--------|------|--------|------|----------|---------|
| **11.2.9** | `anthropic_401_structured_error` | P0 | 无 | S | 低 | 401 + `{"error":{"type":"authentication_error","message":"..."}}` | `LlmError::Authentication` 消息透传 |
| **11.2.10** | `anthropic_tool_use_response` | P0 | 无 | S | 低 | 响应含 `type:"tool_use"` content block + `stop_reason:"tool_use"` | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
| **11.2.11** | `anthropic_version_header` | P1 | 无 | S | 低 | `header("anthropic-version", "2023-06-01")` | header 精确匹配 |
| **11.2.12** | `anthropic_529_overloaded_structured` | P1 | 无 | S | 低 | 529 + `{"error":{"type":"overloaded_error","message":"Overloaded"}}` | `RateLimit { retry_after: None }` |
**Step 验证**
```
cargo test --all-targets # 258 + 12 = 270+ passed
cargo clippy --all-targets -- -D warnings # 0 warning
```
每个测试自包含(`MockServer::start()``Mock::given(...)``provider.chat_blocking()/chat_stream_inner()` → assert),无需共享 helper。
---
### Step 11.3 — 并发测试补强(5 测试)
**前置**:无。与 Step 11.1/11.2 无文件冲突。
| 任务 | 测试名 | 文件 | 前置 | 工作量 | 风险 | 模式 | 验收条件 |
|------|--------|------|------|--------|------|------|---------|
| **11.3.1** | `concurrent_writers_max_pressure` | `in_memory.rs` | 无 | S | 中 | 100 task × 1 write | 总量 100id 无重复 |
| **11.3.2** | `concurrent_writers_max_pressure` | `sqlite_store.rs` | 无 | S | 中 | 100 task × 1 write | 总量 100id 无重复 |
| **11.3.3** | `concurrent_mixed_read_write` | `in_memory.rs` | 无 | S | 中 | 预热 20 条,5 写 + 5 读并发 2 秒 | 无 panic |
| **11.3.4** | `concurrent_mixed_read_write` | `sqlite_store.rs` | 无 | S | 中 | 预热 20 条,5 写 + 5 读并发 2 秒 | 无 panic |
| **11.3.5** | `concurrent_capacity_eviction` | `in_memory.rs` | 无 | S | 中 | 15 写者,max_items=10 | 最终 ≤ 10(竞争激烈时可能过渡态 >10,主断言 ≤ 10,宽松备选 ≤ 15) |
**实现要点**
- 沿用现有 `Arc<Store> + tokio::spawn + h.await.unwrap()` 模式(参考 `sqlite_store.rs:458-483`
- `make_item` 辅助函数直接复用各文件现有实现
- SqliteStore 测试使用 `:memory:` 数据库(与现有并发测试一致)
- 混合读写测试使用 `tokio::time::Instant::now() + Duration` 做时限
- 100 并发写是一次性 spawn 100 task(非分批),暴露最大锁竞争压力
**Step 验证**
```
cargo test --all-targets # 270 + 5 = 275+ passed
cargo clippy --all-targets -- -D warnings # 0 warning
```
---
### 整体发布核查清单
| # | 检查项 | 验证命令 | 预期结果 |
|---|--------|---------|---------|
| 1 | 编译 | `cargo build --all-targets` | 通过,0 warning |
| 2 | 全量测试 | `cargo test --all-targets` | 275+ passed0 failed |
| 3 | Lint | `cargo clippy --all-targets -- -D warnings` | 0 warning |
| 4 | 文档 | `cargo doc --no-deps` | 0 warningVectorRetriever trait 公共 API doc 完整) |
| 5 | 确认新增测试 | `cargo test --all-targets 2>&1 \| grep -E "test result:"` | 所有新增测试名出现在执行列表中 |
| 6 | wiremock 隔离 | 新增 wiremock 测试不依赖网络 | 纯本地 mock,无需 API key |
| 7 | 并行安全 | 并发测试独立运行时无 flaky | 连续 3 次 `cargo test` 结果一致 |
| 8 | 存量零回归 | 已有 254 个测试全部通过 | 与 Phase 10 基线对比无 fail |
| 9 | 公共 API doc comment | `grep -r "pub trait VectorRetriever" src/ && rg "^///" -c src/memory/vector.rs` | trait 和方法都有 `///` 注释 |
**若核查项失败的回退策略**
- **测试失败(P0)**:阻断发布。定位到具体测试名 → 检查 Mock JSON 格式与 Provider 解析逻辑是否匹配(结构化错误体格式变化 → 更新 mock body;流式状态机变化 → 更新 `chat_stream_inner` 路径测试)
- **测试失败(P1)**:不阻断发布。标记 `#[ignore]` + file issue,确认无 P0 失败后即可发布
- **clippy warning**:修复 lint 后重跑;若为 `#[allow(...)]` 可抑制,在 code review 中申明理由
- **flaky 并发测试**:检查 `tokio::spawn` 是否跨 `.await` 持锁;若 SqliteStore 超时,增加 `busy_timeout` 或降低并发数
- **11.1 模块发布阻塞**:若 `InMemoryVectorRetriever` 无法按时交付,可临时注释 `src/memory.rs` 中的 `pub mod vector;` 行,跳过整个模块(零消费者,不影响发布)。回退后再补交