gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-26 15:40:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(web): persist tool calls, restore approvals on thread switch, and UI fixes (#382)

Browse Source
This commit is contained in:
Henry Park
2026-02-27 17:46:37 +04:00
committed by GitHub
parent a89c5f7348
commit ddd01a628c
10 changed files with 644 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/thread_ops.rs
+95 -4
View File
@@ -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<serde_json::Value> = 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<Mutex<Session>>,
@@ -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
src/channels/web/handlers/chat.rs
+126 -44
View File
@@ -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<Arc<GatewayState>>,
@@ -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<TurnInfo> = 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<TurnInfo> {
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<Arc<GatewayState>>,
) -> Result<Json<ThreadListResponse>, (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");
}
}
src/channels/web/mod.rs
+1
View File
@@ -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;
src/channels/web/server.rs
+25 -42
View File
@@ -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<TurnInfo> = 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<TurnInfo> {
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<Arc<GatewayState>>,
) -> Result<Json<ThreadListResponse>, (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(),
src/channels/web/static/app.js
+73 -2
View File
@@ -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();
src/channels/web/static/style.css
+69
View File
@@ -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;
src/channels/web/types.rs
+19
View File
@@ -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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
#[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<String>,
/// 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<PendingApprovalInfo>,
}
/// 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 ---
src/channels/web/util.rs
+234
View File
@@ -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<TurnInfo> {
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::<Vec<serde_json::Value>>(&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");
}
}
src/db/libsql/conversations.rs
+1 -1
View File
@@ -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'
src/history/store.rs
+1 -1
View File
@@ -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'