diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index 4c2fcdd5..38cf8d0f 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -16,6 +16,7 @@ use crate::agent::dispatcher::{ }; use crate::agent::session::{PendingApproval, Session, ThreadState}; use crate::agent::submission::SubmissionResult; +use crate::channels::web::util::truncate_preview; use crate::channels::{IncomingMessage, StatusUpdate}; use crate::context::JobContext; use crate::error::Error; @@ -69,6 +70,8 @@ impl Agent { .filter_map(|m| match m.role.as_str() { "user" => Some(ChatMessage::user(&m.content)), "assistant" => Some(ChatMessage::assistant(&m.content)), + // tool_calls rows are UI metadata (tool name + preview), + // not part of the LLM conversation context. _ => None, }) .collect(); @@ -315,6 +318,11 @@ impl Agent { }; thread.complete_turn(&response); + let tool_calls = thread + .turns + .last() + .map(|t| t.tool_calls.clone()) + .unwrap_or_default(); let _ = self .channels .send_status( @@ -324,7 +332,9 @@ impl Agent { ) .await; - // Persist assistant response (user message already persisted at turn start) + // Persist tool calls then assistant response (user message already persisted at turn start) + self.persist_tool_calls(thread_id, &message.user_id, &tool_calls) + .await; self.persist_assistant_response(thread_id, &message.user_id, &response) .await; @@ -423,6 +433,68 @@ impl Agent { } } + /// Persist tool call summaries to the DB as a `role="tool_calls"` message. + /// + /// Stored between the user and assistant messages so that + /// `build_turns_from_db_messages` can reconstruct the tool call history. + /// Content is a JSON array of tool call summaries. + pub(super) async fn persist_tool_calls( + &self, + thread_id: Uuid, + user_id: &str, + tool_calls: &[crate::agent::session::TurnToolCall], + ) { + if tool_calls.is_empty() { + return; + } + + let store = match self.store() { + Some(s) => Arc::clone(s), + None => return, + }; + + let summaries: Vec = tool_calls + .iter() + .map(|tc| { + let mut obj = serde_json::json!({ "name": tc.name }); + if let Some(ref result) = tc.result { + let preview = match result { + serde_json::Value::String(s) => truncate_preview(s, 500), + other => truncate_preview(&other.to_string(), 500), + }; + obj["result_preview"] = serde_json::Value::String(preview); + } + if let Some(ref error) = tc.error { + obj["error"] = serde_json::Value::String(truncate_preview(error, 200)); + } + obj + }) + .collect(); + + let content = match serde_json::to_string(&summaries) { + Ok(c) => c, + Err(e) => { + tracing::warn!("Failed to serialize tool calls: {}", e); + return; + } + }; + + if let Err(e) = store + .ensure_conversation(thread_id, "gateway", user_id, None) + .await + { + tracing::warn!("Failed to ensure conversation {}: {}", thread_id, e); + return; + } + + if let Err(e) = store + .add_conversation_message(thread_id, "tool_calls", &content) + .await + { + tracing::warn!("Failed to persist tool calls: {}", e); + } + } + pub(super) async fn process_undo( &self, session: Arc>, @@ -591,7 +663,13 @@ impl Agent { .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; if thread.state != ThreadState::AwaitingApproval { - return Ok(SubmissionResult::error("No pending approval request.")); + // Stale or duplicate approval (tool already executed) — silently ignore. + tracing::debug!( + %thread_id, + state = ?thread.state, + "Ignoring stale approval: thread not in AwaitingApproval state" + ); + return Ok(SubmissionResult::ok_with_message("")); } thread.take_pending_approval() @@ -599,7 +677,13 @@ impl Agent { let pending = match pending { Some(p) => p, - None => return Ok(SubmissionResult::error("No pending approval request.")), + None => { + tracing::debug!( + %thread_id, + "Ignoring stale approval: no pending approval found" + ); + return Ok(SubmissionResult::ok_with_message("")); + } }; // Verify request ID if provided @@ -1040,7 +1124,14 @@ impl Agent { match result { Ok(AgenticLoopResult::Response(response)) => { thread.complete_turn(&response); - // User message already persisted at turn start; save assistant response + let tool_calls = thread + .turns + .last() + .map(|t| t.tool_calls.clone()) + .unwrap_or_default(); + // User message already persisted at turn start; save tool calls then assistant response + self.persist_tool_calls(thread_id, &message.user_id, &tool_calls) + .await; self.persist_assistant_response(thread_id, &message.user_id, &response) .await; let _ = self diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 753b6d99..6cab65e2 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -14,6 +14,7 @@ use uuid::Uuid; use crate::channels::IncomingMessage; use crate::channels::web::server::GatewayState; use crate::channels::web::types::*; +use crate::channels::web::util::{build_turns_from_db_messages, truncate_preview}; pub async fn chat_send_handler( State(state): State>, @@ -317,12 +318,13 @@ pub async fn chat_history_handler( turns, has_more, oldest_timestamp, + pending_approval: None, })); } // Try in-memory first (freshest data for active threads) if let Some(thread) = sess.threads.get(&thread_id) - && !thread.turns.is_empty() + && (!thread.turns.is_empty() || thread.pending_approval.is_some()) { let turns: Vec = thread .turns @@ -341,16 +343,35 @@ pub async fn chat_history_handler( name: tc.name.clone(), has_result: tc.result.is_some(), has_error: tc.error.is_some(), + result_preview: tc.result.as_ref().map(|r| { + let s = match r { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + truncate_preview(&s, 500) + }), + error: tc.error.clone(), }) .collect(), }) .collect(); + let pending_approval = thread + .pending_approval + .as_ref() + .map(|pa| PendingApprovalInfo { + request_id: pa.request_id.to_string(), + tool_name: pa.tool_name.clone(), + description: pa.description.clone(), + parameters: serde_json::to_string_pretty(&pa.parameters).unwrap_or_default(), + }); + return Ok(Json(HistoryResponse { thread_id, turns, has_more: false, oldest_timestamp: None, + pending_approval, })); } @@ -369,6 +390,7 @@ pub async fn chat_history_handler( turns, has_more, oldest_timestamp, + pending_approval: None, })); } } @@ -379,51 +401,10 @@ pub async fn chat_history_handler( turns: Vec::new(), has_more: false, oldest_timestamp: None, + pending_approval: None, })) } -/// Build TurnInfo pairs from flat DB messages (alternating user/assistant). -pub fn build_turns_from_db_messages( - messages: &[crate::history::ConversationMessage], -) -> Vec { - let mut turns = Vec::new(); - let mut turn_number = 0; - let mut iter = messages.iter().peekable(); - - while let Some(msg) = iter.next() { - if msg.role == "user" { - let mut turn = TurnInfo { - turn_number, - user_input: msg.content.clone(), - response: None, - state: "Completed".to_string(), - started_at: msg.created_at.to_rfc3339(), - completed_at: None, - tool_calls: Vec::new(), - }; - - // Check if next message is an assistant response - if let Some(next) = iter.peek() - && next.role == "assistant" - { - let assistant_msg = iter.next().expect("peeked"); - turn.response = Some(assistant_msg.content.clone()); - turn.completed_at = Some(assistant_msg.created_at.to_rfc3339()); - } - - // Incomplete turn (user message without response) - if turn.response.is_none() { - turn.state = "Failed".to_string(); - } - - turns.push(turn); - turn_number += 1; - } - } - - turns -} - pub async fn chat_threads_handler( State(state): State>, ) -> Result, (StatusCode, String)> { @@ -454,7 +435,7 @@ pub async fn chat_threads_handler( let info = ThreadInfo { id: s.id, state: "Idle".to_string(), - turn_count: (s.message_count / 2).max(0) as usize, + turn_count: s.message_count.max(0) as usize, created_at: s.started_at.to_rfc3339(), updated_at: s.last_activity.to_rfc3339(), title: s.title.clone(), @@ -630,4 +611,105 @@ mod tests { assert!(turns[1].response.is_none()); assert_eq!(turns[1].state, "Failed"); } + + #[test] + fn test_build_turns_with_tool_calls() { + let now = chrono::Utc::now(); + let tool_calls_json = serde_json::json!([ + {"name": "shell", "result_preview": "file1.txt\nfile2.txt"}, + {"name": "http", "error": "timeout"} + ]); + let messages = vec![ + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "user".to_string(), + content: "List files".to_string(), + created_at: now, + }, + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "tool_calls".to_string(), + content: tool_calls_json.to_string(), + created_at: now + chrono::TimeDelta::milliseconds(500), + }, + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "assistant".to_string(), + content: "Here are the files".to_string(), + created_at: now + chrono::TimeDelta::seconds(1), + }, + ]; + + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].tool_calls.len(), 2); + assert_eq!(turns[0].tool_calls[0].name, "shell"); + assert!(turns[0].tool_calls[0].has_result); + assert!(!turns[0].tool_calls[0].has_error); + assert_eq!( + turns[0].tool_calls[0].result_preview.as_deref(), + Some("file1.txt\nfile2.txt") + ); + assert_eq!(turns[0].tool_calls[1].name, "http"); + assert!(turns[0].tool_calls[1].has_error); + assert_eq!(turns[0].tool_calls[1].error.as_deref(), Some("timeout")); + assert_eq!(turns[0].response.as_deref(), Some("Here are the files")); + assert_eq!(turns[0].state, "Completed"); + } + + #[test] + fn test_build_turns_with_malformed_tool_calls() { + let now = chrono::Utc::now(); + let messages = vec![ + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "user".to_string(), + content: "Hello".to_string(), + created_at: now, + }, + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "tool_calls".to_string(), + content: "not valid json".to_string(), + created_at: now + chrono::TimeDelta::milliseconds(500), + }, + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "assistant".to_string(), + content: "Done".to_string(), + created_at: now + chrono::TimeDelta::seconds(1), + }, + ]; + + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert!(turns[0].tool_calls.is_empty()); + assert_eq!(turns[0].response.as_deref(), Some("Done")); + } + + #[test] + fn test_build_turns_backward_compatible_no_tool_calls() { + // Old threads without tool_calls messages still work + let now = chrono::Utc::now(); + let messages = vec![ + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "user".to_string(), + content: "Hello".to_string(), + created_at: now, + }, + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: "assistant".to_string(), + content: "Hi!".to_string(), + created_at: now + chrono::TimeDelta::seconds(1), + }, + ]; + + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert!(turns[0].tool_calls.is_empty()); + assert_eq!(turns[0].response.as_deref(), Some("Hi!")); + assert_eq!(turns[0].state, "Completed"); + } } diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 733c8a60..8970651e 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -21,6 +21,7 @@ pub mod openai_compat; pub mod server; pub mod sse; pub mod types; +pub(crate) mod util; pub mod ws; use std::net::SocketAddr; diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index e8db24a2..974aebba 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -35,6 +35,7 @@ use crate::channels::web::handlers::skills::{ use crate::channels::web::log_layer::LogBroadcaster; use crate::channels::web::sse::SseManager; use crate::channels::web::types::*; +use crate::channels::web::util::{build_turns_from_db_messages, truncate_preview}; use crate::db::Database; use crate::extensions::ExtensionManager; use crate::orchestrator::job_manager::ContainerJobManager; @@ -735,12 +736,13 @@ async fn chat_history_handler( turns, has_more, oldest_timestamp, + pending_approval: None, })); } // Try in-memory first (freshest data for active threads) if let Some(thread) = sess.threads.get(&thread_id) - && !thread.turns.is_empty() + && (!thread.turns.is_empty() || thread.pending_approval.is_some()) { let turns: Vec = thread .turns @@ -759,16 +761,35 @@ async fn chat_history_handler( name: tc.name.clone(), has_result: tc.result.is_some(), has_error: tc.error.is_some(), + result_preview: tc.result.as_ref().map(|r| { + let s = match r { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + truncate_preview(&s, 500) + }), + error: tc.error.clone(), }) .collect(), }) .collect(); + let pending_approval = thread + .pending_approval + .as_ref() + .map(|pa| PendingApprovalInfo { + request_id: pa.request_id.to_string(), + tool_name: pa.tool_name.clone(), + description: pa.description.clone(), + parameters: serde_json::to_string_pretty(&pa.parameters).unwrap_or_default(), + }); + return Ok(Json(HistoryResponse { thread_id, turns, has_more: false, oldest_timestamp: None, + pending_approval, })); } @@ -787,6 +808,7 @@ async fn chat_history_handler( turns, has_more, oldest_timestamp, + pending_approval: None, })); } } @@ -797,49 +819,10 @@ async fn chat_history_handler( turns: Vec::new(), has_more: false, oldest_timestamp: None, + pending_approval: None, })) } -/// Build TurnInfo pairs from flat DB messages (alternating user/assistant). -fn build_turns_from_db_messages(messages: &[crate::history::ConversationMessage]) -> Vec { - let mut turns = Vec::new(); - let mut turn_number = 0; - let mut iter = messages.iter().peekable(); - - while let Some(msg) = iter.next() { - if msg.role == "user" { - let mut turn = TurnInfo { - turn_number, - user_input: msg.content.clone(), - response: None, - state: "Completed".to_string(), - started_at: msg.created_at.to_rfc3339(), - completed_at: None, - tool_calls: Vec::new(), - }; - - // Check if next message is an assistant response - if let Some(next) = iter.peek() - && next.role == "assistant" - { - let assistant_msg = iter.next().expect("peeked"); - turn.response = Some(assistant_msg.content.clone()); - turn.completed_at = Some(assistant_msg.created_at.to_rfc3339()); - } - - // Incomplete turn (user message without response) - if turn.response.is_none() { - turn.state = "Failed".to_string(); - } - - turns.push(turn); - turn_number += 1; - } - } - - turns -} - async fn chat_threads_handler( State(state): State>, ) -> Result, (StatusCode, String)> { @@ -870,7 +853,7 @@ async fn chat_threads_handler( let info = ThreadInfo { id: s.id, state: "Idle".to_string(), - turn_count: (s.message_count / 2).max(0) as usize, + turn_count: s.message_count.max(0) as usize, created_at: s.started_at.to_rfc3339(), updated_at: s.last_activity.to_rfc3339(), title: s.title.clone(), diff --git a/src/channels/web/static/app.js b/src/channels/web/static/app.js index d9253a91..2566878b 100644 --- a/src/channels/web/static/app.js +++ b/src/channels/web/static/app.js @@ -135,7 +135,6 @@ function connectSSE() { if (!isCurrentThread(data.thread_id)) return; finalizeActivityGroup(); addMessage('assistant', data.content); - setStatus(''); enableChatInput(); // Refresh thread list so new titles appear after first message loadThreads(); @@ -175,10 +174,10 @@ function connectSSE() { eventSource.addEventListener('status', (e) => { const data = JSON.parse(e.data); if (!isCurrentThread(data.thread_id)) return; - setStatus(data.message); // "Done" and "Awaiting approval" are terminal signals from the agent: // the agentic loop finished, so re-enable input as a safety net in case // the response SSE event is empty or lost. + // Status text is not displayed — inline activity cards handle visual feedback. if (data.message === 'Done' || data.message === 'Awaiting approval') { finalizeActivityGroup(); enableChatInput(); @@ -320,6 +319,8 @@ function sendApprovalAction(requestId, action) { const labelText = action === 'approve' ? 'Approved' : action === 'always' ? 'Always approved' : 'Denied'; label.textContent = labelText; actions.appendChild(label); + // Remove the card after showing the confirmation briefly + setTimeout(() => { card.remove(); }, 1500); } } @@ -907,10 +908,22 @@ function loadHistory(before) { container.innerHTML = ''; for (const turn of data.turns) { addMessage('user', turn.user_input); + if (turn.tool_calls && turn.tool_calls.length > 0) { + addToolCallsSummary(turn.tool_calls); + } if (turn.response) { addMessage('assistant', turn.response); } } + // Show processing indicator if the last turn is still in-progress + var lastTurn = data.turns.length > 0 ? data.turns[data.turns.length - 1] : null; + if (lastTurn && !lastTurn.response && lastTurn.state === 'Processing') { + showActivityThinking('Processing...'); + } + // Re-render pending approval card if the thread is awaiting approval + if (data.pending_approval) { + showApproval(data.pending_approval); + } } else { // Pagination: prepend older messages const savedHeight = container.scrollHeight; @@ -918,6 +931,9 @@ function loadHistory(before) { for (const turn of data.turns) { const userDiv = createMessageElement('user', turn.user_input); fragment.appendChild(userDiv); + if (turn.tool_calls && turn.tool_calls.length > 0) { + fragment.appendChild(createToolCallsSummaryElement(turn.tool_calls)); + } if (turn.response) { const assistantDiv = createMessageElement('assistant', turn.response); fragment.appendChild(assistantDiv); @@ -951,6 +967,61 @@ function createMessageElement(role, content) { return div; } +function addToolCallsSummary(toolCalls) { + const container = document.getElementById('chat-messages'); + container.appendChild(createToolCallsSummaryElement(toolCalls)); + container.scrollTop = container.scrollHeight; +} + +function createToolCallsSummaryElement(toolCalls) { + const div = document.createElement('div'); + div.className = 'tool-calls-summary'; + + const header = document.createElement('div'); + header.className = 'tool-calls-header'; + header.textContent = toolCalls.length + ' tool' + (toolCalls.length !== 1 ? 's' : '') + ' used'; + div.appendChild(header); + + const list = document.createElement('div'); + list.className = 'tool-calls-list'; + + for (const tc of toolCalls) { + const item = document.createElement('div'); + item.className = 'tool-call-item' + (tc.has_error ? ' tool-error' : ''); + + const icon = tc.has_error ? '\u2717' : '\u2713'; + const nameSpan = document.createElement('span'); + nameSpan.className = 'tool-call-name'; + nameSpan.textContent = icon + ' ' + tc.name; + item.appendChild(nameSpan); + + if (tc.result_preview) { + const preview = document.createElement('div'); + preview.className = 'tool-call-preview'; + preview.textContent = tc.result_preview; + item.appendChild(preview); + } + if (tc.error) { + const errDiv = document.createElement('div'); + errDiv.className = 'tool-call-error-text'; + errDiv.textContent = tc.error; + item.appendChild(errDiv); + } + + list.appendChild(item); + } + + div.appendChild(list); + + header.style.cursor = 'pointer'; + header.addEventListener('click', () => { + list.classList.toggle('expanded'); + header.classList.toggle('expanded'); + }); + + return div; +} + function removeScrollSpinner() { const spinner = document.getElementById('scroll-load-spinner'); if (spinner) spinner.remove(); diff --git a/src/channels/web/static/style.css b/src/channels/web/static/style.css index a3adae3d..6a8838a3 100644 --- a/src/channels/web/static/style.css +++ b/src/channels/web/static/style.css @@ -833,6 +833,75 @@ body { font-style: italic; } +/* Tool calls summary (persisted between user/assistant messages) */ +.tool-calls-summary { + background: var(--bg-secondary); + border-left: 3px solid var(--warning); + padding: 6px 12px; + margin: 4px 0; + font-size: 0.85em; + border-radius: 4px; +} + +.tool-calls-header { + color: var(--text-secondary); + font-weight: 500; + user-select: none; +} + +.tool-calls-header::before { + content: '\25B6'; + display: inline-block; + margin-right: 6px; + font-size: 0.7em; + transition: transform 0.15s; +} + +.tool-calls-header.expanded::before { + transform: rotate(90deg); +} + +.tool-calls-list { + margin-top: 6px; + display: none; +} + +.tool-calls-list.expanded { + display: block; +} + +.tool-call-item { + padding: 3px 0; + border-bottom: 1px solid var(--border); +} + +.tool-call-item:last-child { + border-bottom: none; +} + +.tool-call-name { + font-weight: 500; + color: var(--text-primary); +} + +.tool-call-preview { + color: var(--text-secondary); + font-size: 0.9em; + max-height: 60px; + overflow: hidden; + white-space: pre-wrap; + word-break: break-word; +} + +.tool-call-error-text { + color: var(--danger); + font-size: 0.9em; +} + +.tool-error .tool-call-name { + color: var(--danger); +} + /* Auth card (inline in chat) */ .auth-card { align-self: flex-start; diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index d79f7513..f1daaf95 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -55,6 +55,10 @@ pub struct ToolCallInfo { pub name: String, pub has_result: bool, pub has_error: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub result_preview: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, } #[derive(Debug, Serialize)] @@ -67,6 +71,21 @@ pub struct HistoryResponse { /// Cursor for the next page (ISO8601 timestamp of the oldest message returned). #[serde(skip_serializing_if = "Option::is_none")] pub oldest_timestamp: Option, + /// Pending tool approval that needs user action (re-rendered on thread switch). + /// + /// Only populated from in-memory state; not persisted to DB. + /// Server restart clears pending approvals. + #[serde(skip_serializing_if = "Option::is_none")] + pub pending_approval: Option, +} + +/// Lightweight DTO for a pending tool approval (excludes context_messages). +#[derive(Debug, Serialize)] +pub struct PendingApprovalInfo { + pub request_id: String, + pub tool_name: String, + pub description: String, + pub parameters: String, } // --- Approval --- diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs new file mode 100644 index 00000000..e1c242cf --- /dev/null +++ b/src/channels/web/util.rs @@ -0,0 +1,234 @@ +//! Shared utility functions for the web gateway. + +use crate::channels::web::types::{ToolCallInfo, TurnInfo}; + +/// Truncate a string to at most `max_bytes` bytes at a char boundary, appending "...". +pub fn truncate_preview(s: &str, max_bytes: usize) -> String { + if s.len() <= max_bytes { + return s.to_string(); + } + // Walk backwards from max_bytes to find a valid char boundary + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + format!("{}...", &s[..end]) +} + +/// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples). +/// +/// Handles three message patterns: +/// - `user → assistant` (legacy, no tool calls) +/// - `user → tool_calls → assistant` (with persisted tool call summaries) +/// - `user` alone (incomplete turn) +pub fn build_turns_from_db_messages( + messages: &[crate::history::ConversationMessage], +) -> Vec { + let mut turns = Vec::new(); + let mut turn_number = 0; + let mut iter = messages.iter().peekable(); + + while let Some(msg) = iter.next() { + if msg.role == "user" { + let mut turn = TurnInfo { + turn_number, + user_input: msg.content.clone(), + response: None, + state: "Completed".to_string(), + started_at: msg.created_at.to_rfc3339(), + completed_at: None, + tool_calls: Vec::new(), + }; + + // Check if next message is a tool_calls record + if let Some(next) = iter.peek() + && next.role == "tool_calls" + { + let tc_msg = iter.next().expect("peeked"); + match serde_json::from_str::>(&tc_msg.content) { + Ok(calls) => { + turn.tool_calls = calls + .iter() + .map(|c| ToolCallInfo { + name: c["name"].as_str().unwrap_or("unknown").to_string(), + has_result: c.get("result_preview").is_some(), + has_error: c.get("error").is_some(), + result_preview: c["result_preview"].as_str().map(String::from), + error: c["error"].as_str().map(String::from), + }) + .collect(); + } + Err(e) => { + tracing::warn!( + message_id = %tc_msg.id, + "Malformed tool_calls JSON in DB, skipping: {e}" + ); + } + } + } + + // Check if next message is an assistant response + if let Some(next) = iter.peek() + && next.role == "assistant" + { + let assistant_msg = iter.next().expect("peeked"); + turn.response = Some(assistant_msg.content.clone()); + turn.completed_at = Some(assistant_msg.created_at.to_rfc3339()); + } + + // Incomplete turn (user message without response) + if turn.response.is_none() { + turn.state = "Failed".to_string(); + } + + turns.push(turn); + turn_number += 1; + } + } + + turns +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + // ---- truncate_preview tests ---- + + #[test] + fn test_truncate_preview_short_string() { + assert_eq!(truncate_preview("hello", 10), "hello"); + } + + #[test] + fn test_truncate_preview_exact_boundary() { + assert_eq!(truncate_preview("hello", 5), "hello"); + } + + #[test] + fn test_truncate_preview_truncates_ascii() { + assert_eq!(truncate_preview("hello world", 5), "hello..."); + } + + #[test] + fn test_truncate_preview_empty_string() { + assert_eq!(truncate_preview("", 10), ""); + } + + #[test] + fn test_truncate_preview_multibyte_char_boundary() { + // '€' is 3 bytes (E2 82 AC). "a€b" = [61, E2, 82, AC, 62] = 5 bytes + // Truncating at max_bytes=3 should not split the euro sign. + let s = "a€b"; + let result = truncate_preview(s, 3); + // max_bytes=3 lands mid-€, so it walks back to byte 1 ("a") + assert_eq!(result, "a..."); + } + + #[test] + fn test_truncate_preview_emoji() { + // '🦀' is 4 bytes. "hi🦀" = 6 bytes + let s = "hi🦀"; + let result = truncate_preview(s, 4); + // max_bytes=4 lands mid-🦀, walks back to byte 2 ("hi") + assert_eq!(result, "hi..."); + } + + #[test] + fn test_truncate_preview_cjk() { + // CJK characters are 3 bytes each. "你好世界" = 12 bytes + let s = "你好世界"; + let result = truncate_preview(s, 7); + // max_bytes=7 lands mid-character (byte 7 is inside 世), walks back to 6 ("你好") + assert_eq!(result, "你好..."); + } + + #[test] + fn test_truncate_preview_zero_max_bytes() { + assert_eq!(truncate_preview("hello", 0), "..."); + } + + // ---- build_turns_from_db_messages tests ---- + + fn make_msg(role: &str, content: &str, offset_ms: i64) -> crate::history::ConversationMessage { + crate::history::ConversationMessage { + id: Uuid::new_v4(), + role: role.to_string(), + content: content.to_string(), + created_at: chrono::Utc::now() + chrono::TimeDelta::milliseconds(offset_ms), + } + } + + #[test] + fn test_build_turns_complete() { + let messages = vec![ + make_msg("user", "Hello", 0), + make_msg("assistant", "Hi!", 1000), + make_msg("user", "How?", 2000), + make_msg("assistant", "Good", 3000), + ]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].user_input, "Hello"); + assert_eq!(turns[0].response.as_deref(), Some("Hi!")); + assert_eq!(turns[0].state, "Completed"); + assert_eq!(turns[1].user_input, "How?"); + assert_eq!(turns[1].response.as_deref(), Some("Good")); + } + + #[test] + fn test_build_turns_incomplete() { + let messages = vec![make_msg("user", "Hello", 0)]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert!(turns[0].response.is_none()); + assert_eq!(turns[0].state, "Failed"); + } + + #[test] + fn test_build_turns_with_tool_calls() { + let tc_json = serde_json::json!([ + {"name": "shell", "result_preview": "output"}, + {"name": "http", "error": "timeout"} + ]); + let messages = vec![ + make_msg("user", "Run it", 0), + make_msg("tool_calls", &tc_json.to_string(), 500), + make_msg("assistant", "Done", 1000), + ]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].tool_calls.len(), 2); + assert_eq!(turns[0].tool_calls[0].name, "shell"); + assert!(turns[0].tool_calls[0].has_result); + assert_eq!(turns[0].tool_calls[1].name, "http"); + assert!(turns[0].tool_calls[1].has_error); + assert_eq!(turns[0].response.as_deref(), Some("Done")); + } + + #[test] + fn test_build_turns_malformed_tool_calls() { + let messages = vec![ + make_msg("user", "Hello", 0), + make_msg("tool_calls", "not json", 500), + make_msg("assistant", "Done", 1000), + ]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert!(turns[0].tool_calls.is_empty()); + assert_eq!(turns[0].response.as_deref(), Some("Done")); + } + + #[test] + fn test_build_turns_backward_compatible() { + let messages = vec![ + make_msg("user", "Hello", 0), + make_msg("assistant", "Hi!", 1000), + ]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 1); + assert!(turns[0].tool_calls.is_empty()); + assert_eq!(turns[0].state, "Completed"); + } +} diff --git a/src/db/libsql/conversations.rs b/src/db/libsql/conversations.rs index b6cd7c4e..d9912805 100644 --- a/src/db/libsql/conversations.rs +++ b/src/db/libsql/conversations.rs @@ -97,7 +97,7 @@ impl ConversationStore for LibSqlBackend { c.started_at, c.last_activity, c.metadata, - (SELECT COUNT(*) FROM conversation_messages m WHERE m.conversation_id = c.id) AS message_count, + (SELECT COUNT(*) FROM conversation_messages m WHERE m.conversation_id = c.id AND m.role = 'user') AS message_count, (SELECT substr(m2.content, 1, 100) FROM conversation_messages m2 WHERE m2.conversation_id = c.id AND m2.role = 'user' diff --git a/src/history/store.rs b/src/history/store.rs index f01c94d7..8f7c41cb 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1316,7 +1316,7 @@ impl Store { c.started_at, c.last_activity, c.metadata, - (SELECT COUNT(*) FROM conversation_messages m WHERE m.conversation_id = c.id) AS message_count, + (SELECT COUNT(*) FROM conversation_messages m WHERE m.conversation_id = c.id AND m.role = 'user') AS message_count, (SELECT LEFT(m2.content, 100) FROM conversation_messages m2 WHERE m2.conversation_id = c.id AND m2.role = 'user'