diff --git a/crates/focus-core/src/agent.rs b/crates/focus-core/src/agent.rs index 5dbc1ae..95d47e7 100644 --- a/crates/focus-core/src/agent.rs +++ b/crates/focus-core/src/agent.rs @@ -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(), diff --git a/crates/focus-core/src/event.rs b/crates/focus-core/src/event.rs index 4da2f0d..3c4d558 100644 --- a/crates/focus-core/src/event.rs +++ b/crates/focus-core/src/event.rs @@ -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); diff --git a/crates/focus-core/src/tool.rs b/crates/focus-core/src/tool.rs index ee9c51a..068c7ae 100644 --- a/crates/focus-core/src/tool.rs +++ b/crates/focus-core/src/tool.rs @@ -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; } -/// 工具用来推送增量更新的 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. diff --git a/crates/focus-core/tests/agent_loop.rs b/crates/focus-core/tests/agent_loop.rs index 8a2c017..ce57df2 100644 --- a/crates/focus-core/tests/agent_loop.rs +++ b/crates/focus-core/tests/agent_loop.rs @@ -33,7 +33,7 @@ impl Tool for EchoTool { &self, _id: &str, args: &JsonValue, - _on_update: Option<&ToolUpdateSink>, + _on_update: Option<&ToolUpdateSink<'_>>, ) -> Result { let text = args.get_str("text").unwrap_or("(none)").to_string(); Ok(ToolResult::text(format!("echo: {}", text))) diff --git a/crates/focus-core/tests/tool_tests.rs b/crates/focus-core/tests/tool_tests.rs index cbe6807..e05658e 100644 --- a/crates/focus-core/tests/tool_tests.rs +++ b/crates/focus-core/tests/tool_tests.rs @@ -25,7 +25,7 @@ impl Tool for EchoTool { &self, _id: &str, args: &JsonValue, - _on_update: Option<&ToolUpdateSink>, + _on_update: Option<&ToolUpdateSink<'_>>, ) -> CoreResult { 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 { + 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>>>, + } + impl StaticProvider { + fn tool_turn(model: &str) -> Vec { + 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 { + 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, + } + impl StreamIterator for Replay { + fn next_event(&mut self) -> Option { + 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"]); +} diff --git a/crates/focus-providers/tests/agent_integration.rs b/crates/focus-providers/tests/agent_integration.rs index 6a20599..c59a99d 100644 --- a/crates/focus-providers/tests/agent_integration.rs +++ b/crates/focus-providers/tests/agent_integration.rs @@ -50,7 +50,7 @@ impl Tool for EchoTool { &self, _id: &str, args: &JsonValue, - _u: Option<&ToolUpdateSink>, + _u: Option<&ToolUpdateSink<'_>>, ) -> Result { self.calls.lock().unwrap().push(args.clone()); Ok(ToolResult::text(format!( diff --git a/crates/focus-providers/tests/anthropic_tests.rs b/crates/focus-providers/tests/anthropic_tests.rs index bc735ea..e7297b7 100644 --- a/crates/focus-providers/tests/anthropic_tests.rs +++ b/crates/focus-providers/tests/anthropic_tests.rs @@ -271,7 +271,7 @@ fn request_body_carries_full_transcript() { &self, _id: &str, _args: &focus_json::JsonValue, - _u: Option<&ToolUpdateSink>, + _u: Option<&ToolUpdateSink<'_>>, ) -> Result { Ok(ToolResult::text("ok")) } diff --git a/crates/focus-providers/tests/compaction_integration.rs b/crates/focus-providers/tests/compaction_integration.rs index 9eae6d7..5922461 100644 --- a/crates/focus-providers/tests/compaction_integration.rs +++ b/crates/focus-providers/tests/compaction_integration.rs @@ -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 { Ok(ToolResult::text("ok")) } diff --git a/crates/focus-providers/tests/openai_tests.rs b/crates/focus-providers/tests/openai_tests.rs index 83edd9a..b072cd6 100644 --- a/crates/focus-providers/tests/openai_tests.rs +++ b/crates/focus-providers/tests/openai_tests.rs @@ -309,7 +309,7 @@ fn rich_request() -> ProviderRequest { &self, _id: &str, _args: &focus_json::JsonValue, - _u: Option<&ToolUpdateSink>, + _u: Option<&ToolUpdateSink<'_>>, ) -> Result { 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 { 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 { 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 { Ok(ToolResult::text("ok")) } diff --git a/crates/focus-tools/src/edit.rs b/crates/focus-tools/src/edit.rs index e86ca8f..94441bc 100644 --- a/crates/focus-tools/src/edit.rs +++ b/crates/focus-tools/src/edit.rs @@ -84,7 +84,7 @@ impl Tool for EditTool { &self, _id: &str, args: &JsonValue, - _on_update: Option<&ToolUpdateSink>, + _on_update: Option<&ToolUpdateSink<'_>>, ) -> CoreResult { let path = args .get_str("path") diff --git a/crates/focus-tools/src/read.rs b/crates/focus-tools/src/read.rs index 76952b5..5a0f0c4 100644 --- a/crates/focus-tools/src/read.rs +++ b/crates/focus-tools/src/read.rs @@ -66,7 +66,7 @@ impl Tool for ReadTool { &self, _id: &str, args: &JsonValue, - _on_update: Option<&ToolUpdateSink>, + _on_update: Option<&ToolUpdateSink<'_>>, ) -> CoreResult { let path = args .get_str("path") diff --git a/crates/focus-tools/src/shell.rs b/crates/focus-tools/src/shell.rs index 63d11ed..4507e8b 100644 --- a/crates/focus-tools/src/shell.rs +++ b/crates/focus-tools/src/shell.rs @@ -81,7 +81,7 @@ impl Tool for ShellTool { &self, _id: &str, args: &JsonValue, - on_update: Option<&ToolUpdateSink>, + on_update: Option<&ToolUpdateSink<'_>>, ) -> CoreResult { 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 diff --git a/crates/focus-tools/src/write.rs b/crates/focus-tools/src/write.rs index 8f221f9..babcbd1 100644 --- a/crates/focus-tools/src/write.rs +++ b/crates/focus-tools/src/write.rs @@ -59,7 +59,7 @@ impl Tool for WriteTool { &self, _id: &str, args: &JsonValue, - _on_update: Option<&ToolUpdateSink>, + _on_update: Option<&ToolUpdateSink<'_>>, ) -> CoreResult { let path = args .get_str("path") diff --git a/crates/focus-tui/src/ui.rs b/crates/focus-tui/src/ui.rs index 4e37475..80fdecd 100644 --- a/crates/focus-tui/src/ui.rs +++ b/crates/focus-tui/src/ui.rs @@ -327,9 +327,14 @@ fn block_lines(block: &UiBlock, width: usize) -> Vec> { " ── 结果 ──", Style::new().fg(Color::DarkGray), ))); - // 结果:部分/全部(在那返回)。 - // Result: partial/full (the return). - out.extend(preview_lines(output, *full, width, " ", dim)); + // 运行中:实时显示最新的输出(尾部);结束后:头部 + 剩余提示。 + // 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> { + 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> = 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 +}