fix(core): stream tool output live via ToolExecutionUpdate events

The agent passed None for on_update, so tools (e.g. shell) never forwarded
their incremental output — the UI only saw the final result. Now:

- EventSink: Send + Sync (the forwarding closure must satisfy ToolUpdateSink)
- ToolUpdateSink becomes lifetime-parameterized (bare trait-object aliases
  default to 'static, which forbade borrowing the caller's sink); all tool
  impls updated to &ToolUpdateSink<'_>
- Agent::execute_single forwards each tool update as a ToolExecutionUpdate
  event (Mutex provides interior mutability inside the Fn closure)
- TUI: running tools show the tail of the output live (newest lines), done
  tools show the head + remaining-line hint

regression test: a streaming tool's on_update texts appear as agent events in
order.
This commit is contained in:
DaiChaoXiong 2026-08-09 22:55:16 +08:00
parent 0760911e2a
commit 9a2d50ee58
14 changed files with 220 additions and 24 deletions

View File

@ -14,7 +14,7 @@ use crate::error::{CoreError, CoreResult};
use crate::event::{AgentEvent, EventSink};
use crate::model::*;
use crate::provider::{ProviderRequest, StreamEvent, StreamProvider};
use crate::tool::{ToolEffects, ToolRegistry, ToolResult};
use crate::tool::{ToolEffects, ToolRegistry, ToolResult, ToolUpdate};
/// 一次 [`Agent`] 运行的配置。
/// Configuration for an [`Agent`] run.
@ -385,7 +385,35 @@ impl Agent {
.tools
.get(&tc.name)
.ok_or_else(|| CoreError::ToolNotFound(tc.name.clone()))?;
let result = tool.execute(&tc.id, &tc.arguments, None)?;
// 把工具的增量更新(如 shell 流式输出)转发为 ToolExecutionUpdate 事件,
// 让 UI 实时显示输出,而不是等工具结束。
// Forward the tool's incremental updates (e.g. shell streaming output)
// as ToolExecutionUpdate events so the UI shows output live instead of
// waiting for the tool to finish.
let tc_id = tc.id.clone();
let tc_name = tc.name.clone();
// `Fn` 闭包不能可变借用 sink用 Mutex 提供内部可变性(满足
// ToolUpdateSink 的 Send + Sync 约束);闭包在 execute 返回后被丢弃,
// 随后把 sink 取回供 ToolExecutionEnd 使用。
// An `Fn` closure cannot mutably borrow the sink, so a Mutex provides
// interior mutability (satisfying ToolUpdateSink's Send + Sync bound);
// the closure is dropped when execute returns, then the sink is
// recovered for ToolExecutionEnd.
let sink_mutex = std::sync::Mutex::new(sink);
let on_update = |update: ToolUpdate| {
if let Ok(mut sink) = sink_mutex.lock() {
sink.emit(AgentEvent::ToolExecutionUpdate {
tool_call_id: tc_id.clone(),
tool_name: tc_name.clone(),
partial_result: update,
});
}
};
let result = tool.execute(&tc.id, &tc.arguments, Some(&on_update))?;
// NLL 在最后使用后结束闭包对 sink_mutex 的借用,随后取回 sink。
// NLL ends the closure's borrow of sink_mutex after its last use;
// the sink is then recovered.
let sink = sink_mutex.into_inner().expect("sink mutex not poisoned");
sink.emit(AgentEvent::ToolExecutionEnd {
tool_call_id: tc.id.clone(),
tool_name: tc.name.clone(),

View File

@ -61,7 +61,12 @@ pub enum AgentEvent {
/// [`AgentEvent`] 的接收端。
/// A sink for [`AgentEvent`]s.
pub trait EventSink: Send {
///
/// `Sync` 是必需的:工具增量更新转发为事件时,闭包要满足
/// [`ToolUpdateSink`](crate::tool::ToolUpdateSink)`Fn + Send + Sync`)约束。
/// `Sync` is required: forwarding tool updates as events needs a closure that
/// satisfies [`ToolUpdateSink`](crate::tool::ToolUpdateSink) (`Fn + Send + Sync`).
pub trait EventSink: Send + Sync {
/// 接收一个事件。
/// Receive one event.
fn emit(&mut self, event: AgentEvent);

View File

@ -182,13 +182,19 @@ pub trait Tool: Send + Sync + fmt::Debug {
&self,
tool_call_id: &str,
arguments: &JsonValue,
on_update: Option<&ToolUpdateSink>,
on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult>;
}
/// 工具用来推送增量更新的 channel 式 sink。
/// A channel-style sink a tool uses to push incremental updates.
pub type ToolUpdateSink = dyn Fn(ToolUpdate) + Send + Sync;
/// 工具用来推送增量更新的回调。
/// A callback a tool uses to push incremental updates.
///
/// 带生命周期参数:裸 trait object 别名的默认对象生命周期是 `'static`
/// 会导致增量回调无法借用调用方的局部状态;`'_` 允许短生命周期对象。
/// Lifetime-parameterized: a bare trait-object alias defaults to `'static`,
/// which would forbid callbacks borrowing caller-local state; `'_` admits
/// short-lived objects.
pub type ToolUpdateSink<'a> = dyn Fn(ToolUpdate) + Send + Sync + 'a;
/// 可用工具的注册表,按名查找。
/// A registry of available tools, looked up by name.

View File

@ -33,7 +33,7 @@ impl Tool for EchoTool {
&self,
_id: &str,
args: &JsonValue,
_on_update: Option<&ToolUpdateSink>,
_on_update: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, focus_core::CoreError> {
let text = args.get_str("text").unwrap_or("(none)").to_string();
Ok(ToolResult::text(format!("echo: {}", text)))

View File

@ -25,7 +25,7 @@ impl Tool for EchoTool {
&self,
_id: &str,
args: &JsonValue,
_on_update: Option<&ToolUpdateSink>,
_on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let text = focus_json::to_string(args);
Ok(ToolResult::text(text))
@ -64,3 +64,131 @@ fn read_effects_compatible() {
assert!(!ToolEffects::NETWORK.compatible_with(ToolEffects::NONE));
assert!(!ToolEffects::PROCESS.compatible_with(ToolEffects::NONE));
}
/// 工具的增量更新on_update必须被 agent 转发为 ToolExecutionUpdate 事件,
/// 而不是等工具结束才一次性出现(回归:此前 agent 把 on_update 传成 None
/// A tool's incremental updates (on_update) must be forwarded by the agent as
/// ToolExecutionUpdate events, not appear only when the tool finishes
/// (regression: the agent used to pass None for on_update).
#[test]
fn tool_updates_stream_to_agent_events() {
use focus_core::event::{AgentEvent, VecSink};
use focus_core::model::ContentBlock;
use focus_core::provider::{
ProviderRequest, StreamEvent, StreamIterator, StreamProvider, StreamResult,
};
use focus_core::{Agent, AgentConfig};
use std::sync::{Arc, Mutex};
/// 在 execute 期间通过 on_update 推送两段增量文本的工具。
/// A tool that pushes two incremental texts via on_update during execute.
#[derive(Debug)]
struct StreamingTool;
impl Tool for StreamingTool {
fn name(&self) -> &str {
"stream"
}
fn description(&self) -> &str {
"streams updates"
}
fn parameters(&self) -> JsonValue {
JsonValue::obj()
}
fn effects(&self) -> ToolEffects {
ToolEffects::READ
}
fn execute(
&self,
_id: &str,
_args: &JsonValue,
on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let cb = on_update.expect("agent must pass on_update");
cb(focus_core::tool::ToolUpdate {
content: vec![ContentBlock::text("line one")],
details: JsonValue::obj(),
});
cb(focus_core::tool::ToolUpdate {
content: vec![ContentBlock::text("line two")],
details: JsonValue::obj(),
});
Ok(ToolResult::text("done"))
}
}
/// 回放固定事件序列的 mock provider第一段模型调用 stream 工具,
/// 第二段给出最终文本。
/// A mock provider replaying fixed event sequences: turn 1 calls the
/// stream tool, turn 2 gives the final text.
#[derive(Debug, Clone, Default)]
struct StaticProvider {
responses: Arc<Mutex<Vec<Vec<StreamEvent>>>>,
}
impl StaticProvider {
fn tool_turn(model: &str) -> Vec<StreamEvent> {
let mut reducer = focus_core::provider::ProviderEventReducer::new(model);
let mut events = reducer.tool_call_start(0, "stream");
events.extend(reducer.finalize_events());
let message = reducer.finish().expect("final message");
events.push(StreamEvent::Done { message });
events
}
fn text_turn(model: &str) -> Vec<StreamEvent> {
let mut reducer = focus_core::provider::ProviderEventReducer::new(model);
let mut events = reducer.text_delta("final text");
events.extend(reducer.finalize_events());
let message = reducer.finish().expect("final message");
events.push(StreamEvent::Done { message });
events
}
}
impl StreamProvider for StaticProvider {
fn stream(&self, _request: &ProviderRequest) -> StreamResult {
let mut queue = self.responses.lock().expect("not poisoned");
if queue.is_empty() {
return Err(focus_core::CoreError::Provider("mock exhausted".into()));
}
let events = queue.remove(0);
Ok(Box::new(Replay { events }))
}
}
struct Replay {
events: Vec<StreamEvent>,
}
impl StreamIterator for Replay {
fn next_event(&mut self) -> Option<StreamEvent> {
if self.events.is_empty() {
None
} else {
Some(self.events.remove(0))
}
}
}
let provider = StaticProvider {
responses: Arc::new(Mutex::new(vec![
StaticProvider::tool_turn("m"),
StaticProvider::text_turn("m"),
])),
};
let config = AgentConfig {
model: "m".into(),
..Default::default()
};
let tools = ToolRegistry::with(Box::new(StreamingTool));
let mut agent = Agent::new(config, Box::new(provider), tools);
let mut sink = VecSink::new();
agent.prompt("go", &mut sink).expect("run failed");
let updates: Vec<&String> = sink
.events
.iter()
.filter_map(|e| match e {
AgentEvent::ToolExecutionUpdate { partial_result, .. } => {
partial_result.content[0].as_text().map(|t| &t.text)
}
_ => None,
})
.collect();
assert_eq!(updates, vec!["line one", "line two"]);
}

View File

@ -50,7 +50,7 @@ impl Tool for EchoTool {
&self,
_id: &str,
args: &JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, CoreError> {
self.calls.lock().unwrap().push(args.clone());
Ok(ToolResult::text(format!(

View File

@ -271,7 +271,7 @@ fn request_body_carries_full_transcript() {
&self,
_id: &str,
_args: &focus_json::JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, focus_core::CoreError> {
Ok(ToolResult::text("ok"))
}

View File

@ -43,7 +43,7 @@ impl focus_core::Tool for NoopTool {
&self,
_id: &str,
_args: &JsonValue,
_u: Option<&focus_core::tool::ToolUpdateSink>,
_u: Option<&focus_core::tool::ToolUpdateSink<'_>>,
) -> Result<ToolResult, CoreError> {
Ok(ToolResult::text("ok"))
}

View File

@ -309,7 +309,7 @@ fn rich_request() -> ProviderRequest {
&self,
_id: &str,
_args: &focus_json::JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, focus_core::CoreError> {
Ok(ToolResult::text("ok"))
}
@ -373,7 +373,7 @@ fn chat_replays_reasoning_content() {
&self,
_id: &str,
_args: &focus_json::JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, focus_core::CoreError> {
Ok(ToolResult::text("ok"))
}
@ -496,7 +496,7 @@ fn chat_replays_both_reasoning_field_names() {
&self,
_id: &str,
_args: &focus_json::JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, focus_core::CoreError> {
Ok(ToolResult::text("ok"))
}
@ -729,7 +729,7 @@ fn agent_multi_turn_replays_reasoning() {
&self,
_id: &str,
_args: &focus_json::JsonValue,
_u: Option<&ToolUpdateSink>,
_u: Option<&ToolUpdateSink<'_>>,
) -> Result<ToolResult, CoreError> {
Ok(ToolResult::text("ok"))
}

View File

@ -84,7 +84,7 @@ impl Tool for EditTool {
&self,
_id: &str,
args: &JsonValue,
_on_update: Option<&ToolUpdateSink>,
_on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let path = args
.get_str("path")

View File

@ -66,7 +66,7 @@ impl Tool for ReadTool {
&self,
_id: &str,
args: &JsonValue,
_on_update: Option<&ToolUpdateSink>,
_on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let path = args
.get_str("path")

View File

@ -81,7 +81,7 @@ impl Tool for ShellTool {
&self,
_id: &str,
args: &JsonValue,
on_update: Option<&ToolUpdateSink>,
on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let command = args
.get_str("command")
@ -189,7 +189,7 @@ enum StreamMsg {
fn run_child(
child: &mut Child,
timeout: Duration,
on_update: Option<&ToolUpdateSink>,
on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<(String, String, i32, bool)> {
let stdout = child
.stdout

View File

@ -59,7 +59,7 @@ impl Tool for WriteTool {
&self,
_id: &str,
args: &JsonValue,
_on_update: Option<&ToolUpdateSink>,
_on_update: Option<&ToolUpdateSink<'_>>,
) -> CoreResult<ToolResult> {
let path = args
.get_str("path")

View File

@ -327,10 +327,15 @@ fn block_lines(block: &UiBlock, width: usize) -> Vec<Line<'static>> {
" ── 结果 ──",
Style::new().fg(Color::DarkGray),
)));
// 结果:部分/全部(在那返回)。
// Result: partial/full (the return).
// 运行中:实时显示最新的输出(尾部);结束后:头部 + 剩余提示。
// Running: show the newest output live (the tail); done:
// head plus a remaining-line hint.
if running {
out.extend(tail_preview(output, width, " ", dim));
} else {
out.extend(preview_lines(output, *full, width, " ", dim));
}
}
if !running && *is_error {
out.push(Line::from(Span::styled(
" ✗ 工具执行失败",
@ -671,3 +676,27 @@ fn preview_lines(
}
out
}
/// 运行中输出的实时预览:只显示最后若干行(最新的内容)。
/// Live preview of running output: only the last few lines (the newest).
fn tail_preview(text: &str, width: usize, indent: &str, style: Style) -> Vec<Line<'static>> {
const TAIL_MAX_LINES: usize = 15;
let lines: Vec<&str> = text.lines().collect();
let tail = lines.len().saturating_sub(TAIL_MAX_LINES);
let mut out: Vec<Line<'static>> = Vec::new();
if tail > 0 {
out.push(Line::from(Span::styled(
format!("{}… (以上 {} 行已折叠,运行中仅显示最新)", indent, tail),
style,
)));
}
for line in lines.into_iter().skip(tail) {
for wrapped in wrap_text(line, width.saturating_sub(2)) {
out.push(Line::from(Span::styled(
format!("{}{}", indent, wrapped),
style,
)));
}
}
out
}