From f6447783a327b828c67f6aae5ba1a6054187470d Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Mon, 31 Aug 2026 14:48:01 +0800 Subject: [PATCH] fix: tui tool render fix --- crates/client/src/client_core.rs | 39 ++ crates/client/src/stdio.rs | 20 + crates/server/src/runtime/connection.rs | 8 + crates/server/src/runtime/outbound.rs | 7 +- .../src/runtime/turn_exec/event_stream.rs | 36 +- .../server/tests/tool_call_param_refresh.rs | 373 +++++++++++++ crates/tui/README.md | 7 + crates/tui/src/chatwidget.rs | 4 + crates/tui/src/chatwidget/history_commit.rs | 31 +- crates/tui/src/chatwidget/session_history.rs | 1 + crates/tui/src/chatwidget/transcript_sync.rs | 23 +- crates/tui/src/chatwidget/transcript_view.rs | 67 ++- crates/tui/src/chatwidget/worker_events.rs | 66 ++- crates/tui/src/chatwidget_tests.rs | 266 +++++++-- crates/tui/src/insert_history.rs | 91 ++- crates/tui/src/interactive.rs | 81 ++- crates/tui/src/transcript/lifecycle.rs | 16 +- crates/tui/src/transcript/model.rs | 12 +- crates/tui/src/transcript/presentation.rs | 22 +- crates/tui/src/transcript/projector.rs | 519 +++++++++++++++++- crates/tui/src/transcript/render.rs | 63 ++- crates/tui/src/transcript/tool_state.rs | 245 +++++++++ crates/tui/src/tui.rs | 228 ++++++-- crates/tui/src/worker.rs | 255 ++++++++- crates/tui/src/worker/native_items.rs | 104 ++++ 25 files changed, 2397 insertions(+), 187 deletions(-) create mode 100644 crates/server/tests/tool_call_param_refresh.rs diff --git a/crates/client/src/client_core.rs b/crates/client/src/client_core.rs index 65b27330..5a85d9c7 100644 --- a/crates/client/src/client_core.rs +++ b/crates/client/src/client_core.rs @@ -505,6 +505,45 @@ impl ServerClientCore { .await } + pub(crate) async fn turn_read_native( + &mut self, + session_id: SessionId, + turn_id: TurnId, + ) -> Result { + self.request( + "turn/read", + devo_protocol::native::rpc_turn::TurnReadParams { + session_id: devo_protocol::native::ids::SessionId::from_string( + session_id.to_string(), + ), + turn_id: devo_protocol::native::ids::TurnId::from_string(turn_id.to_string()), + }, + ) + .await + } + + pub(crate) async fn turn_items_list_native( + &mut self, + session_id: SessionId, + turn_id: TurnId, + cursor: Option, + limit: Option, + ) -> Result> { + self.request( + "session/items/list", + devo_protocol::native::rpc_session::SessionItemsListParams { + session_id: devo_protocol::native::ids::SessionId::from_string( + session_id.to_string(), + ), + turn_id: Some(devo_protocol::native::ids::TurnId::from_string( + turn_id.to_string(), + )), + page: devo_protocol::native::page::PageParams { cursor, limit }, + }, + ) + .await + } + /// Native `session/fork` (L2-DES-APP-008 Phase B). pub(crate) async fn session_fork_native( &mut self, diff --git a/crates/client/src/stdio.rs b/crates/client/src/stdio.rs index 94e943b7..48abfc47 100644 --- a/crates/client/src/stdio.rs +++ b/crates/client/src/stdio.rs @@ -221,6 +221,26 @@ impl StdioServerClient { .await } + pub async fn turn_read_native( + &mut self, + session_id: SessionId, + turn_id: TurnId, + ) -> Result { + self.core.turn_read_native(session_id, turn_id).await + } + + pub async fn turn_items_list_native( + &mut self, + session_id: SessionId, + turn_id: TurnId, + cursor: Option, + limit: Option, + ) -> Result> { + self.core + .turn_items_list_native(session_id, turn_id, cursor, limit) + .await + } + /// Native `session/compact/start`; see /// `client_core::session_compact_start_native`. pub async fn session_compact_start_native( diff --git a/crates/server/src/runtime/connection.rs b/crates/server/src/runtime/connection.rs index fef4cf54..18923cf9 100644 --- a/crates/server/src/runtime/connection.rs +++ b/crates/server/src/runtime/connection.rs @@ -1129,6 +1129,14 @@ async fn remove_pending_client_request( fn outbound_delivery_policy(event: &ServerEvent) -> OutboundDeliveryPolicy { match event { + // Tool-call argument deltas are low-volume and drive the client's + // running-row parameter display; losing one permanently truncates the + // accumulated JSON until the item refresh, so they ride the reliable + // lane unlike the high-volume text/output deltas below. + ServerEvent::ItemDelta { + delta_kind: ItemDeltaKind::ToolCallInputDelta, + .. + } => OutboundDeliveryPolicy::Reliable, ServerEvent::ItemDelta { .. } | ServerEvent::TurnUsageUpdated(_) | ServerEvent::ContextUsageUpdated(_) diff --git a/crates/server/src/runtime/outbound.rs b/crates/server/src/runtime/outbound.rs index fad4ad7d..d005bbe7 100644 --- a/crates/server/src/runtime/outbound.rs +++ b/crates/server/src/runtime/outbound.rs @@ -12,8 +12,11 @@ pub(crate) const OUTBOUND_RELIABLE_RESERVED_CAPACITY: usize = 64; pub(crate) const OUTBOUND_BACKPRESSURE_LOG_THRESHOLD: Duration = Duration::from_millis(50); /// Max time streaming notifications wait for outbound capacity before being /// dropped. Event streams must not park forever on a slow client: parent+child -/// turns share one connection and can fill the queue quickly. -pub(crate) const OUTBOUND_NOTIFICATION_MAX_WAIT: Duration = Duration::from_millis(200); +/// turns share one connection and can fill the queue quickly. Lifecycle +/// traffic (`item/completed`, `turn/completed`, …) rides this wait too — +/// dropping it wedges client render state — so the budget is sized for bursty +/// slow consumers (Windows console pipes) rather than round-trip latency. +pub(crate) const OUTBOUND_NOTIFICATION_MAX_WAIT: Duration = Duration::from_secs(2); #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum OutboundDeliveryPolicy { diff --git a/crates/server/src/runtime/turn_exec/event_stream.rs b/crates/server/src/runtime/turn_exec/event_stream.rs index 0b891516..36d42e1c 100644 --- a/crates/server/src/runtime/turn_exec/event_stream.rs +++ b/crates/server/src/runtime/turn_exec/event_stream.rs @@ -525,9 +525,43 @@ async fn handle_tool_use_start( event_tool_registry: &Arc, ) { tool_names_by_id.insert(id.clone(), name.clone()); - if let Some(pending) = pending_tool_calls.get_mut(&id) { + if let Some(mut pending) = pending_tool_calls.remove(&id) { + let input_is_empty = |value: &serde_json::Value| { + value.is_null() || matches!(value, serde_json::Value::Object(map) if map.is_empty()) + }; + let previously_empty_input = input_is_empty(&pending.input); pending.input = input.clone(); pending.command = command_display_from_input(&name, &input); + // The first `item/started` for a streamed tool call carries empty + // parameters (the provider streams arguments afterwards). When the + // assembled turn delivers the complete input, re-broadcast the same + // item so live clients can render the running row's parameters — + // the input-delta channel alone is best-effort and only parses once + // the full JSON accumulates. + if previously_empty_input + && !input_is_empty(&pending.input) + && let (Some(item_id), Some(item_seq)) = (pending.item_id, pending.item_seq) + { + let start_item = tool_start_item_from_input( + &id, + &name, + &pending.command, + &pending.input, + pending.display_kind, + event_tool_registry.preparation_feedback(&name), + ); + runtime + .emit_item_started( + session_id, + turn_id, + item_id, + Some(item_seq), + start_item.item_kind, + start_item.payload, + ) + .await; + } + pending_tool_calls.insert(id, pending); return; } if let (Some(item_id), Some(item_seq)) = (reasoning_item_id.take(), reasoning_item_seq.take()) { diff --git a/crates/server/tests/tool_call_param_refresh.rs b/crates/server/tests/tool_call_param_refresh.rs new file mode 100644 index 00000000..4f810f58 --- /dev/null +++ b/crates/server/tests/tool_call_param_refresh.rs @@ -0,0 +1,373 @@ +//! Audits the live tool-call parameter refresh: a streamed tool call starts +//! with empty parameters (`item/started`), and when the assembled model turn +//! delivers the complete input the server must re-broadcast `item/started` +//! for the same item so native clients can render the running row's command. +//! Without the refresh the parameters only appear at completion. + +use std::path::Path; +use std::pin::Pin; +use std::sync::Arc; + +use anyhow::Context; +use anyhow::Result; +use async_trait::async_trait; +use devo_core::AppConfigStore; +use devo_core::ProviderVendorCatalog; +use futures::stream; +use pretty_assertions::assert_eq; +use serde_json::json; +use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +use devo_core::BundledSkillsConfig; +use devo_core::FileSystemSkillCatalog; +use devo_core::PresetModelCatalog; +use devo_core::SkillsConfig; +use devo_core::tools::ToolCallError; +use devo_core::tools::ToolResult; +use devo_core::tools::ToolResultContent; +use devo_core::tools::json_schema::JsonSchema; +use devo_core::tools::registry::ToolRegistryBuilder; +use devo_core::tools::tool_handler::ToolHandler; +use devo_core::tools::tool_spec::ToolExecutionMode; +use devo_core::tools::tool_spec::ToolOutputMode; +use devo_core::tools::tool_spec::ToolSpec; +use devo_protocol::ModelRequest; +use devo_protocol::ModelResponse; +use devo_protocol::ResponseContent; +use devo_protocol::ResponseMetadata; +use devo_protocol::StopReason; +use devo_protocol::StreamEvent; +use devo_protocol::Usage; +use devo_provider::ModelProviderSDK; +use devo_provider::SingleProviderRouter; +use devo_server::ClientTransportKind; +use devo_server::ServerRuntime; +use devo_server::ServerRuntimeDependencies; + +const TOOL_COMMAND: &str = "cargo test -p devo-server"; + +/// Streams a tool call the way real providers do: `ToolCallStart` with empty +/// input, the arguments via `ToolCallInputDelta`, then the assembled response +/// (whose ToolUse input is still empty — the merged arguments come from the +/// delta accumulation). +#[derive(Default)] +struct StreamedToolProvider { + requests: std::sync::atomic::AtomicUsize, +} + +#[async_trait] +impl ModelProviderSDK for StreamedToolProvider { + async fn completion(&self, _request: ModelRequest) -> Result { + anyhow::bail!("test provider does not support completion") + } + + async fn completion_stream( + &self, + _request: ModelRequest, + ) -> Result> + Send>>> { + let request_number = self + .requests + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let events = if request_number == 0 { + let tool_input = json!({ "command": TOOL_COMMAND }); + vec![ + Ok(StreamEvent::ToolCallStart { + index: 0, + id: "bash-1".to_string(), + name: "bash".to_string(), + input: json!({}), + }), + Ok(StreamEvent::ToolCallInputDelta { + index: 0, + partial_json: tool_input.to_string(), + }), + Ok(StreamEvent::MessageDone { + response: ModelResponse { + id: "resp-tools".to_string(), + content: vec![ResponseContent::ToolUse { + id: "bash-1".to_string(), + name: "bash".to_string(), + input: json!({}), + }], + stop_reason: Some(StopReason::ToolUse), + usage: Usage::default(), + metadata: ResponseMetadata::default(), + }, + }), + ] + } else { + vec![ + Ok(StreamEvent::TextDelta { + index: 0, + text: "done".to_string(), + }), + Ok(StreamEvent::MessageDone { + response: ModelResponse { + id: "resp-done".to_string(), + content: vec![ResponseContent::Text("done".to_string())], + stop_reason: Some(StopReason::EndTurn), + usage: Usage::default(), + metadata: ResponseMetadata::default(), + }, + }), + ] + }; + Ok(Box::pin(stream::iter(events))) + } + + fn name(&self) -> &str { + "streamed-tool-test-provider" + } +} + +struct EchoTool; + +#[async_trait] +impl ToolHandler for EchoTool { + fn spec(&self) -> &ToolSpec { + Box::leak(Box::new(ToolSpec { + name: "bash".into(), + description: "Returns its input as output.".into(), + input_schema: JsonSchema::object(Default::default(), None, None), + output_mode: ToolOutputMode::Text, + execution_mode: ToolExecutionMode::ReadOnly, + capability_tags: vec![], + supports_parallel: true, + preparation_feedback: devo_core::tools::ToolPreparationFeedback::None, + display_name: None, + supports_cancellation: None, + supports_streaming: None, + })) + } + + async fn handle( + &self, + _ctx: devo_core::tools::ToolContext, + _input: serde_json::Value, + _progress: Option, + ) -> std::result::Result { + Ok(ToolResult::success( + ToolResultContent::Text("ok".into()), + "ok", + )) + } +} + +fn build_runtime(data_root: &Path) -> Arc { + let provider: Arc = Arc::new(StreamedToolProvider::default()); + let mut builder = ToolRegistryBuilder::new(); + builder.register_handler("bash", Arc::new(EchoTool)); + builder.push_spec(ToolSpec { + name: "bash".into(), + description: "Returns its input as output.".into(), + input_schema: JsonSchema::object(Default::default(), None, None), + output_mode: ToolOutputMode::Text, + execution_mode: ToolExecutionMode::ReadOnly, + capability_tags: vec![], + supports_parallel: true, + preparation_feedback: devo_core::tools::ToolPreparationFeedback::None, + display_name: None, + supports_cancellation: None, + supports_streaming: None, + }); + let db_path = data_root.join("test_tool_param_refresh.db"); + let db = Arc::new(devo_server::db::Database::open(db_path).expect("open test database")); + ServerRuntime::new( + data_root.to_path_buf(), + ServerRuntimeDependencies::new( + Arc::clone(&provider), + Arc::new(SingleProviderRouter::new(provider)), + Arc::new(builder.build()), + devo_server::empty_mcp_manager(), + "test-model".to_string(), + Arc::new(PresetModelCatalog::default()), + Arc::new(ProviderVendorCatalog::default()), + Box::new(FileSystemSkillCatalog::new(SkillsConfig { + enabled: false, + user_roots: Vec::new(), + workspace_roots: Vec::new(), + watch_for_changes: false, + bundled: Some(BundledSkillsConfig { enabled: false }), + include_instructions: Some(false), + config: Vec::new(), + })), + devo_core::AgentsMdConfig::default(), + db, + Arc::new(std::sync::Mutex::new( + AppConfigStore::load(data_root.to_path_buf(), None).expect("load app config store"), + )), + ), + ) +} + +fn tool_call_started_payload(value: &serde_json::Value) -> Option<&serde_json::Value> { + if value.get("method").and_then(serde_json::Value::as_str) != Some("item/started") { + return None; + } + let item = value.get("params")?.get("item")?; + if item + .get("item")? + .get("type") + .and_then(serde_json::Value::as_str) + != Some("toolCall") + { + return None; + } + Some(item) +} + +#[tokio::test] +async fn streamed_tool_call_rebroadcasts_started_with_complete_parameters() -> Result<()> { + let temp_dir = TempDir::new()?; + let workspace_root = temp_dir.path().join("workspace"); + std::fs::create_dir_all(&workspace_root)?; + + let runtime = build_runtime(temp_dir.path()); + let (notifications_tx, mut notifications_rx) = devo_server::test_outbound_channel(4096); + let connection_id = runtime + .register_connection(ClientTransportKind::Stdio, notifications_tx) + .await; + + let initialize_response = runtime + .handle_incoming( + connection_id, + json!({ + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": {}, + "_meta": { "devo": { "protocol": "native", "typedItems": true } }, + "clientInfo": { "name": "test", "title": "test", "version": "1.0.0" } + } + }), + ) + .await + .context("initialize response")?; + assert!( + initialize_response.get("error").is_none(), + "initialize failed: {initialize_response}" + ); + + let session_response = runtime + .handle_incoming( + connection_id, + json!({ + "id": 2, + "method": "session/new", + "params": { + "cwd": workspace_root, + "idempotencyKey": "tool-param-refresh-session" + } + }), + ) + .await + .context("session/new response")?; + assert!( + session_response.get("error").is_none(), + "session/new failed: {session_response}" + ); + let session_id = session_response["result"]["session"]["id"] + .as_str() + .context("session id in session/new response")? + .to_string(); + + let subscription_response = runtime + .handle_incoming( + connection_id, + json!({ + "id": 3, + "method": "subscription/create", + "params": { + "selectors": [{ "kind": "session", "sessionId": session_id }], + "includeSnapshot": false + } + }), + ) + .await + .context("subscription/create response")?; + assert!( + subscription_response.get("error").is_none(), + "subscription/create failed: {subscription_response}" + ); + + let turn_response = runtime + .handle_incoming( + connection_id, + json!({ + "id": 4, + "method": "turn/start", + "params": { + "sessionId": session_id, + "input": [{ "type": "text", "text": "Run the tool." }], + "idempotencyKey": format!("native-test-turn-{}", uuid::Uuid::new_v4()), + "model": null, + "thinking": null, + "sandbox": null, + "approval_policy": null, + "cwd": null + } + }), + ) + .await + .context("turn/start response")?; + assert!( + turn_response.get("error").is_none(), + "turn/start failed: {turn_response}" + ); + + let mut tool_call_started = Vec::new(); + let deadline = Duration::from_secs(30); + let start = std::time::Instant::now(); + while start.elapsed() < deadline { + let Ok(Some(value)) = timeout(Duration::from_secs(10), notifications_rx.recv()).await + else { + break; + }; + if tool_call_started_payload(&value).is_some() { + tool_call_started.push(value.clone()); + } + let is_turn_completed = + value.get("method").and_then(serde_json::Value::as_str) == Some("turn/completed"); + if is_turn_completed { + break; + } + } + + assert_eq!( + tool_call_started.len(), + 2, + "expected exactly two item/started notifications for the streamed tool call, got: {tool_call_started:?}" + ); + + let first = tool_call_started[0]["params"]["item"].clone(); + let second = tool_call_started[1]["params"]["item"].clone(); + assert_eq!( + first["id"], second["id"], + "refresh must reuse the original item id" + ); + assert_eq!(first["seq"], second["seq"]); + + let first_input = first["item"]["input"].clone(); + let first_input_empty = first_input.is_null() + || matches!(&first_input, serde_json::Value::Object(map) if map.is_empty()); + assert!( + first_input_empty, + "first item/started should carry empty streamed parameters: {first}" + ); + + assert_eq!( + second["item"]["input"]["command"], + json!(TOOL_COMMAND), + "refreshed item/started must carry the complete parameters: {second}" + ); + assert_eq!( + second["state"], + json!("running"), + "refresh keeps the item in the running state: {second}" + ); + + Ok(()) +} diff --git a/crates/tui/README.md b/crates/tui/README.md index 3d862a37..86b86a76 100644 --- a/crates/tui/README.md +++ b/crates/tui/README.md @@ -365,6 +365,13 @@ The bottom pane is the user-facing input area. It contains: ### Execution Display — `exec_cell/` +Tool rows have one presentation owner per `tool_use_id`. A shell row stays in +the live viewport for the whole turn: completion changes its title in place +from `Running` to `Ran`, and the turn boundary commits exactly one durable +`Ran` row to scrollback. Missing authoritative results use the explicit +`result unavailable` degraded state; ordinary viewport flushes never invent a +successful or failed result. + | Module | Purpose | |--------|---------| | `model.rs` | `ExecCell` data model: command line, output state, exit status. | diff --git a/crates/tui/src/chatwidget.rs b/crates/tui/src/chatwidget.rs index c139933e..6c1cafb3 100644 --- a/crates/tui/src/chatwidget.rs +++ b/crates/tui/src/chatwidget.rs @@ -212,7 +212,9 @@ struct ActiveToolCall { output: String, parsed_commands: Vec, exec_like: bool, + owned_by_active_cell: bool, start_time: Option, + phase: crate::transcript::model::ToolPhase, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -256,6 +258,7 @@ pub(crate) struct ChatWidget { active_cell_revision: u64, last_terminal_assistant_visible_hash: Option<(String, u64)>, active_tool_calls: HashMap, + detached_exec_tool_ids: HashSet, pending_tool_calls: Vec, history: Vec>, next_history_flush_index: usize, @@ -535,6 +538,7 @@ impl ChatWidget { active_cell_revision: 0, last_terminal_assistant_visible_hash: None, active_tool_calls: HashMap::new(), + detached_exec_tool_ids: HashSet::new(), pending_tool_calls: Vec::new(), history, next_history_flush_index: 0, diff --git a/crates/tui/src/chatwidget/history_commit.rs b/crates/tui/src/chatwidget/history_commit.rs index 8f4f2446..a45ff236 100644 --- a/crates/tui/src/chatwidget/history_commit.rs +++ b/crates/tui/src/chatwidget/history_commit.rs @@ -25,6 +25,10 @@ pub(crate) fn is_exploration_tool(tool: &ToolCellModel) -> bool { && !matches!(tool.command_source, Some(ExecCommandSource::UserShell)) } +pub(crate) fn tool_uses_exec_cell(tool: &ToolCellModel) -> bool { + is_exploration_tool(tool) || matches!(tool.command_source, Some(ExecCommandSource::UserShell)) +} + /// Where a finished tool row should land in the transcript. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum ToolCommitTarget { @@ -139,16 +143,22 @@ impl ChatWidget { fn commit_exec_tool(&mut self, tool: ToolCellModel, target: ToolCommitTarget) { if self.complete_exec_tool_from_committed(&tool) { - return; - } - - if target == ToolCommitTarget::LiveOverlay { + // The live overlay or an existing history cell already owns this + // call; it was completed in place. return; } let exec = exec_cell_from_tool(&tool, &self.session.cwd); - self.add_history_entry_without_redraw(Box::new(exec)); - self.apply_tool_io_to_history_exec(&tool); + match target { + ToolCommitTarget::LiveOverlay => { + self.active_cell = Some(Box::new(exec)); + self.apply_tool_io_to_active_exec(&tool); + } + ToolCommitTarget::ScrollbackHistory => { + self.add_history_entry_without_redraw(Box::new(exec)); + self.apply_tool_io_to_history_exec(&tool); + } + } } fn commit_exploration_tool(&mut self, tool: ToolCellModel, target: ToolCommitTarget) { @@ -274,11 +284,16 @@ impl ChatWidget { let Some(cell) = self .history - .last_mut() + .get_mut(self.next_history_flush_index..) + .and_then(|history| history.last_mut()) .and_then(|cell| cell.as_any_mut().downcast_mut::()) else { return false; }; + if cell.contains_call(&call_id) { + self.apply_tool_io_to_history_exec(tool); + return true; + } let Some(grouped) = cell.with_added_call( call_id, command_tokens, @@ -364,6 +379,7 @@ impl ChatWidget { for cell in self .history .iter_mut() + .skip(self.next_history_flush_index) .rev() .filter_map(|cell| cell.as_any_mut().downcast_mut::()) { @@ -423,6 +439,7 @@ impl ChatWidget { for cell in self .history .iter_mut() + .skip(self.next_history_flush_index) .rev() .filter_map(|cell| cell.as_any_mut().downcast_mut::()) { diff --git a/crates/tui/src/chatwidget/session_history.rs b/crates/tui/src/chatwidget/session_history.rs index a8eb6b08..3a658c34 100644 --- a/crates/tui/src/chatwidget/session_history.rs +++ b/crates/tui/src/chatwidget/session_history.rs @@ -42,6 +42,7 @@ impl ChatWidget { self.active_proposed_plan = None; self.pending_proposed_plan_actions = false; self.active_tool_calls.clear(); + self.detached_exec_tool_ids.clear(); self.pending_tool_calls.clear(); self.active_text_items.clear(); self.boundary_committed_assistant_items.clear(); diff --git a/crates/tui/src/chatwidget/transcript_sync.rs b/crates/tui/src/chatwidget/transcript_sync.rs index c63892d5..83d97a38 100644 --- a/crates/tui/src/chatwidget/transcript_sync.rs +++ b/crates/tui/src/chatwidget/transcript_sync.rs @@ -8,6 +8,7 @@ use ratatui::text::Line; use crate::events::TextItemKind; use crate::events::WorkerEvent; use crate::transcript::lifecycle::ItemLifecycleEvent; +use crate::transcript::lifecycle::TurnToolOutcome; use crate::transcript::model::CommittedCellModel; use crate::transcript::model::ToolPhase; @@ -32,8 +33,10 @@ impl ChatWidget { false } - pub(super) fn clear_turn_live_projection(&mut self) { - self.apply_item_lifecycle(ItemLifecycleEvent::TurnLiveToolsCleared); + pub(super) fn clear_turn_live_projection(&mut self, outcome: TurnToolOutcome) { + self.active_cell = None; + self.detached_exec_tool_ids.clear(); + self.apply_item_lifecycle(ItemLifecycleEvent::TurnLiveToolsCleared { outcome }); } pub(super) fn sync_transcript_projection(&mut self) { @@ -67,7 +70,7 @@ impl ChatWidget { self.add_markdown_history_without_redraw(title, &text.text); } CommittedCellModel::Tool(tool) => { - self.commit_committed_tool_to_live_turn(tool); + self.commit_committed_tool_to_history(tool); } } } @@ -91,7 +94,10 @@ impl ChatWidget { output: tool.output_preview.clone(), parsed_commands: tool.parsed_commands.clone(), exec_like: tool.exec_like, + owned_by_active_cell: crate::chatwidget::history_commit::tool_uses_exec_cell(tool) + && !self.detached_exec_tool_ids.contains(&tool.tool_use_id), start_time: tool.start_time, + phase: tool.phase, }; if tool.phase == ToolPhase::Preparing { self.pending_tool_calls.push(tool_call); @@ -136,7 +142,16 @@ impl ChatWidget { .iter() .any(|item| item.item_id == item_id) { - self.flush_active_cell(); + if let Some(cell) = self + .active_cell + .as_ref() + .and_then(|cell| cell.as_any().downcast_ref::()) + .filter(|cell| cell.is_exploring_cell()) + { + self.detached_exec_tool_ids + .extend(cell.iter_calls().map(|call| call.call_id.clone())); + self.active_cell = None; + } self.start_text_item(item_id, live.kind); } diff --git a/crates/tui/src/chatwidget/transcript_view.rs b/crates/tui/src/chatwidget/transcript_view.rs index 76c7220e..1a2425a9 100644 --- a/crates/tui/src/chatwidget/transcript_view.rs +++ b/crates/tui/src/chatwidget/transcript_view.rs @@ -45,6 +45,7 @@ enum LiveViewportLineMode { #[allow(clippy::large_enum_variant)] enum LiveItem { + ActiveCell, Text(usize), Tool(String), } @@ -180,18 +181,24 @@ impl ChatWidget { LiveViewportLineMode::Display => cell.display_lines(width), LiveViewportLineMode::Transcript => cell.transcript_lines(width), }; - if let Some(cell) = &self.active_cell { - Self::extend_lines_with_separator(&mut lines, cell_lines(cell.as_ref())); - } - let mut items: Vec<(u64, LiveItem)> = Vec::new(); + if self.active_cell.is_some() { + let seq = self + .active_tool_calls + .values() + .filter(|tool| tool.owned_by_active_cell) + .map(|tool| tool.seq) + .min() + .unwrap_or(0); + items.push((seq, LiveItem::ActiveCell)); + } for (idx, item) in self.active_text_items.iter().enumerate() { if item.cell.is_some() { items.push((item.seq, LiveItem::Text(idx))); } } for tool_call in self.active_tool_calls.values() { - if tool_call.exec_like { + if tool_call.owned_by_active_cell { continue; } items.push((tool_call.seq, LiveItem::Tool(tool_call.tool_use_id.clone()))); @@ -208,19 +215,41 @@ impl ChatWidget { for (_, item) in items { match item { + LiveItem::ActiveCell => { + if let Some(cell) = &self.active_cell { + Self::extend_lines_with_separator(&mut lines, cell_lines(cell.as_ref())); + } + } LiveItem::Text(idx) => { if let Some(cell) = &self.active_text_items[idx].cell { Self::extend_lines_with_separator(&mut lines, cell_lines(cell.as_ref())); } } LiveItem::Tool(tool_use_id) => { - if let Some(tool_call) = self.active_tool_calls.get(&tool_use_id) { + if let Some(tool) = self.transcript_projector.live_tool(&tool_use_id) { + let dot_prefix = if tool.is_error { + Self::failed_dot_prefix() + } else { + Self::tool_dot_prefix() + }; let tool_lines = match mode { LiveViewportLineMode::Display => { - Self::live_tool_display_lines(width, tool_call) + crate::transcript::render::live_tool_display_lines( + tool, + width, + &self.session.cwd, + dot_prefix, + Self::tool_text_style(), + ) } LiveViewportLineMode::Transcript => { - Self::live_tool_transcript_lines(width, tool_call) + crate::transcript::render::live_tool_transcript_lines( + tool, + width, + &self.session.cwd, + dot_prefix, + Self::tool_text_style(), + ) } }; Self::extend_lines_with_separator(&mut lines, tool_lines); @@ -277,7 +306,7 @@ impl ChatWidget { (Some(tool_name), Some(input)) if is_agent_task_tool_name(tool_name) => { AgentToolCell::new( tool_name.clone(), - ToolPhase::Running, + tool_call.phase, Some(input.clone()), None, tool_call.output.clone(), @@ -287,9 +316,9 @@ impl ChatWidget { } (Some(tool_name), Some(input)) => { let title_line = tool_title_line( - ToolPhase::Running, + tool_call.phase, &tool_title_parts( - ToolPhase::Running, + tool_call.phase, Some(tool_name.as_str()), Some(input), &tool_call.parsed_commands, @@ -313,9 +342,9 @@ impl ChatWidget { } _ => { let title_line = tool_title_line( - ToolPhase::Running, + tool_call.phase, &tool_title_parts( - ToolPhase::Running, + tool_call.phase, tool_call.tool_name.as_deref(), tool_call.input.as_ref(), &tool_call.parsed_commands, @@ -344,7 +373,7 @@ impl ChatWidget { (Some(tool_name), Some(input)) if is_agent_task_tool_name(tool_name) => { AgentToolCell::new( tool_name.clone(), - ToolPhase::Running, + tool_call.phase, Some(input.clone()), None, tool_call.output.clone(), @@ -354,9 +383,9 @@ impl ChatWidget { } (Some(tool_name), Some(input)) => { let title_line = tool_title_line( - ToolPhase::Running, + tool_call.phase, &tool_title_parts( - ToolPhase::Running, + tool_call.phase, Some(tool_name.as_str()), Some(input), &tool_call.parsed_commands, @@ -380,9 +409,9 @@ impl ChatWidget { } _ => { let title_line = tool_title_line( - ToolPhase::Running, + tool_call.phase, &tool_title_parts( - ToolPhase::Running, + tool_call.phase, tool_call.tool_name.as_deref(), tool_call.input.as_ref(), &tool_call.parsed_commands, @@ -422,7 +451,7 @@ impl ChatWidget { let text_kind = |item: &LiveItem| match item { LiveItem::Text(idx) => active_text_items.get(*idx).map(|item| item.kind), - LiveItem::Tool(_) => None, + LiveItem::ActiveCell | LiveItem::Tool(_) => None, }; if let (Some(kind_a), Some(kind_b)) = (text_kind(item_a), text_kind(item_b)) { if Self::text_item_precedes_assistant(kind_a) && kind_b == TextItemKind::Assistant { diff --git a/crates/tui/src/chatwidget/worker_events.rs b/crates/tui/src/chatwidget/worker_events.rs index f78047d9..21416e87 100644 --- a/crates/tui/src/chatwidget/worker_events.rs +++ b/crates/tui/src/chatwidget/worker_events.rs @@ -22,6 +22,8 @@ use crate::exec_cell::CommandOutput; use crate::exec_cell::ExecCell; use crate::exec_cell::new_active_exec_command; use crate::history_cell; +use crate::transcript::lifecycle::TurnToolOutcome; +use crate::transcript::model::ToolPhase; use super::ActiveToolCall; use super::ChatWidget; @@ -127,7 +129,9 @@ impl ChatWidget { output: String::new(), parsed_commands: parsed.clone(), exec_like: true, + owned_by_active_cell: true, start_time: None, + phase: ToolPhase::Running, }, ); self.active_cell_revision = self.active_cell_revision.wrapping_add(1); @@ -157,7 +161,9 @@ impl ChatWidget { output: String::new(), parsed_commands, exec_like: true, + owned_by_active_cell: true, start_time: None, + phase: ToolPhase::Running, }, ); self.active_cell_revision = self.active_cell_revision.wrapping_add(1); @@ -220,13 +226,16 @@ impl ChatWidget { self.frame_requester.schedule_frame(); return true; } - self.history.iter_mut().rev().any(|cell| { - cell.as_any_mut() - .downcast_mut::() - .is_some_and(|cell| { - cell.update_call(tool_use_id, command_parts.clone(), parsed.clone()) - }) - }) + self.history[self.next_history_flush_index..] + .iter_mut() + .rev() + .any(|cell| { + cell.as_any_mut() + .downcast_mut::() + .is_some_and(|cell| { + cell.update_call(tool_use_id, command_parts.clone(), parsed.clone()) + }) + }) } pub(super) fn complete_exec_tool_from_committed( @@ -277,15 +286,8 @@ impl ChatWidget { .and_then(|cell| cell.as_any_mut().downcast_mut::()) && cell.complete_call(tool_use_id, command_output.clone(), duration) { - if cell.is_exploring_cell() { - self.active_cell_revision = self.active_cell_revision.wrapping_add(1); - self.frame_requester.schedule_frame(); - } else if cell.should_flush() { - self.flush_active_cell(); - } else { - self.active_cell_revision = self.active_cell_revision.wrapping_add(1); - self.frame_requester.schedule_frame(); - } + self.active_cell_revision = self.active_cell_revision.wrapping_add(1); + self.frame_requester.schedule_frame(); self.set_status_message(if tool.is_error { "Tool returned an error" } else { @@ -297,6 +299,7 @@ impl ChatWidget { for cell in self .history .iter_mut() + .skip(self.next_history_flush_index) .rev() .filter_map(|cell| cell.as_any_mut().downcast_mut::()) { @@ -315,7 +318,10 @@ impl ChatWidget { let exec_tools: Vec<_> = self .transcript_projector .live_tools() - .filter(|tool| tool.exec_like) + .filter(|tool| { + crate::chatwidget::history_commit::tool_uses_exec_cell(tool) + && !self.detached_exec_tool_ids.contains(&tool.tool_use_id) + }) .cloned() .collect(); for tool in exec_tools { @@ -369,6 +375,9 @@ impl ChatWidget { self.active_cell_revision = self.active_cell_revision.wrapping_add(1); } } + if tool.phase.is_terminal() { + let _ = self.complete_exec_tool_from_committed(&tool); + } } } @@ -401,6 +410,7 @@ impl ChatWidget { self.refresh_header_box(); self.busy = true; self.active_text_items.clear(); + self.detached_exec_tool_ids.clear(); self.active_proposed_plan = None; self.bottom_pane.set_task_running(true); } @@ -482,6 +492,17 @@ impl ChatWidget { WorkerEvent::ShellCommandFinished { exit_code } => { let standalone_shell = self.active_turn_id.is_none(); let interrupted = exit_code.is_none(); + if standalone_shell { + // A standalone shell command runs outside any agent turn, + // so this event is its commit boundary: flush the live + // exec cell into scrollback before the summary row lands. + let outcome = if interrupted { + TurnToolOutcome::Interrupted + } else { + TurnToolOutcome::Completed + }; + self.clear_turn_live_projection(outcome); + } let accent_color = self.active_accent_color(); let cell = if interrupted { history_cell::TurnSummaryCell::new_interrupted( @@ -694,7 +715,14 @@ impl ChatWidget { self.commit_active_streams(stream_status); } if !failed_turn_was_finalized { - self.clear_turn_live_projection(); + let tool_outcome = if was_failed { + TurnToolOutcome::Failed + } else if was_interrupted { + TurnToolOutcome::Interrupted + } else { + TurnToolOutcome::Completed + }; + self.clear_turn_live_projection(tool_outcome); } if !failed_turn_was_finalized && (was_interrupted || was_failed) @@ -803,7 +831,7 @@ impl ChatWidget { let failed_turn_was_finalized = self.failed_turn_visually_finalized; if !failed_turn_was_finalized { self.commit_active_streams(DotStatus::Failed); - self.clear_turn_live_projection(); + self.clear_turn_live_projection(TurnToolOutcome::Failed); if let Some(cell) = self .active_cell .as_mut() diff --git a/crates/tui/src/chatwidget_tests.rs b/crates/tui/src/chatwidget_tests.rs index a802de8c..bdc91fd9 100644 --- a/crates/tui/src/chatwidget_tests.rs +++ b/crates/tui/src/chatwidget_tests.rs @@ -3919,6 +3919,7 @@ fn live_and_resume_error_share_same_rendering_chain() { true, false, )); + finalize_live_turn_for_history(&mut live_widget); let live_blob = scrollback_plain_lines(&live_widget.drain_scrollback_lines(80)) .into_iter() .filter(|line| line.contains("Ran bash error") || line.contains("permission denied")) @@ -5699,7 +5700,7 @@ fn committed_assistant_multiline_text_has_no_extra_blank_rows() { } #[test] -fn tool_call_start_and_finish_are_both_visible_in_history() { +fn tool_call_running_row_changes_to_ran_before_turn_commit() { let cwd = std::env::current_dir().expect("current directory is available"); let model = Model { slug: "test-model".to_string(), @@ -5726,7 +5727,7 @@ fn tool_call_start_and_finish_are_both_visible_in_history() { let running = rendered_rows(&widget, 80, 12).join("\n"); assert!( - running.contains("Running Get-Date"), + running.contains("Running") && running.contains("Get-Date"), "expected running tool cell, got:\n{running}" ); @@ -5738,19 +5739,17 @@ fn tool_call_start_and_finish_are_both_visible_in_history() { false, )); - let ran = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); - assert!( - !ran.contains("Running Get-Date"), - "running tool cell should not remain in history, got:\n{ran}" - ); + let ran_live = rendered_rows(&widget, 80, 12).join("\n"); assert!( - ran.contains("Ran Get-Date"), - "expected ran tool cell, got:\n{ran}" + !ran_live.contains("Running powershell"), + "running tool cell should update in place, got:\n{ran_live}" ); assert!( - !ran.contains("2026-05-09"), - "shell output should stay out of inline scrollback, got:\n{ran}" + ran_live.contains("Ran") && ran_live.contains("Get-Date"), + "expected ran tool cell, got:\n{ran_live}" ); + assert!(widget.drain_scrollback_lines(80).is_empty()); + assert!(!ran_live.contains("2026-05-09")); let transcript = widget .transcript_overlay_lines(80) .into_iter() @@ -5766,6 +5765,20 @@ fn tool_call_start_and_finish_are_both_visible_in_history() { transcript.contains("2026-05-09"), "shell output should appear in transcript overlay, got:\n{transcript}" ); + + widget.handle_worker_event(crate::events::WorkerEvent::TurnFinished { + stop_reason: "Completed".to_string(), + turn_count: 1, + total_input_tokens: 0, + total_output_tokens: 0, + total_tokens: 0, + total_cache_read_tokens: 0, + last_query_total_tokens: 0, + last_query_input_tokens: 0, + prompt_token_estimate: 0, + }); + let committed = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + assert_eq!(committed.matches("Ran").count(), 1, "{committed}"); } #[test] @@ -5817,7 +5830,7 @@ fn web_search_tool_call_renders_title_and_status_without_running_prefix() { false, )); - let rendered = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join( + let rendered = rendered_rows(&widget, 80, 12).join( " ", ); @@ -5887,7 +5900,7 @@ fn web_fetch_tool_call_renders_title_and_status_without_running_prefix() { false, )); - let rendered = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join( + let rendered = rendered_rows(&widget, 80, 12).join( " ", ); @@ -5996,6 +6009,23 @@ fn generic_running_tool_call_disappears_after_result() { !rendered.contains("Running code_search"), "running row should disappear after result:\n{rendered}" ); + assert!(rendered.contains("Ran code_search"), "{rendered}"); + assert!( + rendered.contains("Missing necessary parameter display"), + "{rendered}" + ); + + widget.handle_worker_event(crate::events::WorkerEvent::TurnFinished { + stop_reason: "Completed".to_string(), + turn_count: 1, + total_input_tokens: 0, + total_output_tokens: 0, + total_tokens: 0, + total_cache_read_tokens: 0, + last_query_total_tokens: 0, + last_query_input_tokens: 0, + prompt_token_estimate: 0, + }); let history = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); assert!( @@ -6066,10 +6096,9 @@ fn edit_running_row_is_path_free_and_disappears_after_patch_result() { !after.contains("Editing"), "completed Edit should leave no live row:\n{after}" ); - let history = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); assert!( - history.contains("Edited test_edit_test.md") || history.contains("Edited 1 file"), - "completed Edit diff should remain visible:\n{history}" + after.contains("Edited test_edit_test.md") || after.contains("Edited 1 file"), + "completed Edit diff should remain visible:\n{after}" ); } @@ -6435,6 +6464,166 @@ fn legacy_failed_turn_finished_flushes_explored_before_footer() { assert!(!history.contains("interrupted"), "history:\n{history}"); } +#[test] +fn late_tool_events_after_turn_finish_do_not_repin_row_to_live_viewport() { + let cwd = std::env::current_dir().expect("current directory is available"); + let model = Model { + slug: "test-model".to_string(), + display_name: "Test Model".to_string(), + ..Model::default() + }; + let (mut widget, _app_event_rx) = widget_with_model(model, cwd); + let _ = widget.drain_scrollback_lines(100); + + widget.handle_worker_event(crate::worker_event_test_helpers::command_execution_started( + "bash-1".to_string(), + "cargo test".to_string(), + None, + devo_protocol::protocol::ExecCommandSource::Agent, + Vec::new(), + )); + widget.handle_worker_event(crate::worker_event_test_helpers::tool_output_delta( + "bash-1".to_string(), + "test result: ok\n".to_string(), + )); + let live = line_texts(widget.active_viewport_lines_for_test(100)).join("\n"); + assert!( + live.contains("cargo test"), + "running tool row should render in the live viewport:\n{live}" + ); + + // Result lands while the turn is still active, then the turn boundary + // commits the row into scrollback history. + widget.handle_worker_event(crate::worker_event_test_helpers::tool_result( + "bash-1".to_string(), + "Shell cargo test".to_string(), + "test result: ok\n".to_string(), + false, + false, + )); + finalize_live_turn_for_history(&mut widget); + + let history = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); + assert!( + history.contains("cargo test"), + "committed tool row should land in history:\n{history}" + ); + + // The `ToolResult`/`item` notifications race past the turn's terminal + // event and are dispatched afterwards. They must not re-pin the finished + // row to the live viewport: after the boundary nothing would ever flush + // it into history, so it would sit above the composer forever. + widget.handle_worker_event(crate::worker_event_test_helpers::tool_result( + "bash-1".to_string(), + "Shell cargo test".to_string(), + "test result: ok\n".to_string(), + false, + false, + )); + widget.handle_worker_event(crate::worker_event_test_helpers::command_execution_started( + "bash-1".to_string(), + "cargo test".to_string(), + None, + devo_protocol::protocol::ExecCommandSource::Agent, + Vec::new(), + )); + + let live_after = line_texts(widget.active_viewport_lines_for_test(100)).join("\n"); + assert!( + !live_after.contains("Ran cargo test") && !live_after.contains("Running cargo test"), + "late duplicate events must not re-pin the tool row to the live viewport:\n{live_after}" + ); + let history_after = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); + assert!( + !history_after.contains("cargo test"), + "late duplicate events must not append a second committed cell:\n{history_after}" + ); +} + +#[test] +fn text_completion_commits_older_explored_tools_before_text() { + let cwd = std::env::current_dir().expect("current directory is available"); + let model = Model { + slug: "test-model".to_string(), + display_name: "Test Model".to_string(), + ..Model::default() + }; + let (mut widget, _app_event_rx) = widget_with_model(model, cwd); + let _ = widget.drain_scrollback_lines(100); + + widget.handle_worker_event(crate::worker_event_test_helpers::tool_call( + "tool-1".to_string(), + "grep 'plan' in crates".to_string(), + false, + Some(vec![devo_protocol::parse_command::ParsedCommand::Search { + cmd: "grep 'plan' in crates".to_string(), + query: Some("plan".to_string()), + path: Some("crates".to_string()), + }]), + )); + widget.handle_worker_event(crate::worker_event_test_helpers::tool_result( + "tool-1".to_string(), + "grep 'plan' in crates".to_string(), + String::new(), + false, + false, + )); + + // Assistant text starts streaming: the exploring group detaches and the + // finished tools render as individual live rows while the text streams + // below them. + let text_id = devo_core::ItemId::new(); + widget.handle_worker_event(crate::worker_event_test_helpers::text_item_started( + text_id, + crate::events::TextItemKind::Assistant, + )); + widget.handle_worker_event(crate::worker_event_test_helpers::text_item_delta( + text_id, + crate::events::TextItemKind::Assistant, + "Found it in ", + )); + let live = line_texts(widget.active_viewport_lines_for_test(100)).join("\n"); + assert!( + live.contains("Found it in"), + "streaming text should render in the live viewport:\n{live}" + ); + + // When the text commits mid-turn, the tools that ran before it must + // commit first; otherwise scrollback would show [text, tools] even + // though the tools happened first, with the tool rows repinned right + // above the composer below the finished reply. + widget.handle_worker_event(crate::worker_event_test_helpers::text_item_completed( + text_id, + crate::events::TextItemKind::Assistant, + "Found it in crates/chatwidget.rs.", + )); + + let history = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); + let tool_row = history.find("Grepped plan in crates"); + let text_row = history.find("Found it in crates/chatwidget.rs."); + assert!( + tool_row.is_some() && text_row.is_some(), + "expected tool row and assistant text in history:\n{history}" + ); + assert!( + tool_row < text_row, + "tools that ran before the text must commit above it:\n{history}" + ); + let live_after = line_texts(widget.active_viewport_lines_for_test(100)).join("\n"); + assert!( + !live_after.contains("Found it in"), + "committed text must leave the live viewport:\n{live_after}" + ); + + // The turn boundary has nothing left to flush: no duplicate commits. + finalize_live_turn_for_history(&mut widget); + let history_after = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); + assert!( + !history_after.contains("Grepped plan in crates"), + "boundary must not re-commit the flushed tool row:\n{history_after}" + ); +} + #[test] fn preparing_write_disappears_after_patch_applied() { let cwd = std::env::current_dir().expect("current directory is available"); @@ -6474,11 +6663,10 @@ fn preparing_write_disappears_after_patch_applied() { !after.contains("Preparing write..."), "preparing state should disappear after patch applied:\n{after}" ); - let history = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); assert!( - history.contains("Added src/lib.rs") - || history.contains("Edited src/lib.rs") - || history.contains("Added 1 file") + after.contains("Added src/lib.rs") + || after.contains("Edited src/lib.rs") + || after.contains("Added 1 file") ); } @@ -9928,7 +10116,7 @@ fn duplicate_command_execution_start_is_idempotent() { let transcript = line_texts(widget.transcript_overlay_lines(100)).join("\n"); assert_eq!( - transcript.matches("Ran pwd").count(), + transcript.matches("Running pwd").count(), 1, "duplicate starts should retain one command cell:\n{transcript}" ); @@ -9969,7 +10157,7 @@ fn transcript_overlay_lines_include_completed_tool_input_and_full_output() { false, )); - let inline = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + let inline = line_texts(widget.active_viewport_lines_for_test(80)).join("\n"); let transcript = line_texts(widget.transcript_overlay_lines(80)).join("\n"); assert!( @@ -10957,9 +11145,15 @@ fn reasoning_start_closes_current_explored_group() { .join("\n"); assert_eq!( - transcript.matches("Explored").count() + transcript.matches("Exploring").count(), - 2, - "reasoning boundary should split explored groups:\n{transcript}" + transcript.matches("Grepping 'plan' in crates").count() + + transcript.matches("Grepped 'plan' in crates").count(), + 1, + "{transcript}" + ); + assert_eq!( + transcript.matches("Finding crates").count() + transcript.matches("Found crates").count(), + 1, + "{transcript}" ); } @@ -11011,9 +11205,15 @@ fn assistant_text_start_closes_current_explored_group() { .join("\n"); assert_eq!( - transcript.matches("Explored").count() + transcript.matches("Exploring").count(), - 2, - "assistant text boundary should split explored groups:\n{transcript}" + transcript.matches("Grepping 'plan' in crates").count() + + transcript.matches("Grepped 'plan' in crates").count(), + 1, + "{transcript}" + ); + assert_eq!( + transcript.matches("Finding crates").count() + transcript.matches("Found crates").count(), + 1, + "{transcript}" ); } @@ -11221,7 +11421,7 @@ fn patch_applied_event_renders_edited_block() { changes, )); - let blob = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + let blob = transcript_overlay_text(&widget, 80); assert!( blob.contains("Edited foo.txt") || blob.contains("Edited 1 file"), "expected edited patch block, got:\n{blob}" @@ -11250,7 +11450,7 @@ fn added_file_patch_applied_event_renders_added_content_lines() { changes, )); - let blob = scrollback_plain_lines(&widget.drain_scrollback_lines(100)).join("\n"); + let blob = transcript_overlay_text(&widget, 100); assert!( blob.contains("Added quicksort.rs") || blob.contains("Edited quicksort.rs") @@ -11291,7 +11491,7 @@ fn apply_patch_style_full_git_diff_reports_non_zero_counts() { changes, )); - let blob = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + let blob = transcript_overlay_text(&widget, 80); assert!( blob.contains("(+1 -1)"), "full git-style apply_patch diff should report non-zero counts:\n{blob}" @@ -11345,7 +11545,7 @@ fn write_patch_applied_event_renders_edited_block() { changes, )); - let blob = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + let blob = transcript_overlay_text(&widget, 80); assert!( blob.contains("Edited foo.txt") || blob.contains("Edited 1 file"), "expected edited patch block for write result, got:\n{blob}" @@ -11377,7 +11577,7 @@ fn write_patch_applied_event_reports_non_zero_counts() { changes, )); - let blob = scrollback_plain_lines(&widget.drain_scrollback_lines(80)).join("\n"); + let blob = transcript_overlay_text(&widget, 80); assert!( !blob.contains("Edited 0 files (+0 -0)"), "write-derived edited block should not collapse to zero summary:\n{blob}" diff --git a/crates/tui/src/insert_history.rs b/crates/tui/src/insert_history.rs index 61060dd7..0ec67d5b 100644 --- a/crates/tui/src/insert_history.rs +++ b/crates/tui/src/insert_history.rs @@ -48,6 +48,15 @@ impl InsertHistoryMode { } } +/// Geometry produced by one history insertion transaction. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct HistoryInsertOutcome { + pub(crate) inserted_rows: u16, + pub(crate) scrolled_rows: u16, + pub(crate) viewport_before: ratatui::layout::Rect, + pub(crate) viewport_after: ratatui::layout::Rect, +} + /// Insert `lines` above the viewport using the terminal's backend writer /// (avoids direct stdout references). pub fn insert_history_lines( @@ -57,7 +66,7 @@ pub fn insert_history_lines( where B: Backend + Write, { - insert_history_lines_with_mode(terminal, lines, InsertHistoryMode::Standard) + insert_history_lines_with_mode(terminal, lines, InsertHistoryMode::Standard).map(|_| ()) } /// Insert `lines` above the viewport, using the escape strategy selected by `mode`. @@ -72,14 +81,16 @@ pub fn insert_history_lines_with_mode( terminal: &mut crate::custom_terminal::Terminal, lines: Vec, mode: InsertHistoryMode, -) -> io::Result<()> +) -> io::Result where B: Backend + Write, { let screen_size = terminal.backend().size().unwrap_or(Size::new(0, 0)); - let mut area = terminal.viewport_area; + let viewport_before = terminal.viewport_area; + let mut area = viewport_before; let mut should_update_area = false; + let mut scrolled_rows = 0; let last_cursor_pos = terminal.last_known_cursor_pos; let writer = terminal.backend_mut(); @@ -97,10 +108,19 @@ where } let wrapped_lines = wrapped_rows as u16; - if matches!(mode, InsertHistoryMode::Zellij) { + // The Standard strategy needs room above the viewport for a valid DECSTBM + // region (`SetScrollRegion(1..area.top())` degenerates to the invalid + // `\x1b[1;0r` when the viewport sits at the very top — e.g. right after a + // full-screen clear or when the live viewport fills the screen). With no + // room above, fall back to the bottom-newline strategy, which only uses + // cursor moves and prints and is safe on every terminal. + let use_bottom_scroll = matches!(mode, InsertHistoryMode::Zellij) || area.top() < 2; + + if use_bottom_scroll { let space_below = screen_size.height.saturating_sub(area.bottom()); let shift_down = wrapped_lines.min(space_below); let scroll_up_amount = wrapped_lines.saturating_sub(shift_down); + scrolled_rows = scroll_up_amount; if scroll_up_amount > 0 { // Scroll the entire screen up by emitting \n at the bottom @@ -127,6 +147,7 @@ where } else { let cursor_top = if area.bottom() < screen_size.height { let scroll_amount = wrapped_lines.min(screen_size.height - area.bottom()); + scrolled_rows = wrapped_lines.saturating_sub(scroll_amount); let top_1based = area.top() + 1; queue!(writer, SetScrollRegion(top_1based..screen_size.height))?; @@ -185,7 +206,12 @@ where terminal.note_history_rows_inserted(wrapped_lines); } - Ok(()) + Ok(HistoryInsertOutcome { + inserted_rows: wrapped_lines, + scrolled_rows, + viewport_before, + viewport_after: terminal.viewport_area, + }) } /// Render a single wrapped history line: clear continuation rows for wide lines, @@ -804,6 +830,52 @@ mod tests { ); } + #[test] + fn standard_mode_with_viewport_at_screen_top_uses_bottom_scroll_strategy() { + // Degenerate geometry: the viewport sits at the very top of the + // screen (e.g. right after a full-screen clear, or a live viewport + // that grew to fill the screen). The DECSTBM strategy would emit the + // invalid `\x1b[1;0r`; the insert must fall back to the safe + // bottom-newline strategy instead. + let width: u16 = 30; + let height: u16 = 6; + let backend = VT100Backend::new(width, height); + let mut term = crate::custom_terminal::Terminal::with_options(backend).expect("terminal"); + let viewport = Rect::new(0, 0, width, height); + term.set_viewport_area(viewport); + + insert_history_lines(&mut term, vec![Line::from("history line").into()]) + .expect("insert history with viewport at screen top"); + + let rows: Vec = term.backend().vt100().screen().rows(0, width).collect(); + assert!( + rows.iter().any(|row| row.contains("history line")), + "expected history row in screen output, rows: {rows:?}" + ); + } + + #[test] + fn standard_mode_with_small_viewport_at_screen_top_shifts_viewport_down() { + // Viewport at the top with room below: the fallback writes history at + // row 0 and moves the tracked viewport area down accordingly. + let width: u16 = 30; + let height: u16 = 6; + let backend = VT100Backend::new(width, height); + let mut term = crate::custom_terminal::Terminal::with_options(backend).expect("terminal"); + let viewport = Rect::new(0, 0, width, 2); + term.set_viewport_area(viewport); + + insert_history_lines(&mut term, vec![Line::from("history line").into()]) + .expect("insert history above top-pinned small viewport"); + + let rows: Vec = term.backend().vt100().screen().rows(0, width).collect(); + assert!( + rows[0].contains("history line"), + "expected history row at screen top, rows: {rows:?}" + ); + assert_eq!(term.viewport_area, Rect::new(0, 1, width, 2)); + } + #[test] fn vt100_zellij_mode_inserts_history_and_updates_viewport() { let width: u16 = 32; @@ -814,8 +886,9 @@ mod tests { term.set_viewport_area(viewport); let line: Line<'static> = Line::from("zellij history"); - insert_history_lines_with_mode(&mut term, vec![line.into()], InsertHistoryMode::Zellij) - .expect("insert zellij history"); + let outcome = + insert_history_lines_with_mode(&mut term, vec![line.into()], InsertHistoryMode::Zellij) + .expect("insert zellij history"); let rows: Vec = term.backend().vt100().screen().rows(0, width).collect(); assert!( @@ -824,5 +897,9 @@ mod tests { ); assert_eq!(term.viewport_area, Rect::new(0, 5, width, 2)); assert_eq!(term.visible_history_rows(), 1); + assert_eq!(outcome.inserted_rows, 1); + assert_eq!(outcome.scrolled_rows, 0); + assert_eq!(outcome.viewport_before, viewport); + assert_eq!(outcome.viewport_after, term.viewport_area); } } diff --git a/crates/tui/src/interactive.rs b/crates/tui/src/interactive.rs index ec6ca42e..cd513f1d 100644 --- a/crates/tui/src/interactive.rs +++ b/crates/tui/src/interactive.rs @@ -125,6 +125,9 @@ struct InteractiveLoopState { // True after clearing the inline UI for a session switch and before the // replacement session has been restored into widget state. session_switch_pending: bool, + // When the pending switch started; guards against a wiped screen staying + // blank forever if the switch flow never emits a terminal event. + session_switch_pending_since: Option, pending_backtrack_restore: Option, last_ctrl_c_at: Option, esc_backtrack_primed: bool, @@ -137,6 +140,33 @@ enum LoopAction { ClearAndExit, } +/// How long a pending session switch may suppress draws before the loop +/// force-resumes painting. The inline UI is wiped when the switch begins; if +/// the worker never emits `SessionSwitched`/`TurnFinished`/`TurnFailed` +/// (e.g. a failed goal-pause step), the screen would otherwise stay blank. +const SESSION_SWITCH_PENDING_TIMEOUT: Duration = Duration::from_secs(2); + +impl InteractiveLoopState { + fn begin_session_switch(&mut self) { + self.session_switch_pending = true; + self.session_switch_pending_since = Some(Instant::now()); + } + + fn end_session_switch(&mut self) { + self.session_switch_pending = false; + self.session_switch_pending_since = None; + } + + /// True when a pending switch has outlived its budget and draws must + /// resume regardless of the missing terminal event. + fn session_switch_pending_expired(&self) -> bool { + self.session_switch_pending + && self + .session_switch_pending_since + .is_some_and(|started| started.elapsed() > SESSION_SWITCH_PENDING_TIMEOUT) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum CtrlCKeyAction { Interrupt, @@ -343,7 +373,6 @@ pub async fn run_interactive_tui(config: InteractiveTuiConfig) -> Result {} LoopAction::ClearAndExit => { tracing::info!("interactive loop exiting from tui event"); - clear_before_exit(&mut tui)?; break; } } @@ -366,7 +395,6 @@ pub async fn run_interactive_tui(config: InteractiveTuiConfig) -> Result {} LoopAction::ClearAndExit => { tracing::info!("interactive loop exiting from app event"); - clear_before_exit(&mut tui)?; break; } } @@ -381,7 +409,6 @@ pub async fn run_interactive_tui(config: InteractiveTuiConfig) -> Result {} LoopAction::ClearAndExit => { tracing::info!("interactive loop exiting from worker event"); - clear_before_exit(&mut tui)?; break; } } @@ -389,6 +416,8 @@ pub async fn run_interactive_tui(config: InteractiveTuiConfig) -> Result Result<()> { +fn clear_before_exit(tui: &mut Tui, chat_widget: &mut ChatWidget) -> Result<()> { tracing::info!("clearing tui before exit"); - let result = tui.shutdown_terminal_safe(); + let size = tui.terminal.size()?; + let width = size.width.max(1); + let completed_history = chat_widget.drain_scrollback_lines(width); + if !completed_history.is_empty() { + tui.insert_history_lines(completed_history); + } + let final_live_height = chat_widget + .desired_height(width) + .min(size.height.saturating_sub(1)) + .max(3); + let result = tui.shutdown_terminal_safe(final_live_height); tracing::info!( success = result.is_ok(), "finished clearing tui before exit" @@ -532,7 +571,7 @@ fn handle_tui_event( } loop_state.pending_backtrack_restore = Some(user_message); loop_state.overlay.close(tui)?; - loop_state.session_switch_pending = true; + loop_state.begin_session_switch(); tui.replace_inline_session_ui()?; worker.rollback_before_user_turn(user_turn_index)?; return Ok(LoopAction::Continue); @@ -552,7 +591,15 @@ fn handle_tui_event( match tui_event { TuiEvent::Draw => { if loop_state.session_switch_pending { - return Ok(LoopAction::Continue); + if loop_state.session_switch_pending_expired() { + // The switch flow died without a terminal event; the + // screen was already wiped, so resume painting now + // instead of leaving it blank. + loop_state.end_session_switch(); + chat_widget.set_status_message("Session switch stalled; resuming display"); + } else { + return Ok(LoopAction::Continue); + } } // Update time-sensitive widget state before measuring or rendering. @@ -852,7 +899,7 @@ fn handle_worker_event( loop_state.total_output_tokens = *next_total_output_tokens; loop_state.total_tokens = *next_total_tokens; loop_state.total_cache_read_tokens = *next_total_cache_read_tokens; - loop_state.session_switch_pending = false; + loop_state.end_session_switch(); } WorkerEvent::InterruptFailed { .. } => {} WorkerEvent::TurnStarted { .. } => { @@ -950,7 +997,7 @@ fn handle_worker_event( total_cache_read_tokens, .. } => { - loop_state.session_switch_pending = false; + loop_state.end_session_switch(); loop_state.session_id = devo_core::SessionId::try_from(session_id.as_str()).ok(); loop_state.total_input_tokens = *total_input_tokens; loop_state.total_output_tokens = *total_output_tokens; @@ -1003,11 +1050,19 @@ fn handle_worker_event( | WorkerEvent::GoalReplaceConfirmationRequested { .. } | WorkerEvent::GoalEditLoaded { .. } | WorkerEvent::GoalCleared { .. } - | WorkerEvent::GoalOperationFailed { .. } | WorkerEvent::BtwStarted { .. } | WorkerEvent::BtwCompleted { .. } | WorkerEvent::BtwFailed { .. } | WorkerEvent::EffectiveContextWindowUpdated { .. } => {} + WorkerEvent::GoalOperationFailed { .. } => { + // The switch/rollback flow aborts its goal-pause step with this + // event and emits no SessionSwitched/TurnFinished afterwards. + // Without clearing the pending flag here the wiped screen would + // stay blank (the draw-suppression timeout is the backstop). + if loop_state.session_switch_pending { + loop_state.end_session_switch(); + } + } } let session_switched = matches!(&worker_event, WorkerEvent::SessionSwitched { .. }); let turn_failed = matches!(&worker_event, WorkerEvent::TurnFailed { .. }); @@ -1308,12 +1363,12 @@ fn handle_app_command( } AppCommand::SwitchSession { session_id } => { tracing::trace!(session_id = ?session_id, "switch session requested"); - loop_state.session_switch_pending = true; + loop_state.begin_session_switch(); tui.replace_inline_session_ui()?; worker.switch_session(*session_id)?; } AppCommand::RollbackToUserTurn { user_turn_index } => { - loop_state.session_switch_pending = true; + loop_state.begin_session_switch(); tui.replace_inline_session_ui()?; worker.rollback_to_user_turn(*user_turn_index)?; } @@ -1321,7 +1376,7 @@ fn handle_app_command( user_turn_index, cut, } => { - loop_state.session_switch_pending = true; + loop_state.begin_session_switch(); tui.replace_inline_session_ui()?; worker.fork_at_user_turn(*user_turn_index, *cut)?; } diff --git a/crates/tui/src/transcript/lifecycle.rs b/crates/tui/src/transcript/lifecycle.rs index d003f693..2cc67015 100644 --- a/crates/tui/src/transcript/lifecycle.rs +++ b/crates/tui/src/transcript/lifecycle.rs @@ -13,6 +13,14 @@ use devo_protocol::protocol::FileChange; use crate::events::PlanStep; use crate::events::TextItemKind; +/// Authoritative outcome used when committing the current turn's live tools. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TurnToolOutcome { + Completed, + Failed, + Interrupted, +} + /// One transcript-affecting lifecycle transition. #[derive(Debug, Clone, PartialEq)] pub(crate) enum ItemLifecycleEvent { @@ -60,7 +68,7 @@ pub(crate) enum ItemLifecycleEvent { tool_use_id: String, chunk: String, }, - /// A tool row finished and should commit to history. + /// A tool row finished. It remains live until the turn commit boundary. ToolClosed { tool_use_id: String, tool_name: String, @@ -75,6 +83,8 @@ pub(crate) enum ItemLifecycleEvent { explanation: Option, steps: Vec, }, - /// Clears live tool rows when a turn ends without individual completions. - TurnLiveToolsCleared, + /// Commits every tool owned by the current turn in sequence order. + TurnLiveToolsCleared { + outcome: TurnToolOutcome, + }, } diff --git a/crates/tui/src/transcript/model.rs b/crates/tui/src/transcript/model.rs index 30cc84f8..6bd12429 100644 --- a/crates/tui/src/transcript/model.rs +++ b/crates/tui/src/transcript/model.rs @@ -20,6 +20,14 @@ pub(crate) enum ToolPhase { Running, Completed, Failed, + /// The turn completed, but no authoritative result arrived for this tool. + Degraded, +} + +impl ToolPhase { + pub(crate) fn is_terminal(self) -> bool { + matches!(self, Self::Completed | Self::Failed | Self::Degraded) + } } /// Unified tool representation: write/edit, exec, and generic tools share one model. @@ -172,10 +180,6 @@ impl ToolCellModel { truncated: false, } } - - pub(crate) fn is_live(&self) -> bool { - matches!(self.phase, ToolPhase::Preparing | ToolPhase::Running) - } } #[derive(Debug, Clone)] diff --git a/crates/tui/src/transcript/presentation.rs b/crates/tui/src/transcript/presentation.rs index 156e5f66..eb88b8e5 100644 --- a/crates/tui/src/transcript/presentation.rs +++ b/crates/tui/src/transcript/presentation.rs @@ -130,7 +130,7 @@ pub(crate) fn tool_title_parts( } if tool_name.is_some_and(super::tool_state::is_shell_tool_name) { - let completed = matches!(phase, ToolPhase::Completed | ToolPhase::Failed); + let completed = phase.is_terminal(); let verb = if completed { "Ran" } else { "Running" }; let detail = super::tool_state::shell_description_from_input(input).unwrap_or_else(|| { input @@ -180,7 +180,7 @@ pub(crate) fn tool_title_parts( let tool_name = tool_name.unwrap_or("tool"); let kind = ToolVerbKind::from_tool_name(tool_name); - let completed = matches!(phase, ToolPhase::Completed | ToolPhase::Failed); + let completed = phase.is_terminal(); let verb = if completed { kind.completed_verb(tool_name, completed_with_add) .to_string() @@ -202,7 +202,7 @@ fn agent_task_title_parts( phase: ToolPhase, input: Option<&serde_json::Value>, ) -> ToolTitleParts { - let completed = matches!(phase, ToolPhase::Completed | ToolPhase::Failed); + let completed = phase.is_terminal(); match tool_name { "spawn_agent" | "agent_spawn" => { let nickname = input @@ -310,15 +310,24 @@ pub(crate) fn tool_title_line(phase: ToolPhase, parts: &ToolTitleParts) -> Line< ]); } - let completed = matches!(phase, ToolPhase::Completed | ToolPhase::Failed); + let completed = phase.is_terminal(); let verb_style = if completed { tool_status_done_style() } else { tool_status_running_style() }; + let degraded_suffix = if phase == ToolPhase::Degraded { + " · result unavailable" + } else { + "" + }; + if parts.verb.is_empty() { - return Line::from(Span::styled(parts.detail.clone(), tool_text_style())); + return Line::from(Span::styled( + format!("{}{degraded_suffix}", parts.detail), + tool_text_style(), + )); } let detail = if parts.detail.is_empty() { @@ -330,6 +339,7 @@ pub(crate) fn tool_title_line(phase: ToolPhase, parts: &ToolTitleParts) -> Line< Line::from(vec![ Span::styled(parts.verb.clone(), verb_style), Span::styled(detail, tool_text_style()), + Span::styled(degraded_suffix, tool_text_style()), ]) } @@ -349,7 +359,7 @@ pub(crate) fn title_from_parsed_command( parsed: &ParsedCommand, phase: ToolPhase, ) -> ToolTitleParts { - let completed = matches!(phase, ToolPhase::Completed | ToolPhase::Failed); + let completed = phase.is_terminal(); match parsed { ParsedCommand::Read { name, path, cmd } => { let detail = read_display_name(name, path, cmd); diff --git a/crates/tui/src/transcript/projector.rs b/crates/tui/src/transcript/projector.rs index 05fcf118..ff13b8c4 100644 --- a/crates/tui/src/transcript/projector.rs +++ b/crates/tui/src/transcript/projector.rs @@ -1,8 +1,10 @@ //! Applies [`ItemLifecycleEvent`] values to a single transcript projection. use std::collections::HashMap; +use std::collections::HashSet; use crate::transcript::lifecycle::ItemLifecycleEvent; +use crate::transcript::lifecycle::TurnToolOutcome; use crate::transcript::model::CommittedCellModel; use crate::transcript::model::LiveTextCellModel; use crate::transcript::model::TextCellModel; @@ -19,6 +21,12 @@ use super::stream_text::apply_stream_text_delta; pub(crate) struct TranscriptProjector { tools: HashMap, tool_order: Vec, + /// Call ids this projector has already committed (turn boundary or + /// restore). Late lifecycle events for these ids must not re-materialize + /// live rows: after the boundary nothing would ever flush them, so a + /// refreshed row would stay rendered at the bottom of the live viewport + /// next to the composer. + committed_tool_ids: HashSet, live_text: HashMap, text_order: Vec, next_seq: u64, @@ -41,6 +49,13 @@ impl TranscriptProjector { tool.refresh_opened(tool_name, input, command, command_source, parsed_commands); return; } + if self.committed_tool_ids.contains(&tool_use_id) { + // A refresh (`item/started` re-broadcast or the completed + // `ToolCall` item) for a row the boundary already committed. + // Re-materializing it would pin a live row to the bottom of + // the viewport that no future boundary would ever flush. + return; + } let seq = self.reserve_seq(); let tool = ToolModel::new_opened( tool_use_id.clone(), @@ -67,6 +82,14 @@ impl TranscriptProjector { { tool.phase = initial_phase(&name, &parsed); } + } else if let Some(partial) = + super::tool_state::partial_object_members(&tool.input_partial_json) + { + // Display-only fill while the JSON is still streaming: + // the running row shows its parameters (filePath, + // pattern, command, …) as soon as each member + // completes instead of only at the tool result. + tool.input = Some(partial); } } } @@ -96,8 +119,7 @@ impl TranscriptProjector { self.tool_order.retain(|id| id != &tool_use_id); return; } - if let Some(mut tool) = self.tools.remove(&tool_use_id) { - self.tool_order.retain(|id| id != &tool_use_id); + if let Some(tool) = self.tools.get_mut(&tool_use_id) { if tool_name != "tool" { tool.tool_name = Some(tool_name); } @@ -119,24 +141,31 @@ impl TranscriptProjector { } else { ToolPhase::Completed }; - self.committed.push(CommittedCellModel::Tool(tool)); + } else if self.committed_tool_ids.contains(&tool_use_id) { + // The boundary already committed this row; a late close + // (a notification that raced past the turn's terminal + // event) must not re-materialize it in the live viewport. } else { - let seq = self.reserve_seq(); + // A close for a call this projector never saw open + // (recovery sweep, missed open) is already terminal: + // commit it directly instead of parking it live, where + // only the next turn boundary would flush it. + let exec_like = super::tool_state::is_shell_tool_name(&tool_name); let phase = if is_error { ToolPhase::Failed } else { ToolPhase::Completed }; - self.committed.push(CommittedCellModel::Tool(ToolModel { - tool_use_id, - seq, + let tool = ToolModel { + tool_use_id: tool_use_id.clone(), + seq: 0, phase, summary: String::new(), tool_name: Some(tool_name), input: Some(input), input_partial_json: String::new(), parsed_commands: Vec::new(), - exec_like: false, + exec_like, start_time: None, output_preview: display_content.clone().unwrap_or_default(), output_delta_lines: Vec::new(), @@ -149,12 +178,34 @@ impl TranscriptProjector { tool_display_content: display_content, is_error, truncated, - })); + }; + self.commit_tool_model(tool); } } - ItemLifecycleEvent::TurnLiveToolsCleared => { + ItemLifecycleEvent::TurnLiveToolsCleared { outcome } => { + for tool_use_id in std::mem::take(&mut self.tool_order) { + if let Some(mut tool) = self.tools.remove(&tool_use_id) { + if tool.phase == ToolPhase::Preparing && tool.input.is_none() { + // A row that never received any payload carries no + // renderable facts; drop it rather than commit + // an empty cell. + continue; + } + if matches!(tool.phase, ToolPhase::Preparing | ToolPhase::Running) { + tool.phase = match outcome { + TurnToolOutcome::Completed => ToolPhase::Degraded, + TurnToolOutcome::Failed | TurnToolOutcome::Interrupted => { + ToolPhase::Failed + } + }; + if tool.phase == ToolPhase::Failed { + tool.is_error = true; + } + } + self.commit_tool_model(tool); + } + } self.tools.clear(); - self.tool_order.clear(); self.live_text.clear(); self.text_order.clear(); } @@ -203,11 +254,22 @@ impl TranscriptProjector { kind, final_text, } => { + let text_seq = self.live_text.get(&item_id).map(|live| live.seq); self.live_text.remove(&item_id); self.text_order.retain(|id| *id != item_id); if final_text.trim().is_empty() { return; } + // The TUI commits finished text to scrollback immediately, + // while tools stay in the live viewport until the turn + // boundary. Text that follows tools in event order must not + // overtake them: flush terminal tools that ran before this + // text so `committed` stays in transcript order (otherwise + // the tools would land in history below the text they + // preceded). + if let Some(text_seq) = text_seq { + self.commit_terminal_tools_older_than(text_seq); + } self.committed.push(CommittedCellModel::Text(TextCellModel { item_id, kind, @@ -222,14 +284,11 @@ impl TranscriptProjector { } pub(crate) fn live_tools(&self) -> impl Iterator { - self.tool_order - .iter() - .filter_map(|id| self.tools.get(id)) - .filter(|tool| tool.is_live()) + self.tool_order.iter().filter_map(|id| self.tools.get(id)) } pub(crate) fn live_tool(&self, tool_use_id: &str) -> Option<&ToolCellModel> { - self.tools.get(tool_use_id).filter(|tool| tool.is_live()) + self.tools.get(tool_use_id) } pub(crate) fn live_text_items(&self) -> impl Iterator { @@ -268,6 +327,11 @@ impl TranscriptProjector { } pub(crate) fn restore_committed(&mut self, cells: Vec) { + for cell in &cells { + if let CommittedCellModel::Tool(tool) = cell { + self.committed_tool_ids.insert(tool.tool_use_id.clone()); + } + } self.committed = cells; self.synced_committed = 0; } @@ -277,6 +341,34 @@ impl TranscriptProjector { self.next_seq = self.next_seq.wrapping_add(1); seq } + + fn commit_tool_model(&mut self, tool: ToolModel) { + self.committed_tool_ids.insert(tool.tool_use_id.clone()); + self.committed.push(CommittedCellModel::Tool(tool)); + } + + /// Commits live tools that already finished before `seq`, preserving + /// transcript order when a text cell commits mid-turn (see + /// `ItemLifecycleEvent::TextCompleted`). + fn commit_terminal_tools_older_than(&mut self, seq: u64) { + let older: Vec = self + .tool_order + .iter() + .filter(|id| { + self.tools + .get(*id) + .is_some_and(|tool| tool.seq < seq && tool.phase.is_terminal()) + }) + .cloned() + .collect(); + for tool_use_id in older { + let Some(tool) = self.tools.remove(&tool_use_id) else { + continue; + }; + self.tool_order.retain(|id| *id != tool_use_id); + self.commit_tool_model(tool); + } + } } #[cfg(test)] @@ -289,6 +381,61 @@ mod tests { use super::*; + #[test] + fn tool_input_chunks_fill_parameters_progressively_while_streaming() { + use crate::transcript::presentation::tool_title_parts; + + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "read-1".into(), + tool_name: "read".into(), + input: serde_json::Value::Null, + command: None, + command_source: None, + parsed_commands: Vec::new(), + }); + + // First fragment: the key is still streaming — nothing displayable yet. + projector.apply(ItemLifecycleEvent::ToolInputChunk { + tool_use_id: "read-1".into(), + chunk: r#"{"filePa"#.to_string(), + }); + let tool = projector.live_tool("read-1").expect("live tool"); + assert!( + tool.input.is_none() || tool.input.as_ref().is_some_and(serde_json::Value::is_null) + ); + + // filePath completes: the running row can render it immediately. + projector.apply(ItemLifecycleEvent::ToolInputChunk { + tool_use_id: "read-1".into(), + chunk: r#"th": "src/lib.rs", "offs"#.to_string(), + }); + let tool = projector.live_tool("read-1").expect("live tool"); + let input = tool.input.clone().expect("partial input"); + assert_eq!(input["filePath"], serde_json::json!("src/lib.rs")); + let parts = tool_title_parts( + tool.phase, + tool.tool_name.as_deref(), + tool.input.as_ref(), + &tool.parsed_commands, + false, + tool.summary.as_str(), + ); + assert_eq!(parts.verb, "Reading"); + assert!(parts.detail.contains("src/lib.rs")); + + // Full JSON arrives: the authoritative input replaces the partial view. + projector.apply(ItemLifecycleEvent::ToolInputChunk { + tool_use_id: "read-1".into(), + chunk: r#"et": 10}"#.to_string(), + }); + let tool = projector.live_tool("read-1").expect("live tool"); + assert_eq!( + tool.input, + Some(serde_json::json!({"filePath": "src/lib.rs", "offset": 10})), + ); + } + #[test] fn tool_close_preserves_opened_metadata_when_result_omits_tool_name() { use crate::transcript::presentation::tool_title_line; @@ -314,8 +461,13 @@ mod tests { truncated: false, }); + let live = projector.live_tool("grep-1").expect("completed live tool"); + assert_eq!(live.phase, ToolPhase::Completed); + assert!(projector.drain_unsynced_committed().is_empty()); + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: TurnToolOutcome::Completed, + }); let committed = projector.drain_unsynced_committed(); - assert_eq!(committed.len(), 1); let CommittedCellModel::Tool(tool) = &committed[0] else { panic!("expected committed tool cell"); }; @@ -343,6 +495,220 @@ mod tests { assert_eq!(title_text, "Grepped 'plan' in crates"); } + #[test] + fn late_close_after_boundary_does_not_rematerialize_live_row() { + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "bash-1".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "cargo test"}), + command: Some("cargo test".into()), + command_source: None, + parsed_commands: Vec::new(), + }); + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: TurnToolOutcome::Completed, + }); + assert!(projector.live_tool("bash-1").is_none()); + assert_eq!(projector.drain_unsynced_committed().len(), 1); + + // A `ToolResult` notification that raced past the turn's terminal + // event arrives after the boundary. It must not re-open a live row: + // nothing would ever flush it back into history. + projector.apply(ItemLifecycleEvent::ToolClosed { + tool_use_id: "bash-1".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "cargo test"}), + output: Some(serde_json::json!("ok")), + display_content: Some("ok".into()), + file_changes: None, + is_error: false, + truncated: false, + }); + + assert!( + projector.live_tool("bash-1").is_none(), + "late close must not re-materialize a committed row" + ); + assert!( + projector.drain_unsynced_committed().is_empty(), + "late close for a committed row must not append a duplicate cell" + ); + + // Same for the completed `ToolCall` refresh: it must not reopen the + // row as a live "Running" entry. + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "bash-1".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "cargo test"}), + command: Some("cargo test".into()), + command_source: None, + parsed_commands: Vec::new(), + }); + assert!( + projector.live_tool("bash-1").is_none(), + "late open refresh must not re-materialize a committed row" + ); + } + + #[test] + fn text_completion_commits_older_tools_before_text() { + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "grep-1".into(), + tool_name: "grep".into(), + input: serde_json::json!({"pattern": "plan"}), + command: None, + command_source: None, + parsed_commands: Vec::new(), + }); + let item_id = devo_core::ItemId::new(); + projector.apply(ItemLifecycleEvent::TextStarted { + item_id, + kind: crate::events::TextItemKind::Assistant, + }); + projector.apply(ItemLifecycleEvent::ToolClosed { + tool_use_id: "grep-1".into(), + tool_name: "grep".into(), + input: serde_json::json!({"pattern": "plan"}), + output: Some(serde_json::json!("matches")), + display_content: Some("matches".into()), + file_changes: None, + is_error: false, + truncated: false, + }); + // Tools stay live until the turn boundary, so the finished grep must + // still be parked in the live projection here. + assert!(projector.live_tool("grep-1").is_some()); + assert!(projector.drain_unsynced_committed().is_empty()); + + projector.apply(ItemLifecycleEvent::TextCompleted { + item_id, + kind: crate::events::TextItemKind::Assistant, + final_text: "Here is what I found.".into(), + }); + + // Text commits mid-turn; the grep that ran before it must commit + // first or scrollback would order [text, tool] against event order. + let committed = projector.drain_unsynced_committed(); + let labels: Vec<&str> = committed + .iter() + .map(|cell| match cell { + CommittedCellModel::Tool(_) => "tool", + CommittedCellModel::Text(_) => "text", + }) + .collect(); + assert_eq!(labels, vec!["tool", "text"]); + assert!(projector.live_tool("grep-1").is_none()); + } + + #[test] + fn text_completion_leaves_older_running_tools_live() { + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "bash-1".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "cargo build"}), + command: Some("cargo build".into()), + command_source: None, + parsed_commands: Vec::new(), + }); + projector.apply(ItemLifecycleEvent::ToolOutputChunk { + tool_use_id: "bash-1".into(), + chunk: "Compiling\n".into(), + }); + let item_id = devo_core::ItemId::new(); + projector.apply(ItemLifecycleEvent::TextStarted { + item_id, + kind: crate::events::TextItemKind::Assistant, + }); + projector.apply(ItemLifecycleEvent::TextCompleted { + item_id, + kind: crate::events::TextItemKind::Assistant, + final_text: "Still building.".into(), + }); + + // A tool that has not reached a terminal phase is not force-finished + // by the text commit; the turn boundary still owns its outcome. + let committed = projector.drain_unsynced_committed(); + assert_eq!(committed.len(), 1); + assert!(matches!(committed[0], CommittedCellModel::Text(_))); + assert!(projector.live_tool("bash-1").is_some()); + } + + #[test] + fn close_for_unknown_call_commits_directly_instead_of_parking_live() { + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolClosed { + tool_use_id: "bash-9".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "ls"}), + output: Some(serde_json::json!("src\n")), + display_content: Some("src\n".into()), + file_changes: None, + is_error: false, + truncated: false, + }); + + assert!( + projector.live_tool("bash-9").is_none(), + "terminal facts must not park a live row" + ); + let committed = projector.drain_unsynced_committed(); + let CommittedCellModel::Tool(tool) = &committed[0] else { + panic!("expected directly committed tool cell"); + }; + assert_eq!(tool.phase, ToolPhase::Completed); + assert_eq!(tool.tool_name.as_deref(), Some("bash")); + } + + #[test] + fn restored_committed_ids_block_late_rematerialization() { + let restored = CommittedCellModel::Tool(ToolModel { + tool_use_id: "bash-1".into(), + seq: 0, + phase: ToolPhase::Completed, + summary: String::new(), + tool_name: Some("bash".into()), + input: Some(serde_json::json!({"command": "ls"})), + input_partial_json: String::new(), + parsed_commands: Vec::new(), + exec_like: true, + start_time: None, + output_preview: String::new(), + output_delta_lines: Vec::new(), + file_changes: None, + command: Some("ls".into()), + command_source: None, + command_output: None, + command_duration: None, + tool_output: None, + tool_display_content: None, + is_error: false, + truncated: false, + }); + let mut projector = TranscriptProjector::default(); + projector.restore_committed(vec![restored]); + assert_eq!(projector.drain_unsynced_committed().len(), 1); + + projector.apply(ItemLifecycleEvent::ToolClosed { + tool_use_id: "bash-1".into(), + tool_name: "bash".into(), + input: serde_json::json!({"command": "ls"}), + output: Some(serde_json::json!("src\n")), + display_content: Some("src\n".into()), + file_changes: None, + is_error: false, + truncated: false, + }); + + assert!(projector.live_tool("bash-1").is_none()); + assert!( + projector.drain_unsynced_committed().is_empty(), + "late duplicate for a restored row must not append a second cell" + ); + } + #[test] fn file_change_closes_running_tool() { let mut projector = TranscriptProjector::default(); @@ -375,10 +741,129 @@ mod tests { truncated: false, }); + assert_eq!(projector.live_tools().count(), 1); + assert!(projector.committed.is_empty()); + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: TurnToolOutcome::Completed, + }); assert_eq!(projector.live_tools().count(), 0); assert_eq!(projector.committed.len(), 1); } + #[test] + fn duplicate_open_refreshes_one_tool_owner() { + let mut projector = TranscriptProjector::default(); + for input in [ + serde_json::Value::Null, + serde_json::json!({"command": "cargo check"}), + ] { + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "exec-1".into(), + tool_name: "exec_command".into(), + input, + command: Some("cargo check".into()), + command_source: None, + parsed_commands: Vec::new(), + }); + } + + let tools: Vec<_> = projector.live_tools().cloned().collect(); + assert_eq!(tools.len(), 1); + assert_eq!( + tools[0].input, + Some(serde_json::json!({"command": "cargo check"})) + ); + } + + #[test] + fn parallel_tools_complete_in_place_and_commit_in_open_order() { + let mut projector = TranscriptProjector::default(); + for id in ["first", "second"] { + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: id.into(), + tool_name: "exec_command".into(), + input: serde_json::json!({"command": id}), + command: Some(id.into()), + command_source: None, + parsed_commands: Vec::new(), + }); + } + for id in ["second", "first"] { + projector.apply(ItemLifecycleEvent::ToolClosed { + tool_use_id: id.into(), + tool_name: "exec_command".into(), + input: serde_json::Value::Null, + output: Some(serde_json::json!(id)), + display_content: Some(id.into()), + file_changes: None, + is_error: id == "second", + truncated: false, + }); + } + + let live: Vec<_> = projector.live_tools().cloned().collect(); + assert_eq!( + live.iter() + .map(|tool| (tool.tool_use_id.as_str(), tool.phase)) + .collect::>(), + vec![ + ("first", ToolPhase::Completed), + ("second", ToolPhase::Failed), + ] + ); + assert!(projector.drain_unsynced_committed().is_empty()); + + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: TurnToolOutcome::Completed, + }); + let committed = projector.drain_unsynced_committed(); + let ids: Vec<_> = committed + .iter() + .map(|cell| match cell { + CommittedCellModel::Tool(tool) => tool.tool_use_id.as_str(), + CommittedCellModel::Text(_) => panic!("expected tool"), + }) + .collect(); + assert_eq!(ids, vec!["first", "second"]); + } + + #[test] + fn successful_turn_without_tool_result_commits_degraded_row() { + let mut projector = TranscriptProjector::default(); + projector.apply(ItemLifecycleEvent::ToolOpened { + tool_use_id: "missing-result".into(), + tool_name: "exec_command".into(), + input: serde_json::json!({"command": "echo hi"}), + command: Some("echo hi".into()), + command_source: None, + parsed_commands: Vec::new(), + }); + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: TurnToolOutcome::Completed, + }); + + let committed = projector.drain_unsynced_committed(); + let CommittedCellModel::Tool(tool) = &committed[0] else { + panic!("expected tool"); + }; + assert_eq!(tool.phase, ToolPhase::Degraded); + let parts = crate::transcript::presentation::tool_title_parts( + tool.phase, + tool.tool_name.as_deref(), + tool.input.as_ref(), + &tool.parsed_commands, + false, + &tool.summary, + ); + let title = crate::transcript::presentation::tool_title_line(tool.phase, &parts); + let text = title + .spans + .iter() + .map(|span| span.content.as_ref()) + .collect::(); + assert_eq!(text, "Ran echo hi · result unavailable"); + } + #[test] fn text_delta_accepts_incremental_and_cumulative_chunks() { let mut projector = TranscriptProjector::default(); diff --git a/crates/tui/src/transcript/render.rs b/crates/tui/src/transcript/render.rs index 67838017..5b37f561 100644 --- a/crates/tui/src/transcript/render.rs +++ b/crates/tui/src/transcript/render.rs @@ -121,6 +121,7 @@ fn text_cell_to_history(text: &TextCellModel) -> Box { pub(crate) fn live_tool_display_lines( tool: &ToolCellModel, width: u16, + cwd: &Path, pending_dot_prefix: Line<'static>, tool_text_style: ratatui::style::Style, ) -> Vec> { @@ -128,6 +129,10 @@ pub(crate) fn live_tool_display_lines( if tool.phase == ToolPhase::Preparing { return vec![title_line]; } + if tool.phase.is_terminal() { + return tool_cell_to_history(tool, cwd, pending_dot_prefix, tool_text_style) + .display_lines(width); + } match (&tool.tool_name, &tool.input) { (Some(tool_name), Some(input)) => ToolIoCell::from_text_output( ToolIoCellOptions { @@ -142,12 +147,58 @@ pub(crate) fn live_tool_display_lines( tool.output_preview.clone(), ) .display_lines(width), - _ => history_cell::AgentMessageCell::new_with_prefix( - vec![title_line], - pending_dot_prefix, - " ", - false, + _ => { + let mut lines = vec![title_line]; + lines.extend( + tool.output_delta_lines + .iter() + .map(|line| Line::from(line.clone())), + ); + history_cell::AgentMessageCell::new_with_prefix(lines, pending_dot_prefix, " ", false) + .display_lines(width) + } + } +} + +/// Transcript-overlay rendering for a tool still owned by the current turn. +pub(crate) fn live_tool_transcript_lines( + tool: &ToolCellModel, + width: u16, + cwd: &Path, + pending_dot_prefix: Line<'static>, + tool_text_style: ratatui::style::Style, +) -> Vec> { + if tool.phase.is_terminal() { + return tool_cell_to_history(tool, cwd, pending_dot_prefix, tool_text_style) + .transcript_lines(width); + } + let title_line = tool_title_for_cell(tool); + if tool.phase == ToolPhase::Preparing { + return vec![title_line]; + } + match (&tool.tool_name, &tool.input) { + (Some(tool_name), Some(input)) => ToolIoCell::from_text_output( + ToolIoCellOptions { + title_line: Some(title_line), + dot_prefix: pending_dot_prefix, + subsequent_prefix: " ".into(), + output_style: tool_text_style, + show_empty_ellipsis: false, + }, + tool_name.clone(), + input.clone(), + tool.output_preview.clone(), ) - .display_lines(width), + .transcript_lines(width), + _ => { + let mut lines = vec![title_line]; + lines.extend( + tool.output_delta_lines + .iter() + .map(|line| Line::from(line.clone())), + ); + history_cell::AgentMessageCell::new_with_prefix(lines, pending_dot_prefix, " ", false) + .transcript_lines(width) + } } } diff --git a/crates/tui/src/transcript/tool_state.rs b/crates/tui/src/transcript/tool_state.rs index ebebc88b..e1506997 100644 --- a/crates/tui/src/transcript/tool_state.rs +++ b/crates/tui/src/transcript/tool_state.rs @@ -58,3 +58,248 @@ pub(crate) fn command_source_from_tool_name(tool_name: &str) -> Option None, } } + +/// Extracts the members of a partially streamed tool-call JSON object. +/// +/// Providers stream tool arguments as a JSON string in fragments; waiting for +/// the complete parse leaves the running row without any parameters for the +/// whole streaming window. This scanner collects only members whose value has +/// fully arrived (closing quote seen, or a scalar followed by a separator) and +/// skips nested objects/arrays — display fields (command, filePath, pattern, +/// …) are always flat strings or numbers. The result is display-only: the +/// authoritative input arrives with the item refresh or the tool result. +pub(crate) fn partial_object_members(partial: &str) -> Option { + let mut map = serde_json::Map::new(); + let mut chars = partial.char_indices().peekable(); + + // Consume the opening brace; anything else is not a tool-argument object. + skip_json_whitespace(&mut chars); + if chars.peek().map(|(_, ch)| *ch) != Some('{') { + return None; + } + chars.next(); + + loop { + skip_json_whitespace(&mut chars); + match chars.peek() { + None => break, + Some((_, '}')) => break, + Some((_, ',')) => { + chars.next(); + continue; + } + Some((_, '"')) => {} + Some(&(_, _)) => break, + } + // Key string. + chars.next(); + let Some(key) = scan_json_string(&mut chars) else { + break; + }; + skip_json_whitespace(&mut chars); + if chars.next().map(|(_, ch)| ch) != Some(':') { + break; + } + skip_json_whitespace(&mut chars); + let Some((_, value_start)) = chars.peek().copied() else { + break; + }; + match value_start { + '"' => { + chars.next(); + let Some(value) = scan_json_string(&mut chars) else { + break; + }; + map.insert(key, serde_json::Value::String(value)); + } + '{' | '[' => { + // Nested values are not display fields; skip to their close. + let (open, close) = if value_start == '{' { + ('{', '}') + } else { + ('[', ']') + }; + chars.next(); + if !skip_nested_value(&mut chars, open, close) { + break; + } + } + _ => { + // Scalar (number/bool/null): only complete when a separator + // follows — a trailing partial number must not be recorded. + let mut literal = String::new(); + let mut complete = false; + while let Some(&(_, ch)) = chars.peek() { + if ch == ',' || ch == '}' { + complete = true; + break; + } + literal.push(ch); + chars.next(); + } + if !complete { + break; + } + let literal = literal.trim().to_string(); + let value = if literal == "true" { + serde_json::Value::Bool(true) + } else if literal == "false" { + serde_json::Value::Bool(false) + } else if literal == "null" { + serde_json::Value::Null + } else { + match literal.parse::() { + Ok(number) => serde_json::Value::from(number), + Err(_) => break, + } + }; + map.insert(key, value); + } + } + } + + (!map.is_empty()).then_some(serde_json::Value::Object(map)) +} + +/// Reads a JSON string body starting just after the opening quote; returns +/// `None` when the closing quote has not arrived yet. +fn scan_json_string(chars: &mut std::iter::Peekable>) -> Option { + let mut value = String::new(); + while let Some((_, ch)) = chars.next() { + match ch { + '"' => return Some(value), + '\\' => match chars.next() { + Some((_, '"')) => value.push('"'), + Some((_, '\\')) => value.push('\\'), + Some((_, '/')) => value.push('/'), + Some((_, 'n')) => value.push('\n'), + Some((_, 't')) => value.push('\t'), + Some((_, 'r')) => value.push('\r'), + Some((_, 'b')) => value.push('\u{8}'), + Some((_, 'f')) => value.push('\u{c}'), + Some((_, 'u')) => { + let mut code = String::new(); + for _ in 0..4 { + match chars.next() { + Some((_, hex)) => code.push(hex), + None => return None, + } + } + match u32::from_str_radix(&code, 16).ok().and_then(char::from_u32) { + Some(decoded) => value.push(decoded), + None => return None, + } + } + _ => return None, + }, + _ => value.push(ch), + } + } + None +} + +/// Skips a nested object/array value; returns `false` when it is still +/// incomplete. +fn skip_nested_value( + chars: &mut std::iter::Peekable>, + open: char, + close: char, +) -> bool { + let mut depth = 1usize; + while let Some(&(_, ch)) = chars.peek() { + if ch == '"' { + chars.next(); + if scan_json_string(chars).is_none() { + return false; + } + continue; + } + chars.next(); + if ch == open { + depth += 1; + } else if ch == close { + depth -= 1; + if depth == 0 { + return true; + } + } + } + false +} + +fn skip_json_whitespace(chars: &mut std::iter::Peekable>) { + while matches!( + chars.peek().map(|(_, ch)| *ch), + Some(' ' | '\t' | '\n' | '\r') + ) { + chars.next(); + } +} + +#[cfg(test)] +mod tests { + use pretty_assertions::assert_eq; + + use super::*; + + #[test] + fn partial_members_collect_completed_string_values_only() { + let partial = r#"{"filePath": "src/lib.rs", "offset": 12"#; + assert_eq!( + partial_object_members(partial), + Some(serde_json::json!({ "filePath": "src/lib.rs" })), + ); + } + + #[test] + fn partial_members_skip_incomplete_trailing_string() { + let partial = r#"{"pattern": "fn main", "path": "crate"#; + assert_eq!( + partial_object_members(partial), + Some(serde_json::json!({ "pattern": "fn main" })), + ); + } + + #[test] + fn partial_members_record_number_only_after_separator() { + // The trailing `40` has no separator yet — it may still grow, so only + // the offset is displayable. + let streaming = r#"{"offset": 12, "limit": 40"#; + assert_eq!( + partial_object_members(streaming), + Some(serde_json::json!({ "offset": 12.0 })), + ); + // With a separator after it, both members are complete. + let complete = r#"{"offset": 12, "limit": 40,"#; + assert_eq!( + partial_object_members(complete), + Some(serde_json::json!({ "offset": 12.0, "limit": 40.0 })), + ); + // The trailing number may still grow — must not be recorded. + let growing = r#"{"offset": 1"#; + assert_eq!(partial_object_members(growing), None); + } + + #[test] + fn partial_members_skip_nested_values() { + let partial = r#"{"filePath": "a.rs", "edits": [{"old": "x""#; + assert_eq!( + partial_object_members(partial), + Some(serde_json::json!({ "filePath": "a.rs" })), + ); + } + + #[test] + fn partial_members_handle_escapes_and_empty_input() { + let partial = r#"{"command": "echo \"hi\"", "description": "Say hi""#; + assert_eq!( + partial_object_members(partial), + Some(serde_json::json!({ + "command": "echo \"hi\"", + "description": "Say hi", + })), + ); + assert_eq!(partial_object_members(""), None); + assert_eq!(partial_object_members(r#"{ "filePa"#), None); + } +} diff --git a/crates/tui/src/tui.rs b/crates/tui/src/tui.rs index f18c9013..8aa7dbec 100644 --- a/crates/tui/src/tui.rs +++ b/crates/tui/src/tui.rs @@ -111,6 +111,12 @@ where terminal.clear_screen_area(previous_area)?; } terminal.set_viewport_area(area); + // Repaints are diff-based: `set_viewport_area` resizes the diff buffers by + // row-major index, so after a rect change they no longer describe the + // physical rows. Cells the diff believes unchanged are never rewritten, + // leaving stale fragments (e.g. the previous frame's composer chrome) + // visible inside the live area. Wipe from the new origin downward — this + // never touches scrollback above the viewport — and force a full repaint. terminal.clear()?; terminal.invalidate_viewport(); Ok(()) @@ -533,8 +539,8 @@ impl Tui { /// /// shutdown_terminal_safe() /// 1. leave alt-screen if one is active - /// 2. drop pending history that was never rendered - /// 3. clear from the viewport origin downward + /// 2. flush pending completed history + /// 3. clear only the final live viewport /// 4. outer restore guard restores terminal modes /// /// after exit @@ -546,12 +552,40 @@ impl Tui { /// The key idea is to avoid bespoke cursor choreography on exit. Clearing the /// active viewport and then restoring terminal modes is more robust across /// terminals like Terminal.app than trying to place the shell prompt ourselves. - pub(crate) fn shutdown_terminal_safe(&mut self) -> Result<()> { + pub(crate) fn shutdown_terminal_safe(&mut self, final_live_height: u16) -> Result<()> { if self.is_alt_screen_active() { self.leave_alt_screen()?; } - self.clear_pending_history_lines(); - self.terminal.clear()?; + Self::finalize_inline_viewport( + &mut self.terminal, + &mut self.pending_history_lines, + final_live_height, + self.is_zellij, + ) + } + + fn finalize_inline_viewport( + terminal: &mut CustomTerminal, + pending_history_lines: &mut Vec, + final_live_height: u16, + is_zellij: bool, + ) -> Result<()> + where + B: Backend + std::io::Write, + { + let previous_area = terminal.viewport_area; + let mut needs_full_repaint = + Self::update_inline_viewport(terminal, final_live_height, is_zellij)?; + needs_full_repaint |= + Self::flush_pending_history_lines(terminal, pending_history_lines, is_zellij)?; + needs_full_repaint |= Self::clear_vacated_viewport_tail(terminal, previous_area)?; + if needs_full_repaint { + terminal.invalidate_viewport(); + } + let final_area = terminal.viewport_area; + terminal.clear_screen_area(final_area)?; + terminal.set_cursor_position(final_area.as_position())?; + std::io::Write::flush(terminal.backend_mut())?; Ok(()) } @@ -590,11 +624,14 @@ impl Tui { /// the viewport would extend past the bottom of the screen. Returns `true` when /// the caller must invalidate the diff buffer (Zellij mode), because the scroll /// was performed with raw newlines that ratatui cannot track. - fn update_inline_viewport( - terminal: &mut Terminal, + fn update_inline_viewport( + terminal: &mut CustomTerminal, height: u16, is_zellij: bool, - ) -> Result { + ) -> Result + where + B: Backend + std::io::Write, + { let size = terminal.size()?; let (area, scroll_by) = next_inline_viewport_area(terminal.viewport_area, size, height); let mut needs_full_repaint = false; @@ -616,12 +653,15 @@ impl Tui { /// This matches append-only inline TUI behavior: when the live area needs /// more height, we advance the terminal buffer downward so users who are currently viewing /// scrollback do not see previously rendered rows get rewritten in place. - fn append_expanded_viewport( - terminal: &mut Terminal, + fn append_expanded_viewport( + terminal: &mut CustomTerminal, size: Size, scroll_by: u16, is_zellij: bool, - ) -> Result<()> { + ) -> Result<()> + where + B: Backend + std::io::Write, + { if is_zellij { return Self::scroll_zellij_expanded_viewport(terminal, size, scroll_by); } @@ -638,11 +678,14 @@ impl Tui { /// Push content above the viewport upward by `scroll_by` rows using raw /// newlines at the screen bottom. This is the Zellij-safe alternative to /// backend `append_lines`, which Zellij does not expose in a way ratatui can rely on. - fn scroll_zellij_expanded_viewport( - terminal: &mut Terminal, + fn scroll_zellij_expanded_viewport( + terminal: &mut CustomTerminal, size: Size, scroll_by: u16, - ) -> Result<()> { + ) -> Result<()> + where + B: Backend + std::io::Write, + { crossterm::queue!( terminal.backend_mut(), crossterm::cursor::MoveTo(0, size.height.saturating_sub(1)) @@ -656,16 +699,19 @@ impl Tui { /// Write any buffered history lines above the viewport and clear the buffer. /// Returns `true` when Zellij mode was used, signaling that the caller must /// invalidate the diff buffer for a full repaint. - fn flush_pending_history_lines( - terminal: &mut Terminal, + fn flush_pending_history_lines( + terminal: &mut CustomTerminal, pending_history_lines: &mut Vec, is_zellij: bool, - ) -> Result { + ) -> Result + where + B: Backend + std::io::Write, + { if pending_history_lines.is_empty() { return Ok(false); } - crate::insert_history::insert_history_lines_with_mode( + let _outcome = crate::insert_history::insert_history_lines_with_mode( terminal, pending_history_lines.clone(), crate::insert_history::InsertHistoryMode::new(is_zellij), @@ -674,6 +720,28 @@ impl Tui { Ok(is_zellij) } + fn clear_vacated_viewport_tail( + terminal: &mut CustomTerminal, + previous_area: Rect, + ) -> Result + where + B: Backend + std::io::Write, + { + let final_area = terminal.viewport_area; + let stale_top = final_area.bottom().max(previous_area.y); + let stale_bottom = previous_area.bottom(); + if stale_top >= stale_bottom { + return Ok(false); + } + terminal.clear_screen_area(Rect::new( + 0, + stale_top, + final_area.width, + stale_bottom - stale_top, + ))?; + Ok(true) + } + pub fn draw( &mut self, height: u16, @@ -697,6 +765,7 @@ impl Tui { } let terminal = &mut self.terminal; + let previous_area = terminal.viewport_area; if let Some(new_area) = pending_viewport_area.take() { apply_inline_viewport_area_change(terminal, new_area)?; } @@ -712,6 +781,7 @@ impl Tui { &mut self.pending_history_lines, self.is_zellij, )?; + needs_full_repaint |= Self::clear_vacated_viewport_tail(terminal, previous_area)?; if needs_full_repaint { terminal.invalidate_viewport(); @@ -743,16 +813,15 @@ impl Tui { /// Returns `(area, scroll_by)`. When `scroll_by > 0`, the caller must append that /// many rows before applying `area` so growth remains append-only. /// -/// Shrinks keep `area.y` unchanged. Pinning the bottom on shrink would move `y` -/// downward and clear the vacated rows (see `apply_inline_viewport_area_change`), -/// which accumulates blank gaps after tall bottom-pane views such as `/model`. +/// Shrinks always keep the old top. Pending history insertion is responsible +/// for moving the viewport into space released below it. fn next_inline_viewport_area(previous: Rect, size: Size, height: u16) -> (Rect, u16) { let mut area = previous; area.height = height.min(size.height); area.width = size.width; if area.bottom() > size.height { let scroll_by = area.bottom() - size.height; - area.y = size.height - area.height; + area.y = size.height.saturating_sub(area.height); (area, scroll_by) } else { (area, 0) @@ -797,9 +866,7 @@ mod tests { } #[test] - fn next_inline_viewport_area_shrink_keeps_top_to_avoid_blank_gaps() { - // Bottom-aligned full-height viewport shrinking (e.g. closing /model) must - // keep y stable. Moving y down would clear the vacated rows into blank gaps. + fn next_inline_viewport_area_shrink_while_pinned_keeps_top() { let previous = Rect::new(0, 1, 80, 39); let size = Size::new(80, 40); let (area, scroll_by) = next_inline_viewport_area(previous, size, 5); @@ -807,6 +874,17 @@ mod tests { assert_eq!(area, Rect::new(0, 1, 80, 5)); } + #[test] + fn next_inline_viewport_area_mid_screen_shrink_keeps_top() { + // A viewport that has not reached the screen bottom yet keeps its top + // while shrinking; only genuinely vacated rows below are cleared. + let previous = Rect::new(0, 10, 80, 10); + let size = Size::new(80, 40); + let (area, scroll_by) = next_inline_viewport_area(previous, size, 5); + assert_eq!(scroll_by, 0); + assert_eq!(area, Rect::new(0, 10, 80, 5)); + } + #[test] fn next_inline_viewport_area_preserves_mid_screen_growth_without_overflow() { let previous = Rect::new(0, 10, 80, 5); @@ -817,7 +895,7 @@ mod tests { } #[test] - fn apply_inline_viewport_area_change_clears_from_new_viewport_when_old_area_is_empty() { + fn apply_inline_viewport_area_change_preserves_scrollback_and_wipes_live_rows() { let width: u16 = 24; let height: u16 = 6; let backend = VT100Backend::new(width, height); @@ -836,14 +914,18 @@ mod tests { rows_after[0].contains("shell line"), "expected content above viewport to remain visible, rows: {rows_after:?}" ); + // The live region is physically wiped on a rect change so the next + // diff-based repaint cannot leave stale fragments (e.g. composer + // chrome) on rows the diff believes are unchanged. assert!( - rows_after.iter().skip(1).all(|row| !row.contains("stale")), - "expected stale cells in new viewport to be cleared, rows: {rows_after:?}" + rows_after[1..].iter().all(|row| row.trim().is_empty()), + "expected live viewport rows to be wiped, rows: {rows_after:?}" ); + assert_eq!(Rect::new(0, 1, width, height - 1), terminal.viewport_area); } #[test] - fn apply_inline_viewport_area_change_clears_previous_viewport_rows() { + fn moving_viewport_down_preserves_rows_above_previous_origin() { let width: u16 = 24; let height: u16 = 6; let backend = VT100Backend::new(width, height); @@ -863,16 +945,94 @@ mod tests { rows_after[0].contains("history") && rows_after[1].contains("keep"), "expected content above previous viewport to remain visible, rows: {rows_after:?}" ); + // Both the vacated previous viewport rows and the new live region are + // wiped; history insertion owns the vacated rows in the real flow. assert!( - rows_after - .iter() - .skip(2) - .all(|row| !row.contains("old") && !row.contains("stale")), - "expected previous and new viewport rows to be cleared, rows: {rows_after:?}" + rows_after[2..].iter().all(|row| row.trim().is_empty()), + "expected previous and new viewport rows to be wiped, rows: {rows_after:?}" ); assert_eq!(Rect::new(0, 3, width, 2), terminal.viewport_area); } + #[test] + fn shrink_then_history_insert_preserves_existing_transcript() { + let width = 30; + let height = 8; + let backend = VT100Backend::new(width, height); + let mut terminal = CustomTerminal::with_options(backend).expect("terminal"); + write!( + terminal.backend_mut(), + "stable transcript\r\nrunning 1\r\nrunning 2\r\nrunning 3" + ) + .expect("prefill terminal"); + let previous = Rect::new(0, 1, width, height - 1); + terminal.set_viewport_area(previous); + + apply_inline_viewport_area_change(&mut terminal, Rect::new(0, 1, width, 3)) + .expect("logical shrink"); + let outcome = insert_history_lines( + &mut terminal, + vec![ + Line::from("Ran first").into(), + Line::from("Ran second").into(), + Line::from("Ran third").into(), + ], + ); + outcome.expect("insert committed tools"); + let final_area = terminal.viewport_area; + terminal + .clear_screen_area(Rect::new( + 0, + final_area.bottom(), + width, + previous.bottom().saturating_sub(final_area.bottom()), + )) + .expect("clear stale tail"); + + let rows: Vec = terminal.backend().vt100().screen().rows(0, width).collect(); + assert!(rows[0].contains("stable transcript"), "rows: {rows:?}"); + assert!(rows.iter().any(|row| row.contains("Ran first"))); + assert!(rows.iter().any(|row| row.contains("Ran third"))); + assert_eq!(final_area, Rect::new(0, 4, width, 3)); + } + + #[test] + fn exit_finalization_flushes_history_and_clears_only_live_rows() { + let width = 30; + let height = 8; + let backend = VT100Backend::new(width, height); + let mut terminal = CustomTerminal::with_options(backend).expect("terminal"); + write!( + terminal.backend_mut(), + "stable transcript\r\nRunning first\r\ncomposer\r\nstatus" + ) + .expect("prefill terminal"); + terminal.set_viewport_area(Rect::new(0, 1, width, height - 1)); + let mut pending_history = vec![ + ScrollbackLine::from(Line::from("Ran first")), + ScrollbackLine::from(Line::from("assistant reply")), + ]; + + Tui::finalize_inline_viewport(&mut terminal, &mut pending_history, 3, false) + .expect("finalize inline viewport"); + + let rows: Vec = terminal.backend().vt100().screen().rows(0, width).collect(); + assert!(pending_history.is_empty()); + assert!(rows[0].contains("stable transcript"), "rows: {rows:?}"); + assert!(rows.iter().any(|row| row.contains("Ran first"))); + assert!(rows.iter().any(|row| row.contains("assistant reply"))); + assert!( + rows.iter() + .skip(terminal.viewport_area.top() as usize) + .all(|row| row.trim().is_empty()), + "final live rows should be empty: {rows:?}" + ); + assert_eq!( + terminal.last_known_cursor_pos, + terminal.viewport_area.as_position() + ); + } + #[test] fn reset_inline_session_ui_clears_pending_history_and_visible_transcript() { let width: u16 = 24; diff --git a/crates/tui/src/worker.rs b/crates/tui/src/worker.rs index 49736b17..94659b5e 100644 --- a/crates/tui/src/worker.rs +++ b/crates/tui/src/worker.rs @@ -123,6 +123,68 @@ fn should_apply_terminal_turn_usage_fallback( !saw_usage_update_for_turn && !has_authoritative_usage_totals } +async fn reconcile_idle_turn( + client: &mut StdioServerClient, + session_id: SessionId, + turn_id: TurnId, + seen_terminal_item_ids: &mut HashSet, + seen_terminal_call_ids: &mut HashSet, + event_tx: &mpsc::UnboundedSender, +) -> Result { + let turn = client.turn_read_native(session_id, turn_id).await?.turn; + if turn.status == devo_protocol::native::turn::TurnStatus::InProgress { + anyhow::bail!("server still reports turn {turn_id} in progress"); + } + + let mut cursor = None; + loop { + let page = client + .turn_items_list_native(session_id, turn_id, cursor.clone(), Some(200)) + .await?; + let page_len = page.data.len(); + let next_cursor = page.next_cursor; + for envelope in page.data { + if !matches!( + envelope.state, + devo_protocol::native::item::ItemState::Completed + | devo_protocol::native::item::ItemState::Failed + | devo_protocol::native::item::ItemState::Interrupted + | devo_protocol::native::item::ItemState::Lost + ) { + continue; + } + let call_id = match &envelope.item { + devo_protocol::native::item::Item::ToolResult { call_id, .. } + | devo_protocol::native::item::Item::CommandExecution { call_id, .. } + | devo_protocol::native::item::Item::FileChange { call_id, .. } => { + Some(call_id.clone()) + } + _ => None, + }; + let Some(call_id) = call_id else { + continue; + }; + let item_id = envelope.id.to_string(); + if seen_terminal_item_ids.contains(&item_id) + || seen_terminal_call_ids.contains(&call_id) + { + continue; + } + let legacy_item_id = devo_core::ItemId::try_from(item_id.as_str())?; + for event in native_items::completed_events(&envelope.item, legacy_item_id) { + let _ = event_tx.send(WorkerEvent::Transcript(event)); + } + seen_terminal_item_ids.insert(item_id); + seen_terminal_call_ids.insert(call_id); + } + match (next_cursor, page_len) { + (Some(next), len) if len > 0 => cursor = Some(next), + _ => break, + } + } + Ok(turn) +} + /// Spawn discovery from a typed `item/completed` ToolResult (L2-DES-APP-009 /// cutover): the typed item carries the same raw output the ACP tool-call /// path parsed, so discovery no longer depends on the ACP envelope. @@ -1066,6 +1128,8 @@ async fn run_worker_inner( let mut saw_usage_update_for_turn = false; let mut has_authoritative_usage_totals = false; let mut latest_completed_agent_message: Option = None; + let mut seen_terminal_item_ids: HashSet = HashSet::new(); + let mut seen_terminal_call_ids: HashSet = HashSet::new(); let mut child_agent_sessions: HashSet = HashSet::new(); let mut btw_agent_sessions: HashMap = HashMap::new(); let mut input_history_cursor: Option = None; @@ -3192,6 +3256,8 @@ async fn run_worker_inner( continue; } active_turn_id = Some(turn_id); + seen_terminal_item_ids.clear(); + seen_terminal_call_ids.clear(); saw_usage_update_for_turn = false; model = turn.model.model.clone(); model_binding_id = @@ -3506,13 +3572,12 @@ async fn run_worker_inner( let changed_session_id = params["sessionId"] .as_str() .and_then(|id| SessionId::try_from(id).ok()); + let status = params["status"].as_str().unwrap_or("idle"); if let Some(changed_session_id) = changed_session_id && child_agent_sessions.contains(&changed_session_id) { - let status = match params["status"].as_str() { - Some("active") => { - devo_protocol::SessionRuntimeStatus::ActiveTurn - } + let status = match status { + "active" => devo_protocol::SessionRuntimeStatus::ActiveTurn, _ => devo_protocol::SessionRuntimeStatus::Idle, }; let _ = event_tx.send(WorkerEvent::SubagentMonitor { @@ -3521,6 +3586,122 @@ async fn run_worker_inner( status, }, }); + } else if changed_session_id.is_some_and(|id| Some(id) == session_id) + && status != "active" + && let (Some(active_session_id), Some(finished_turn_id)) = + (session_id, active_turn_id) + { + tracing::warn!( + turn_id = %finished_turn_id, + "turn/completed not observed; reconciling authoritative turn state" + ); + match reconcile_idle_turn( + &mut client, + active_session_id, + finished_turn_id, + &mut seen_terminal_item_ids, + &mut seen_terminal_call_ids, + event_tx, + ) + .await + { + Ok(turn) => { + active_turn_id = None; + if matches!( + turn.status, + devo_protocol::native::turn::TurnStatus::Completed + | devo_protocol::native::turn::TurnStatus::Interrupted + ) { + turn_count += 1; + } + if let Some(usage) = &turn.usage { + let input = usage.query.input_tokens as usize; + let total = usage.query.total_tokens as usize; + let cache_read = usage.query.cache_read_input_tokens as usize; + if !saw_usage_update_for_turn { + last_query_input_tokens = input; + last_query_total_tokens = total; + } + if should_apply_terminal_turn_usage_fallback( + saw_usage_update_for_turn, + has_authoritative_usage_totals, + ) { + total_input_tokens += input; + total_output_tokens += + usage.query.output_tokens as usize; + total_tokens += total; + total_cache_read_tokens += cache_read; + } + } + if let Some(usage) = &turn.usage { + let input = usage.query.input_tokens as usize; + let total = usage.query.total_tokens as usize; + let cache_read = usage.query.cache_read_input_tokens as usize; + if !saw_usage_update_for_turn { + last_query_input_tokens = input; + last_query_total_tokens = total; + } + if should_apply_terminal_turn_usage_fallback( + saw_usage_update_for_turn, + has_authoritative_usage_totals, + ) { + total_input_tokens += input; + total_output_tokens += + usage.query.output_tokens as usize; + total_tokens += total; + total_cache_read_tokens += cache_read; + } + } + let prompt_token_estimate = turn + .usage + .as_ref() + .map(|usage| usage.query.input_tokens as usize) + .unwrap_or(total_input_tokens); + if turn.status + == devo_protocol::native::turn::TurnStatus::Failed + { + let message = turn + .error + .as_ref() + .map(|error| error.message.clone()) + .unwrap_or_else(|| "Turn failed".to_string()); + let _ = event_tx.send(WorkerEvent::TurnFailed { + hint: devo_provider::recovery_hint_for_message(&message), + message, + turn_count, + total_input_tokens, + total_output_tokens, + total_tokens, + total_cache_read_tokens, + prompt_token_estimate, + last_query_input_tokens, + }); + } else { + let _ = event_tx.send(WorkerEvent::TurnFinished { + stop_reason: format!("{:?}", turn.status), + turn_count, + total_input_tokens, + total_output_tokens, + total_tokens, + total_cache_read_tokens, + last_query_total_tokens, + last_query_input_tokens, + prompt_token_estimate, + }); + } + latest_completed_agent_message = None; + } + Err(error) => { + tracing::warn!( + turn_id = %finished_turn_id, + %error, + "idle turn reconciliation failed; preserving live tool state" + ); + let _ = event_tx.send(WorkerEvent::InterruptFailed { + message: "Turn ended, but final tool results are unavailable; preserving live state".to_string(), + }); + } + } } continue; } @@ -3565,10 +3746,11 @@ async fn run_worker_inner( continue; } "item/started" | "item/completed" => { - if let Ok(payload) = serde_json::from_value::< + match serde_json::from_value::< devo_protocol::TypedItemEventPayload, >(params.clone()) { + Ok(payload) => { // Child-session items belong to // the subagent monitor, not the // main transcript (L2-DES-APP-009). @@ -3576,6 +3758,32 @@ async fn run_worker_inner( SessionId::try_from(payload.item.session_id.as_str()) .ok(); if item_session_id == session_id { + if method == "item/completed" { + let item_id = payload.item.id.to_string(); + if !seen_terminal_item_ids.insert(item_id) { + continue; + } + let call_id = match &payload.item.item { + devo_protocol::native::item::Item::ToolResult { + call_id, + .. + } + | devo_protocol::native::item::Item::CommandExecution { + call_id, + .. + } + | devo_protocol::native::item::Item::FileChange { + call_id, + .. + } => Some(call_id.clone()), + _ => None, + }; + if let Some(call_id) = call_id + && !seen_terminal_call_ids.insert(call_id) + { + continue; + } + } if method == "item/completed" && let devo_protocol::native::item::Item::AssistantMessage { text, @@ -3616,13 +3824,26 @@ async fn run_worker_inner( ) .await; } - item_dispatch::dispatch_typed_item_lifecycle( - &method, - &payload, - devo_core::ItemId::try_from(payload.item.id.as_str()) - .expect("typed item id"), - event_tx, - ); + match devo_core::ItemId::try_from( + payload.item.id.as_str(), + ) { + Ok(item_id) => { + item_dispatch::dispatch_typed_item_lifecycle( + &method, &payload, item_id, event_tx, + ); + } + Err(error) => { + // A malformed id must not take the + // whole worker down: the row is + // skipped and the turn continues. + tracing::warn!( + method = %method, + item_id = %payload.item.id, + %error, + "dropping typed item with unparseable id" + ); + } + } } else if let Some(child_id) = item_session_id && child_agent_sessions.contains(&child_id) { @@ -3633,6 +3854,16 @@ async fn run_worker_inner( let _ = event_tx.send(event); } } + } + Err(error) => { + // Schema drift or a foreign payload must not + // silently swallow tool completions. + tracing::warn!( + method = %method, + %error, + "failed to decode typed item event" + ); + } } continue; } diff --git a/crates/tui/src/worker/native_items.rs b/crates/tui/src/worker/native_items.rs index aea4e204..fd38b822 100644 --- a/crates/tui/src/worker/native_items.rs +++ b/crates/tui/src/worker/native_items.rs @@ -149,3 +149,107 @@ pub(crate) fn completed_events(item: &Item, item_id: ItemId) -> Vec Vec::new(), } } + +#[cfg(test)] +mod tests { + use pretty_assertions::assert_eq; + + use super::*; + use crate::transcript::model::CommittedCellModel; + use crate::transcript::model::ToolPhase; + use crate::transcript::presentation::tool_title_parts; + use crate::transcript::projector::TranscriptProjector; + use devo_core::ItemId; + use devo_protocol::native::item::ToolSource; + + fn tool_call_item(input: Option) -> Item { + Item::ToolCall { + call_id: "bash-1".to_string(), + tool_name: "bash".to_string(), + source: ToolSource::Builtin, + server_name: None, + input, + } + } + + /// A streamed tool call opens with empty parameters, the server then + /// re-broadcasts `item/started` with the complete input, and the result + /// closes the row. The running row must render the command as soon as the + /// refresh lands, and the completed row must keep it. + #[test] + fn refreshed_started_updates_running_tool_parameters() { + let item_id = ItemId::new(); + let mut projector = TranscriptProjector::default(); + + for event in started_events(&tool_call_item(Some(serde_json::json!({}))), item_id) { + projector.apply(event); + } + + let live = projector.live_tool("bash-1").expect("running tool row"); + assert_eq!(live.phase, ToolPhase::Running); + assert!(live.command.is_none()); + + for event in started_events( + &tool_call_item(Some(serde_json::json!({ "command": "cargo test" }))), + item_id, + ) { + projector.apply(event); + } + + let live = projector + .live_tool("bash-1") + .expect("refreshed running tool row"); + assert_eq!( + live.phase, + ToolPhase::Running, + "refresh must not flip the running phase" + ); + assert_eq!(live.command.as_deref(), Some("cargo test")); + + let parts = tool_title_parts( + live.phase, + live.tool_name.as_deref(), + live.input.as_ref(), + &live.parsed_commands, + false, + live.summary.as_str(), + ); + assert_eq!(parts.verb, "Running"); + assert_eq!(parts.detail, "cargo test"); + + for event in completed_events( + &Item::ToolResult { + call_id: "bash-1".to_string(), + output: serde_json::Value::String("ok".to_string()), + display_content: Some("ok".to_string()), + is_error: false, + truncated: false, + }, + item_id, + ) { + projector.apply(event); + } + + let completed_live = projector.live_tool("bash-1").expect("completed live row"); + assert_eq!(completed_live.phase, ToolPhase::Completed); + assert!(projector.drain_unsynced_committed().is_empty()); + projector.apply(ItemLifecycleEvent::TurnLiveToolsCleared { + outcome: crate::transcript::lifecycle::TurnToolOutcome::Completed, + }); + let committed = projector.drain_unsynced_committed(); + let CommittedCellModel::Tool(tool) = &committed[0] else { + panic!("expected committed tool cell"); + }; + assert_eq!(tool.phase, ToolPhase::Completed); + let parts = tool_title_parts( + tool.phase, + tool.tool_name.as_deref(), + tool.input.as_ref(), + &tool.parsed_commands, + false, + tool.summary.as_str(), + ); + assert_eq!(parts.verb, "Ran"); + assert_eq!(parts.detail, "cargo test"); + } +}