fix(agent): Phase 9 实施审查修复
- cycle.rs: run_tool_loop 实现 PreRequest hook(之前 `let _ = hook_executor.as_ref()` 是空操作,
导致 hook-based logging/monitoring 在流式工具循环中失效;现在与 submit_with_tools 行为对齐,
含 should_block 检查,阻断时通过 StreamEvent::Error 事件化)
- session.rs: 删除 submit_turn_stream 末尾的 `let _ = hook_executor;` 死代码(Arc 引用生命周期
由 Arc 自动管理)
- session.rs: 新增 2 个集成测试覆盖方案 §4 Step 5:
- submit_turn_stream_end_to_end:mock provider → 消费流 → finalize_turn 后 cost_so_far
正确更新(10/5 tokens)+ turn_index=1 + default slot 包含 user/assistant 消息
- submit_turn_stream_triggers_turn_hooks:OnTurnStart 在 submit_turn_stream 返回流前
触发(计数=1)+ OnTurnEnd 在 finalize_turn 前不触发(计数=0)+ finalize_turn 后触发(计数=1)
- docs/16-phase9-streaming-experience.md: 标注 finalize_turn Phase 10 签名变更(new_messages_from_cycle
+ Result 返回),Step 5 测试实现位置
- 测试 288 passed / 0 failed(基线 286 + 2 新增),clippy 0 警告,doc 0 warning
This commit is contained in:
@@ -241,6 +241,12 @@ pub async fn submit_turn_stream(
|
||||
pub async fn finalize_turn(&mut self, response: &MessageResponse)
|
||||
```
|
||||
|
||||
> **实施偏差(Phase 10 适配)**:实际签名扩展为
|
||||
> `pub async fn finalize_turn(&mut self, response: &MessageResponse, new_messages_from_cycle: Vec<Message>) -> Result<(), AgentError>`。
|
||||
> - `new_messages_from_cycle`:本轮新增消息(`[user_input, ...tool_results, final_response]`),由消费者在流消费完毕后从 `cycle.messages()[input_len..]` 提取并传入;`finalize_turn` 增量追加到当前 slot(不覆盖已有消息)。
|
||||
> - 返回 `Result<(), AgentError>`:错误传播更清晰,与 `submit_turn` 的 slot 边界错误(`SlotReadonly` / `SlotNotFound`)对齐。
|
||||
> - Phase 10 ContextSlot 实施时扩展。Phase 9 消费者若不接入 slot 持久化,可传 `vec![response.message.clone()]` 兜底。
|
||||
|
||||
### 3.5 消费者使用模式
|
||||
|
||||
```rust
|
||||
@@ -410,10 +416,10 @@ if let Some(response) = final_response {
|
||||
|
||||
**目标**:端到端验证 `submit_turn_stream` + `finalize_turn` 的完整链路,确保零回归。
|
||||
|
||||
**新增**(`agent/session.rs` 内联测试):
|
||||
**新增**(`agent/session.rs` 内联测试,2026-07-08 实施审查补全):
|
||||
|
||||
- **场景**:`submit_turn_stream` 跑通 mock provider → 消费流(验证各事件到达) → `finalize_turn` 后 cost 更新正确
|
||||
- **场景**:verify `OnTurnStart` hook 在 `submit_turn_stream` 返回流之前已触发
|
||||
- `submit_turn_stream_end_to_end` — `submit_turn_stream` 跑通 mock provider → 消费流(验证收到 TextDelta + MessageComplete) → `finalize_turn` 后 `cost_so_far` 正确更新(`prompt_tokens=10, completion_tokens=5`) + `turn_index=1` + default slot 包含 user/assistant 消息
|
||||
- `submit_turn_stream_triggers_turn_hooks` — 验证 `OnTurnStart` 在 `submit_turn_stream` 返回流之前已触发(计数=1)+ `OnTurnEnd` 在 `finalize_turn` 之前**不**触发(计数=0)+ `finalize_turn` 后 `OnTurnEnd` 触发(计数=1)
|
||||
|
||||
**验证**:
|
||||
|
||||
|
||||
+105
-3
@@ -418,9 +418,6 @@ impl AgentSession {
|
||||
// 6. turn_index 递增 —— 配合 finalize_turn 用 (turn_index - 1) 传递正确的 OnTurnEnd 序号
|
||||
self.turn_index += 1;
|
||||
|
||||
// 注:hook_executor 不显式 drop,生命周期由 Arc 自动管理
|
||||
let _ = hook_executor;
|
||||
|
||||
Ok(stream)
|
||||
}
|
||||
|
||||
@@ -475,10 +472,12 @@ mod tests {
|
||||
use crate::agent::builder::AgentBuilder;
|
||||
use crate::llm::hooks::{Hook, HookContext, HookExecutor, HookResult};
|
||||
use crate::llm::mock::MockProvider;
|
||||
use crate::llm::stream::StreamEvent;
|
||||
use crate::llm::types::message::ContentBlock;
|
||||
use crate::llm::types::response_v2::{MessageResponse, StopReason};
|
||||
use crate::tools::ToolRegistry;
|
||||
use async_trait::async_trait;
|
||||
use futures_util::StreamExt;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
|
||||
/// 计数 hook —— 每被调用一次 +1。
|
||||
@@ -948,4 +947,107 @@ mod tests {
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, AgentError::SlotReadonly(_)));
|
||||
}
|
||||
|
||||
// ====== Phase 9 Step 5: 集成测试 ======
|
||||
|
||||
/// Phase 9 Step 5.1 — `submit_turn_stream` 端到端链路。
|
||||
///
|
||||
/// 验证:mock provider → `submit_turn_stream` 消费流 → 收到 TextDelta + MessageComplete
|
||||
/// → `finalize_turn` 后 `cost_so_far` 正确更新,turn_index 递增。
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn submit_turn_stream_end_to_end() {
|
||||
let (mut session, _, _) = build_session(vec![assistant_text("hi back")]);
|
||||
|
||||
let mut stream = session
|
||||
.submit_turn_stream("user msg")
|
||||
.await
|
||||
.expect("submit_turn_stream 应成功");
|
||||
|
||||
// 消费流并提取 MessageComplete
|
||||
let mut final_response: Option<MessageResponse> = None;
|
||||
while let Some(ev) = stream.next().await {
|
||||
if let StreamEvent::MessageComplete { full_response } = &ev {
|
||||
final_response = Some(full_response.clone());
|
||||
}
|
||||
}
|
||||
|
||||
let response = final_response.expect("流中应有 MessageComplete");
|
||||
// consumer 负责构造本轮新增消息列表(user_input + assistant_response)。
|
||||
// submit_turn_stream 不会自动写入 self.slots(流是延迟的),
|
||||
// 消费者需在 finalize_turn 时把 [user_input, ...tool_results, final_response] 一并传入。
|
||||
let new_messages = vec![Message::user_text("user msg"), response.message.clone()];
|
||||
session
|
||||
.finalize_turn(&response, new_messages)
|
||||
.await
|
||||
.expect("finalize_turn 应成功");
|
||||
|
||||
// cost_so_far 已累计(assistant_text 的 usage 是 from_input_output(10, 5))
|
||||
assert_eq!(session.usage().total().prompt_tokens, 10);
|
||||
assert_eq!(session.usage().total().completion_tokens, 5);
|
||||
// turn_index 已递增
|
||||
assert_eq!(session.turn_index(), 1);
|
||||
|
||||
// default slot 应包含 user 输入和 assistant 响应
|
||||
let slot = session.slots.get("default").expect("default slot");
|
||||
let has_user = slot
|
||||
.messages
|
||||
.iter()
|
||||
.any(|m| matches!(m, Message::User { .. }));
|
||||
let has_resp = slot.messages.iter().any(|m| extract_text(m) == "hi back");
|
||||
assert!(has_user && has_resp, "default slot 应包含 user 和 assistant 消息");
|
||||
}
|
||||
|
||||
/// Phase 9 Step 5.2 — `submit_turn_stream` 触发 OnTurnStart / OnTurnEnd hook。
|
||||
///
|
||||
/// 验证:OnTurnStart 在 `submit_turn_stream` 返回流之前已触发;
|
||||
/// OnTurnEnd 在 `finalize_turn` 调用后才触发。
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn submit_turn_stream_triggers_turn_hooks() {
|
||||
let (mut session, start_count, end_count) = build_session(vec![assistant_text("ok")]);
|
||||
|
||||
// 初始状态:两个 hook 计数都是 0
|
||||
assert_eq!(start_count.0.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(end_count.0.load(Ordering::SeqCst), 0);
|
||||
|
||||
// 调 submit_turn_stream
|
||||
let mut stream = session
|
||||
.submit_turn_stream("user msg")
|
||||
.await
|
||||
.expect("submit_turn_stream 应成功");
|
||||
|
||||
// OnTurnStart 应在流返回前已触发
|
||||
assert_eq!(
|
||||
start_count.0.load(Ordering::SeqCst),
|
||||
1,
|
||||
"OnTurnStart 应在 submit_turn_stream 返回流之前触发"
|
||||
);
|
||||
// OnTurnEnd 此时尚未触发
|
||||
assert_eq!(
|
||||
end_count.0.load(Ordering::SeqCst),
|
||||
0,
|
||||
"OnTurnEnd 不应在 submit_turn_stream 阶段触发"
|
||||
);
|
||||
|
||||
// 消费流
|
||||
let mut final_response: Option<MessageResponse> = None;
|
||||
while let Some(ev) = stream.next().await {
|
||||
if let StreamEvent::MessageComplete { full_response } = &ev {
|
||||
final_response = Some(full_response.clone());
|
||||
}
|
||||
}
|
||||
|
||||
// finalize_turn
|
||||
let response = final_response.expect("流中应有 MessageComplete");
|
||||
session
|
||||
.finalize_turn(&response, vec![])
|
||||
.await
|
||||
.expect("finalize_turn 应成功");
|
||||
|
||||
// OnTurnEnd 已触发
|
||||
assert_eq!(
|
||||
end_count.0.load(Ordering::SeqCst),
|
||||
1,
|
||||
"OnTurnEnd 应在 finalize_turn 后触发"
|
||||
);
|
||||
}
|
||||
}
|
||||
+16
-3
@@ -18,7 +18,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
use crate::llm::compact::{CompactConfig, CompactState, microcompact, should_compact};
|
||||
use crate::llm::cycle::retry::should_retry;
|
||||
use crate::llm::error::LlmError;
|
||||
use crate::llm::hooks::{HookContext, HookExecutor};
|
||||
use crate::llm::hooks::{HookContext, HookEvent, HookExecutor};
|
||||
use crate::llm::provider::LlmProvider;
|
||||
use crate::llm::stream::StreamEvent;
|
||||
use crate::llm::types::message::{ContentBlock, Message};
|
||||
@@ -774,8 +774,21 @@ async fn run_tool_loop(
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// ② PreRequest hook(fire-and-forget,仅占位保留以保持接口对称)
|
||||
let _ = hook_executor.as_ref();
|
||||
// ② PreRequest hook —— 与 `submit_with_tools` / `submit_stream` 行为对齐:
|
||||
// 触发 hook → 检查 should_block → 阻断则事件化 Error + return 结束 task。
|
||||
// 阻断原因透传,让消费者看到完整的拒绝原因。
|
||||
if let Some(ref executor) = hook_executor {
|
||||
let ctx = HookContext::new(HookEvent::PreRequest).with_request(&request);
|
||||
let results = executor.execute(HookEvent::PreRequest, &ctx).await;
|
||||
if let Some(blocking) = results.iter().find(|r| r.should_block) {
|
||||
let reason = blocking
|
||||
.reason
|
||||
.clone()
|
||||
.unwrap_or_else(|| "Blocked by pre-request hook".to_string());
|
||||
let _ = tx.send(StreamEvent::Error { message: reason });
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// ③ chat_stream —— 第一层错误
|
||||
let mut stream = match provider.chat_stream(request).await {
|
||||
|
||||
Reference in New Issue
Block a user