diff --git a/.gitignore b/.gitignore index d0de6ded..9397a220 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,4 @@ bench-results/ # WASM build artifacts (loaded from disk, not bundled) *.wasm +trace_*.json diff --git a/migrations/V11__conversation_unique_indexes.sql b/migrations/V11__conversation_unique_indexes.sql new file mode 100644 index 00000000..750c4069 --- /dev/null +++ b/migrations/V11__conversation_unique_indexes.sql @@ -0,0 +1,13 @@ +-- Partial unique indexes to prevent duplicate singleton conversations. +-- These guard against TOCTOU races in get_or_create_routine_conversation +-- and get_or_create_heartbeat_conversation. + +-- One routine conversation per user per routine_id. +CREATE UNIQUE INDEX IF NOT EXISTS uq_conv_routine +ON conversations (user_id, (metadata->>'routine_id')) +WHERE metadata->>'routine_id' IS NOT NULL; + +-- One heartbeat conversation per user. +CREATE UNIQUE INDEX IF NOT EXISTS uq_conv_heartbeat +ON conversations (user_id) +WHERE metadata->>'thread_type' = 'heartbeat'; diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 60d0ea2e..f76afc75 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -96,6 +96,9 @@ pub struct Agent { pub(super) heartbeat_config: Option, pub(super) hygiene_config: Option, pub(super) routine_config: Option, + /// Optional slot to expose the routine engine to the gateway for manual triggering. + pub(super) routine_engine_slot: + Option>>>>, } impl Agent { @@ -148,9 +151,18 @@ impl Agent { heartbeat_config, hygiene_config, routine_config, + routine_engine_slot: None, } } + /// Set the routine engine slot for exposing the engine to the gateway. + pub fn set_routine_engine_slot( + &mut self, + slot: Arc>>>, + ) { + self.routine_engine_slot = Some(slot); + } + // Convenience accessors /// Get the scheduler (for external wiring, e.g. CreateJobTool). @@ -342,8 +354,13 @@ impl Agent { let heartbeat_handle = if let Some(ref hb_config) = self.heartbeat_config { if hb_config.enabled { if let Some(workspace) = self.workspace() { - let config = AgentHeartbeatConfig::default() + let mut config = AgentHeartbeatConfig::default() .with_interval(std::time::Duration::from_secs(hb_config.interval_secs)); + if let (Some(user), Some(channel)) = + (&hb_config.notify_user, &hb_config.notify_channel) + { + config = config.with_notify(user, channel); + } // Set up notification channel let (notify_tx, mut notify_rx) = @@ -396,6 +413,7 @@ impl Agent { self.cheap_llm().clone(), self.safety().clone(), Some(notify_tx), + self.store().map(Arc::clone), )) } else { tracing::warn!("Heartbeat enabled but no workspace available"); @@ -486,6 +504,11 @@ impl Agent { // SAFETY: self is consumed by run(), we can smuggle the engine in // via a local to use in the message loop below. + // Expose engine to gateway for manual triggering + if let Some(ref slot) = self.routine_engine_slot { + *slot.write().await = Some(Arc::clone(&engine)); + } + tracing::info!( "Routines enabled: cron ticker every {}s, max {} concurrent", rt_config.cron_check_interval_secs, diff --git a/src/agent/heartbeat.rs b/src/agent/heartbeat.rs index 34f56a5c..a034a1a1 100644 --- a/src/agent/heartbeat.rs +++ b/src/agent/heartbeat.rs @@ -29,6 +29,7 @@ use std::time::Duration; use tokio::sync::mpsc; use crate::channels::OutgoingResponse; +use crate::db::Database; use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning}; use crate::safety::SafetyLayer; use crate::workspace::Workspace; @@ -103,6 +104,7 @@ pub struct HeartbeatRunner { llm: Arc, safety: Arc, response_tx: Option>, + store: Option>, consecutive_failures: u32, } @@ -122,6 +124,7 @@ impl HeartbeatRunner { llm, safety, response_tx: None, + store: None, consecutive_failures: 0, } } @@ -132,6 +135,12 @@ impl HeartbeatRunner { self } + /// Set the database store for persistent heartbeat conversations. + pub fn with_store(mut self, store: Arc) -> Self { + self.store = Some(store); + self + } + /// Run the heartbeat loop. /// /// This runs forever, checking periodically based on the configured interval. @@ -292,9 +301,32 @@ impl HeartbeatRunner { return; }; + let user_id = self.config.notify_user_id.as_deref().unwrap_or("default"); + + // Persist to heartbeat conversation and get thread_id + let thread_id = if let Some(ref store) = self.store { + match store.get_or_create_heartbeat_conversation(user_id).await { + Ok(conv_id) => { + if let Err(e) = store + .add_conversation_message(conv_id, "assistant", message) + .await + { + tracing::error!("Failed to persist heartbeat message: {}", e); + } + Some(conv_id.to_string()) + } + Err(e) => { + tracing::error!("Failed to get heartbeat conversation: {}", e); + None + } + } + } else { + None + }; + let response = OutgoingResponse { content: format!("🔔 *Heartbeat Alert*\n\n{}", message), - thread_id: None, + thread_id, attachments: Vec::new(), metadata: serde_json::json!({ "source": "heartbeat", @@ -356,11 +388,15 @@ pub fn spawn_heartbeat( llm: Arc, safety: Arc, response_tx: Option>, + store: Option>, ) -> tokio::task::JoinHandle<()> { let mut runner = HeartbeatRunner::new(config, hygiene_config, workspace, llm, safety); if let Some(tx) = response_tx { runner = runner.with_response_channel(tx); } + if let Some(s) = store { + runner = runner.with_store(s); + } tokio::spawn(async move { runner.run().await; @@ -495,4 +531,22 @@ mod tests { let content = "\nActual task here"; assert!(!is_effectively_empty(content)); } + + #[test] + fn test_spawn_heartbeat_accepts_store_param() { + // Regression: spawn_heartbeat must accept an optional Database store + // for persisting heartbeat notifications to a dedicated conversation. + // Compile-time check: the 7th parameter is `Option>`. + #[allow(clippy::type_complexity)] + let _fn_ptr: fn( + HeartbeatConfig, + HygieneConfig, + Arc, + Arc, + Arc, + Option>, + Option>, + ) -> tokio::task::JoinHandle<()> = spawn_heartbeat; + let _ = _fn_ptr; + } } diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index bc5508d5..da22ffc1 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -184,7 +184,11 @@ impl RoutineEngine { /// /// Bypasses cooldown checks (those only apply to cron/event triggers). /// Still enforces enabled check and concurrent run limit. - pub async fn fire_manual(&self, routine_id: Uuid) -> Result { + pub async fn fire_manual( + &self, + routine_id: Uuid, + user_id: Option<&str>, + ) -> Result { let routine = self .store .get_routine(routine_id) @@ -194,6 +198,13 @@ impl RoutineEngine { })? .ok_or(RoutineError::NotFound { id: routine_id })?; + // Enforce ownership when a user_id is provided (gateway calls). + if let Some(uid) = user_id + && routine.user_id != uid + { + return Err(RoutineError::NotAuthorized { id: routine_id }); + } + if !routine.enabled { return Err(RoutineError::Disabled { name: routine.name.clone(), @@ -396,6 +407,39 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun) tracing::error!(routine = %routine.name, "Failed to update runtime state: {}", e); } + // Persist routine result to its dedicated conversation thread + let thread_id = match ctx + .store + .get_or_create_routine_conversation(routine.id, &routine.name, &routine.user_id) + .await + { + Ok(conv_id) => { + tracing::debug!( + routine = %routine.name, + routine_id = %routine.id, + conversation_id = %conv_id, + "Resolved routine conversation thread" + ); + // Record the run result as a conversation message + let msg = match (&summary, status) { + (Some(s), _) => format!("[{}] {}: {}", run.trigger_type, status, s), + (None, _) => format!("[{}] {}", run.trigger_type, status), + }; + if let Err(e) = ctx + .store + .add_conversation_message(conv_id, "assistant", &msg) + .await + { + tracing::error!(routine = %routine.name, "Failed to persist routine message: {}", e); + } + Some(conv_id.to_string()) + } + Err(e) => { + tracing::error!(routine = %routine.name, "Failed to get routine conversation: {}", e); + None + } + }; + // Send notifications based on config send_notification( &ctx.notify_tx, @@ -403,6 +447,7 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun) &routine.name, status, summary.as_deref(), + thread_id.as_deref(), ) .await; } @@ -611,6 +656,7 @@ async fn send_notification( routine_name: &str, status: RunStatus, summary: Option<&str>, + thread_id: Option<&str>, ) { let should_notify = match status { RunStatus::Ok => notify.on_success, @@ -637,7 +683,7 @@ async fn send_notification( let response = OutgoingResponse { content: message, - thread_id: None, + thread_id: thread_id.map(String::from), attachments: Vec::new(), metadata: serde_json::json!({ "source": "routine", diff --git a/src/agent/worker.rs b/src/agent/worker.rs index 9298c0f2..7d50952c 100644 --- a/src/agent/worker.rs +++ b/src/agent/worker.rs @@ -15,7 +15,8 @@ use crate::db::Database; use crate::error::Error; use crate::hooks::HookRegistry; use crate::llm::{ - ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolSelection, + ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolCall, + ToolSelection, }; use crate::safety::SafetyLayer; use crate::tools::rate_limiter::RateLimitResult; @@ -576,37 +577,54 @@ Report when the job is complete or if you encounter issues you cannot resolve."# } } } - } else if selections.len() == 1 { - consecutive_tool_intent_nudges = 0; - // Single tool: execute directly - let selection = &selections[0]; - tracing::debug!( - "Job {} selecting tool: {} - {}", - self.job_id, - selection.tool_name, - selection.reasoning - ); - - let result = self - .execute_tool(&selection.tool_name, &selection.parameters) - .await; - - self.process_tool_result(reason_ctx, selection, result) - .await?; } else { - // Multiple tools: execute in parallel - tracing::debug!( - "Job {} executing {} tools in parallel", - self.job_id, - selections.len() - ); + consecutive_tool_intent_nudges = 0; - let results = self.execute_tools_parallel(&selections).await; + // Record the assistant tool_calls message so that tool_result + // messages have a matching parent (prevents orphaned rewrites). + let tool_calls: Vec = selections + .iter() + .map(|s| ToolCall { + id: s.tool_call_id.clone(), + name: s.tool_name.clone(), + arguments: s.parameters.clone(), + }) + .collect(); + reason_ctx + .messages + .push(ChatMessage::assistant_with_tool_calls(None, tool_calls)); - // Process all results - for (selection, result) in selections.iter().zip(results) { - self.process_tool_result(reason_ctx, selection, result.result) + if selections.len() == 1 { + // Single tool: execute directly + let selection = &selections[0]; + tracing::debug!( + "Job {} selecting tool: {} - {}", + self.job_id, + selection.tool_name, + selection.reasoning + ); + + let result = self + .execute_tool(&selection.tool_name, &selection.parameters) + .await; + + self.process_tool_result(reason_ctx, selection, result) .await?; + } else { + // Multiple tools: execute in parallel + tracing::debug!( + "Job {} executing {} tools in parallel", + self.job_id, + selections.len() + ); + + let results = self.execute_tools_parallel(&selections).await; + + // Process all results + for (selection, result) in selections.iter().zip(results) { + self.process_tool_result(reason_ctx, selection, result.result) + .await?; + } } } @@ -1087,11 +1105,6 @@ Report when the job is complete or if you encounter issues you cannot resolve."# action.reasoning ); - // Execute the planned tool - let result = self - .execute_tool(&action.tool_name, &action.parameters) - .await; - // Create a synthetic ToolSelection for process_tool_result. // Plan actions don't originate from an LLM tool_call response so // there is no real tool_call_id; generate a unique one. @@ -1103,6 +1116,24 @@ Report when the job is complete or if you encounter issues you cannot resolve."# tool_call_id: format!("plan_{}_{}", self.job_id, i), }; + // Record the assistant tool_calls message so that the tool_result + // has a matching parent (prevents orphaned rewrites). + reason_ctx + .messages + .push(ChatMessage::assistant_with_tool_calls( + None, + vec![ToolCall { + id: selection.tool_call_id.clone(), + name: selection.tool_name.clone(), + arguments: selection.parameters.clone(), + }], + )); + + // Execute the planned tool + let result = self + .execute_tool(&action.tool_name, &action.parameters) + .await; + // Process the result let completed = self .process_tool_result(reason_ctx, &selection, result) diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 934a02dd..e82c2583 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -426,7 +426,7 @@ pub async fn chat_threads_handler( .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; if let Ok(summaries) = store - .list_conversations_with_preview(&state.user_id, "gateway", 50) + .list_conversations_all_channels(&state.user_id, 50) .await { let mut assistant_thread = None; @@ -441,6 +441,7 @@ pub async fn chat_threads_handler( updated_at: s.last_activity.to_rfc3339(), title: s.title.clone(), thread_type: s.thread_type.clone(), + channel: Some(s.channel.clone()), }; if s.id == assistant_id { @@ -460,6 +461,7 @@ pub async fn chat_threads_handler( updated_at: chrono::Utc::now().to_rfc3339(), title: None, thread_type: Some("assistant".to_string()), + channel: Some("gateway".to_string()), }); } @@ -472,9 +474,10 @@ pub async fn chat_threads_handler( } // Fallback: in-memory only (no assistant thread without DB) - let threads: Vec = sess - .threads - .values() + let mut sorted_threads: Vec<_> = sess.threads.values().collect(); + sorted_threads.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + let threads: Vec = sorted_threads + .into_iter() .map(|t| ThreadInfo { id: t.id, state: format!("{:?}", t.state), @@ -483,6 +486,7 @@ pub async fn chat_threads_handler( updated_at: t.updated_at.to_rfc3339(), title: None, thread_type: None, + channel: Some("gateway".to_string()), }) .collect(); @@ -502,38 +506,39 @@ pub async fn chat_new_thread_handler( ))?; let session = session_manager.get_or_create_session(&state.user_id).await; - let mut sess = session.lock().await; - let thread = sess.create_thread(); - let thread_id = thread.id; - let info = ThreadInfo { - id: thread.id, - state: format!("{:?}", thread.state), - turn_count: thread.turns.len(), - created_at: thread.created_at.to_rfc3339(), - updated_at: thread.updated_at.to_rfc3339(), - title: None, - thread_type: Some("thread".to_string()), + let (thread_id, info) = { + let mut sess = session.lock().await; + let thread = sess.create_thread(); + let id = thread.id; + let info = ThreadInfo { + id: thread.id, + state: format!("{:?}", thread.state), + turn_count: thread.turns.len(), + created_at: thread.created_at.to_rfc3339(), + updated_at: thread.updated_at.to_rfc3339(), + title: None, + thread_type: Some("thread".to_string()), + channel: Some("gateway".to_string()), + }; + (id, info) }; - // Persist the empty conversation row with thread_type metadata + // Persist the empty conversation row with thread_type metadata synchronously + // so that the subsequent loadThreads() call from the frontend sees it. if let Some(ref store) = state.store { - let store = Arc::clone(store); - let user_id = state.user_id.clone(); - tokio::spawn(async move { - if let Err(e) = store - .ensure_conversation(thread_id, "gateway", &user_id, None) - .await - { - tracing::warn!("Failed to persist new thread: {}", e); - } - let metadata_val = serde_json::json!("thread"); - if let Err(e) = store - .update_conversation_metadata_field(thread_id, "thread_type", &metadata_val) - .await - { - tracing::warn!("Failed to set thread_type metadata: {}", e); - } - }); + if let Err(e) = store + .ensure_conversation(thread_id, "gateway", &state.user_id, None) + .await + { + tracing::warn!("Failed to persist new thread: {}", e); + } + let metadata_val = serde_json::json!("thread"); + if let Err(e) = store + .update_conversation_metadata_field(thread_id, "thread_type", &metadata_val) + .await + { + tracing::warn!("Failed to set thread_type metadata: {}", e); + } } Ok(Json(info)) diff --git a/src/channels/web/handlers/routines.rs b/src/channels/web/handlers/routines.rs index 6cdccfc6..88abfc68 100644 --- a/src/channels/web/handlers/routines.rs +++ b/src/channels/web/handlers/routines.rs @@ -10,9 +10,9 @@ use axum::{ use serde::Deserialize; use uuid::Uuid; -use crate::channels::IncomingMessage; use crate::channels::web::server::GatewayState; use crate::channels::web::types::*; +use crate::error::RoutineError; pub async fn routines_list_handler( State(state): State>, @@ -133,56 +133,27 @@ pub async fn routines_trigger_handler( State(state): State>, Path(id): Path, ) -> Result, (StatusCode, String)> { - let store = state.store.as_ref().ok_or(( - StatusCode::SERVICE_UNAVAILABLE, - "Database not available".to_string(), - ))?; + // Clone the Arc out of the lock to avoid holding the RwLock across .await. + let engine = { + let guard = state.routine_engine.read().await; + guard.as_ref().cloned().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Routine engine not available".to_string(), + ))? + }; let routine_id = Uuid::parse_str(&id) .map_err(|_| (StatusCode::BAD_REQUEST, "Invalid routine ID".to_string()))?; - let routine = store - .get_routine(routine_id) + let run_id = engine + .fire_manual(routine_id, Some(&state.user_id)) .await - .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? - .ok_or((StatusCode::NOT_FOUND, "Routine not found".to_string()))?; - - if routine.user_id != state.user_id { - return Err((StatusCode::FORBIDDEN, "Access denied".to_string())); - } - - // Send the routine prompt through the message pipeline as a manual trigger. - let prompt = match &routine.action { - crate::agent::routine::RoutineAction::Lightweight { prompt, .. } => prompt.clone(), - crate::agent::routine::RoutineAction::FullJob { - title, description, .. - } => format!("{}: {}", title, description), - }; - - let content = format!("[routine:{}] {}", routine.name, prompt); - let thread_id = format!( - "routine-{}-{}", - routine_id, - chrono::Utc::now().timestamp_millis() - ); - let msg = IncomingMessage::new("gateway", &state.user_id, content).with_thread(thread_id); - - let tx_guard = state.msg_tx.read().await; - let tx = tx_guard.as_ref().ok_or(( - StatusCode::SERVICE_UNAVAILABLE, - "Channel not started".to_string(), - ))?; - - tx.send(msg).await.map_err(|_| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - "Channel closed".to_string(), - ) - })?; + .map_err(|e| (routine_error_status(&e), e.to_string()))?; Ok(Json(serde_json::json!({ "status": "triggered", "routine_id": routine_id, + "run_id": run_id, }))) } @@ -337,3 +308,13 @@ fn routine_to_info(r: &crate::agent::routine::Routine) -> RoutineInfo { status: status.to_string(), } } + +/// Map `RoutineError` variants to appropriate HTTP status codes. +fn routine_error_status(err: &RoutineError) -> StatusCode { + match err { + RoutineError::NotFound { .. } => StatusCode::NOT_FOUND, + RoutineError::NotAuthorized { .. } => StatusCode::FORBIDDEN, + RoutineError::Disabled { .. } | RoutineError::MaxConcurrent { .. } => StatusCode::CONFLICT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + } +} diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 57597af0..92e8ac5f 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -99,6 +99,7 @@ impl GatewayChannel { chat_rate_limiter: server::RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), }); @@ -134,6 +135,7 @@ impl GatewayChannel { chat_rate_limiter: server::RateLimiter::new(30, 60), registry_entries: self.state.registry_entries.clone(), cost_guard: self.state.cost_guard.clone(), + routine_engine: Arc::clone(&self.state.routine_engine), startup_time: self.state.startup_time, }; mutate(&mut new_state); @@ -281,7 +283,15 @@ impl Channel for GatewayChannel { msg: &IncomingMessage, response: OutgoingResponse, ) -> Result<(), ChannelError> { - let thread_id = msg.thread_id.clone().unwrap_or_default(); + let thread_id = match &msg.thread_id { + Some(tid) => tid.clone(), + None => { + tracing::warn!( + "Gateway respond with no thread_id — skipping (clients would drop it)" + ); + return Ok(()); + } + }; self.state.sse.broadcast(SseEvent::Response { content: response.content, @@ -387,9 +397,18 @@ impl Channel for GatewayChannel { _user_id: &str, response: OutgoingResponse, ) -> Result<(), ChannelError> { + let thread_id = match response.thread_id { + Some(tid) => tid, + None => { + tracing::warn!( + "Gateway broadcast with no thread_id — skipping (clients would drop it)" + ); + return Ok(()); + } + }; self.state.sse.broadcast(SseEvent::Response { content: response.content, - thread_id: String::new(), + thread_id, }); Ok(()) } diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 9628bb2c..65264a2b 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -57,6 +57,10 @@ pub type PromptQueue = Arc< >, >; +/// Slot for the routine engine, filled at runtime after the agent starts. +pub type RoutineEngineSlot = + Arc>>>; + /// Simple sliding-window rate limiter. /// /// Tracks the number of requests in the current window. Resets when the window expires. @@ -165,6 +169,8 @@ pub struct GatewayState { pub registry_entries: Vec, /// Cost guard for token/cost tracking. pub cost_guard: Option>, + /// Routine engine slot for manual routine triggering (filled at runtime). + pub routine_engine: RoutineEngineSlot, /// Server startup time for uptime calculation. pub startup_time: std::time::Instant, } @@ -1037,7 +1043,7 @@ async fn chat_threads_handler( .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; if let Ok(summaries) = store - .list_conversations_with_preview(&state.user_id, "gateway", 50) + .list_conversations_all_channels(&state.user_id, 50) .await { let mut assistant_thread = None; @@ -1052,6 +1058,7 @@ async fn chat_threads_handler( updated_at: s.last_activity.to_rfc3339(), title: s.title.clone(), thread_type: s.thread_type.clone(), + channel: Some(s.channel.clone()), }; if s.id == assistant_id { @@ -1071,6 +1078,7 @@ async fn chat_threads_handler( updated_at: chrono::Utc::now().to_rfc3339(), title: None, thread_type: Some("assistant".to_string()), + channel: Some("gateway".to_string()), }); } @@ -1083,9 +1091,10 @@ async fn chat_threads_handler( } // Fallback: in-memory only (no assistant thread without DB) - let threads: Vec = sess - .threads - .values() + let mut sorted_threads: Vec<_> = sess.threads.values().collect(); + sorted_threads.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + let threads: Vec = sorted_threads + .into_iter() .map(|t| ThreadInfo { id: t.id, state: format!("{:?}", t.state), @@ -1094,6 +1103,7 @@ async fn chat_threads_handler( updated_at: t.updated_at.to_rfc3339(), title: None, thread_type: None, + channel: Some("gateway".to_string()), }) .collect(); @@ -1113,38 +1123,39 @@ async fn chat_new_thread_handler( ))?; let session = session_manager.get_or_create_session(&state.user_id).await; - let mut sess = session.lock().await; - let thread = sess.create_thread(); - let thread_id = thread.id; - let info = ThreadInfo { - id: thread.id, - state: format!("{:?}", thread.state), - turn_count: thread.turns.len(), - created_at: thread.created_at.to_rfc3339(), - updated_at: thread.updated_at.to_rfc3339(), - title: None, - thread_type: Some("thread".to_string()), + let (thread_id, info) = { + let mut sess = session.lock().await; + let thread = sess.create_thread(); + let id = thread.id; + let info = ThreadInfo { + id: thread.id, + state: format!("{:?}", thread.state), + turn_count: thread.turns.len(), + created_at: thread.created_at.to_rfc3339(), + updated_at: thread.updated_at.to_rfc3339(), + title: None, + thread_type: Some("thread".to_string()), + channel: Some("gateway".to_string()), + }; + (id, info) }; - // Persist the empty conversation row with thread_type metadata + // Persist the empty conversation row with thread_type metadata synchronously + // so that the subsequent loadThreads() call from the frontend sees it. if let Some(ref store) = state.store { - let store = Arc::clone(store); - let user_id = state.user_id.clone(); - tokio::spawn(async move { - if let Err(e) = store - .ensure_conversation(thread_id, "gateway", &user_id, None) - .await - { - tracing::warn!("Failed to persist new thread: {}", e); - } - let metadata_val = serde_json::json!("thread"); - if let Err(e) = store - .update_conversation_metadata_field(thread_id, "thread_type", &metadata_val) - .await - { - tracing::warn!("Failed to set thread_type metadata: {}", e); - } - }); + if let Err(e) = store + .ensure_conversation(thread_id, "gateway", &state.user_id, None) + .await + { + tracing::warn!("Failed to persist new thread: {}", e); + } + let metadata_val = serde_json::json!("thread"); + if let Err(e) = store + .update_conversation_metadata_field(thread_id, "thread_type", &metadata_val) + .await + { + tracing::warn!("Failed to set thread_type metadata: {}", e); + } } Ok(Json(info)) @@ -1965,47 +1976,35 @@ async fn routines_trigger_handler( State(state): State>, Path(id): Path, ) -> Result, (StatusCode, String)> { - let store = state.store.as_ref().ok_or(( - StatusCode::SERVICE_UNAVAILABLE, - "Database not available".to_string(), - ))?; + let engine = { + let guard = state.routine_engine.read().await; + guard.as_ref().cloned().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Routine engine not available".to_string(), + ))? + }; let routine_id = Uuid::parse_str(&id) .map_err(|_| (StatusCode::BAD_REQUEST, "Invalid routine ID".to_string()))?; - let routine = store - .get_routine(routine_id) + let run_id = engine + .fire_manual(routine_id, Some(&state.user_id)) .await - .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? - .ok_or((StatusCode::NOT_FOUND, "Routine not found".to_string()))?; - - // Send the routine prompt through the message pipeline as a manual trigger. - let prompt = match &routine.action { - crate::agent::routine::RoutineAction::Lightweight { prompt, .. } => prompt.clone(), - crate::agent::routine::RoutineAction::FullJob { - title, description, .. - } => format!("{}: {}", title, description), - }; - - let content = format!("[routine:{}] {}", routine.name, prompt); - let msg = IncomingMessage::new("gateway", &state.user_id, content); - - let tx_guard = state.msg_tx.read().await; - let tx = tx_guard.as_ref().ok_or(( - StatusCode::SERVICE_UNAVAILABLE, - "Channel not started".to_string(), - ))?; - - tx.send(msg).await.map_err(|_| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - "Channel closed".to_string(), - ) - })?; + .map_err(|e| { + let status = match &e { + crate::error::RoutineError::NotFound { .. } => StatusCode::NOT_FOUND, + crate::error::RoutineError::NotAuthorized { .. } => StatusCode::FORBIDDEN, + crate::error::RoutineError::Disabled { .. } + | crate::error::RoutineError::MaxConcurrent { .. } => StatusCode::CONFLICT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + (status, e.to_string()) + })?; Ok(Json(serde_json::json!({ "status": "triggered", "routine_id": routine_id, + "run_id": run_id, }))) } @@ -2463,6 +2462,7 @@ mod tests { chat_rate_limiter: RateLimiter::new(30, 60), registry_entries: vec![], cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), }) } diff --git a/src/channels/web/static/app.js b/src/channels/web/static/app.js index 84bb697e..538be296 100644 --- a/src/channels/web/static/app.js +++ b/src/channels/web/static/app.js @@ -5,6 +5,7 @@ let eventSource = null; let logEventSource = null; let currentTab = 'chat'; let currentThreadId = null; +let currentThreadIsReadOnly = false; let assistantThreadId = null; let hasMore = false; let oldestTimestamp = null; @@ -13,6 +14,8 @@ let sseHasConnectedBefore = false; let jobEvents = new Map(); // job_id -> Array of events let jobListRefreshTimer = null; let pairingPollInterval = null; +let unreadThreads = new Map(); // thread_id -> unread count +let _loadThreadsTimer = null; const JOB_EVENTS_CAP = 500; const MEMORY_SEARCH_QUERY_MAX_LENGTH = 100; @@ -273,7 +276,13 @@ function connectSSE() { eventSource.addEventListener('response', (e) => { const data = JSON.parse(e.data); - if (!isCurrentThread(data.thread_id)) return; + if (!isCurrentThread(data.thread_id)) { + if (data.thread_id) { + unreadThreads.set(data.thread_id, (unreadThreads.get(data.thread_id) || 0) + 1); + debouncedLoadThreads(); + } + return; + } finalizeActivityGroup(); addMessage('assistant', data.content); enableChatInput(); @@ -288,7 +297,10 @@ function connectSSE() { eventSource.addEventListener('thinking', (e) => { const data = JSON.parse(e.data); - if (!isCurrentThread(data.thread_id)) return; + if (!isCurrentThread(data.thread_id)) { + if (data.thread_id) debouncedLoadThreads(); + return; + } showActivityThinking(data.message); }); @@ -324,7 +336,10 @@ function connectSSE() { eventSource.addEventListener('status', (e) => { const data = JSON.parse(e.data); - if (!isCurrentThread(data.thread_id)) return; + if (!isCurrentThread(data.thread_id)) { + if (data.thread_id) debouncedLoadThreads(); + return; + } // "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. @@ -414,9 +429,9 @@ function connectSSE() { } // Check if an SSE event belongs to the currently viewed thread. -// Events without a thread_id (legacy) are always shown. +// Events without a thread_id are dropped (prevents notification leaking). function isCurrentThread(threadId) { - if (!threadId) return true; + if (!threadId) return false; if (!currentThreadId) return true; return threadId === currentThreadId; } @@ -446,7 +461,14 @@ function sendMessage() { } function enableChatInput() { - // no-op: input and send button are always enabled + if (currentThreadIsReadOnly) return; + const input = document.getElementById('chat-input'); + const btn = document.getElementById('send-btn'); + if (input) { + input.disabled = false; + input.placeholder = 'Message or / for commands...'; + } + if (btn) btn.disabled = false; } // --- Slash Autocomplete --- @@ -1134,7 +1156,9 @@ function loadHistory(before) { // Fresh load: clear and render container.innerHTML = ''; for (const turn of data.turns) { - addMessage('user', turn.user_input); + if (turn.user_input) { + addMessage('user', turn.user_input); + } if (turn.tool_calls && turn.tool_calls.length > 0) { addToolCallsSummary(turn.tool_calls); } @@ -1156,8 +1180,10 @@ function loadHistory(before) { const savedHeight = container.scrollHeight; const fragment = document.createDocumentFragment(); for (const turn of data.turns) { - const userDiv = createMessageElement('user', turn.user_input); - fragment.appendChild(userDiv); + if (turn.user_input) { + 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)); } @@ -1256,6 +1282,37 @@ function removeScrollSpinner() { // --- Threads --- +function threadTitle(thread) { + if (thread.title) return thread.title; + const ch = thread.channel || 'gateway'; + if (thread.thread_type === 'heartbeat') return 'Heartbeat Alerts'; + if (thread.thread_type === 'routine') return 'Routine'; + if (ch !== 'gateway') return ch.charAt(0).toUpperCase() + ch.slice(1); + if (thread.turn_count === 0) return 'New chat'; + return thread.id.substring(0, 8); +} + +function relativeTime(isoStr) { + if (!isoStr) return ''; + const diff = Date.now() - new Date(isoStr).getTime(); + const mins = Math.floor(diff / 60000); + if (mins < 1) return 'now'; + if (mins < 60) return mins + 'm ago'; + const hrs = Math.floor(mins / 60); + if (hrs < 24) return hrs + 'h ago'; + const days = Math.floor(hrs / 24); + return days + 'd ago'; +} + +function isReadOnlyChannel(channel) { + return channel && channel !== 'gateway' && channel !== 'routine' && channel !== 'heartbeat'; +} + +function debouncedLoadThreads() { + if (_loadThreadsTimer) clearTimeout(_loadThreadsTimer); + _loadThreadsTimer = setTimeout(() => { _loadThreadsTimer = null; loadThreads(); }, 500); +} + function loadThreads() { apiFetch('/api/chat/threads').then((data) => { // Pinned assistant thread @@ -1264,9 +1321,13 @@ function loadThreads() { const el = document.getElementById('assistant-thread'); const isActive = currentThreadId === assistantThreadId; el.className = 'assistant-item' + (isActive ? ' active' : ''); + const labelEl = document.getElementById('assistant-label'); + if (labelEl) { + const at = data.assistant_thread; + labelEl.textContent = 'Assistant'; + } const meta = document.getElementById('assistant-meta'); - const count = data.assistant_thread.turn_count || 0; - meta.textContent = count > 0 ? count + ' turns' : ''; + meta.textContent = relativeTime(data.assistant_thread.updated_at); } // Regular threads @@ -1275,16 +1336,38 @@ function loadThreads() { const threads = data.threads || []; for (const thread of threads) { const item = document.createElement('div'); - item.className = 'thread-item' + (thread.id === currentThreadId ? ' active' : ''); + const isActive = thread.id === currentThreadId; + item.className = 'thread-item' + (isActive ? ' active' : ''); + + // Channel badge for non-gateway threads + const ch = thread.channel || 'gateway'; + if (ch !== 'gateway') { + const badge = document.createElement('span'); + badge.className = 'thread-badge thread-badge-' + ch; + badge.textContent = ch; + item.appendChild(badge); + } + const label = document.createElement('span'); label.className = 'thread-label'; - label.textContent = thread.title || thread.id.substring(0, 8); - label.title = thread.title ? thread.title + ' (' + thread.id + ')' : thread.id; + label.textContent = threadTitle(thread); + label.title = (thread.title || '') + ' (' + thread.id + ')'; item.appendChild(label); + const meta = document.createElement('span'); meta.className = 'thread-meta'; - meta.textContent = (thread.turn_count || 0) + ' turns'; + meta.textContent = relativeTime(thread.updated_at); item.appendChild(meta); + + // Unread dot + const unread = unreadThreads.get(thread.id) || 0; + if (unread > 0 && !isActive) { + const dot = document.createElement('span'); + dot.className = 'thread-unread'; + dot.textContent = unread > 9 ? '9+' : String(unread); + item.appendChild(dot); + } + item.addEventListener('click', () => switchThread(thread.id)); list.appendChild(item); } @@ -1294,17 +1377,36 @@ function loadThreads() { switchToAssistant(); } - // Enable chat input once a thread is available + // Enable/disable chat input based on channel type if (currentThreadId) { - enableChatInput(); + const currentThread = threads.find(t => t.id === currentThreadId); + const ch = currentThread ? currentThread.channel : 'gateway'; + currentThreadIsReadOnly = isReadOnlyChannel(ch); + if (currentThreadIsReadOnly) { + disableChatInputReadOnly(); + } else { + enableChatInput(); + } } }).catch(() => {}); } +function disableChatInputReadOnly() { + const input = document.getElementById('chat-input'); + const btn = document.getElementById('send-btn'); + if (input) { + input.disabled = true; + input.placeholder = 'Read-only thread (external channel)'; + } + if (btn) btn.disabled = true; +} + function switchToAssistant() { if (!assistantThreadId) return; finalizeActivityGroup(); currentThreadId = assistantThreadId; + currentThreadIsReadOnly = false; + unreadThreads.delete(assistantThreadId); hasMore = false; oldestTimestamp = null; loadHistory(); @@ -1314,6 +1416,7 @@ function switchToAssistant() { function switchThread(threadId) { finalizeActivityGroup(); currentThreadId = threadId; + unreadThreads.delete(threadId); hasMore = false; oldestTimestamp = null; loadHistory(); diff --git a/src/channels/web/static/index.html b/src/channels/web/static/index.html index 1d232d17..be8a0c9e 100644 --- a/src/channels/web/static/index.html +++ b/src/channels/web/static/index.html @@ -113,12 +113,12 @@
- Threads +
- Assistant + Assistant
diff --git a/src/channels/web/static/style.css b/src/channels/web/static/style.css index 2889087d..a21775fb 100644 --- a/src/channels/web/static/style.css +++ b/src/channels/web/static/style.css @@ -3074,7 +3074,7 @@ mark { } .thread-sidebar { - width: 200px; + width: 240px; background: var(--bg-secondary); border-right: 1px solid var(--border); display: flex; @@ -3082,6 +3082,8 @@ mark { flex-shrink: 0; transition: width 0.2s ease; overflow: hidden; + padding: 6px; + gap: 2px; } .thread-sidebar.collapsed { @@ -3099,8 +3101,7 @@ mark { .thread-sidebar-header { display: flex; align-items: center; - padding: 10px 12px; - border-bottom: 1px solid var(--border); + padding: 10px 10px; font-size: 13px; font-weight: 600; gap: 8px; @@ -3134,21 +3135,22 @@ mark { display: flex; align-items: center; justify-content: space-between; - padding: 10px 12px; + padding: 12px 14px; cursor: pointer; font-size: 13px; font-weight: 600; color: var(--text); - border-bottom: 1px solid var(--border); - background: var(--bg-secondary); + background: var(--bg-tertiary); + border-radius: var(--radius); + margin-bottom: 2px; } .assistant-item:hover { - background: var(--bg-tertiary); + background: rgba(255, 255, 255, 0.06); } .assistant-item.active { - background: rgba(52, 211, 153, 0.08); + background: rgba(52, 211, 153, 0.1); color: var(--accent); border-left: 2px solid var(--accent); } @@ -3166,7 +3168,7 @@ mark { } .threads-section-header { - padding: 8px 12px 4px; + padding: 10px 10px 4px; font-size: 11px; font-weight: 500; text-transform: uppercase; @@ -3196,11 +3198,11 @@ mark { display: flex; align-items: center; justify-content: space-between; - padding: 8px 12px; + padding: 10px 14px; cursor: pointer; font-size: 13px; color: var(--text-secondary); - border-bottom: 1px solid rgba(255, 255, 255, 0.03); + border-radius: var(--radius); } .thread-item:hover { @@ -3222,6 +3224,43 @@ mark { .thread-meta { font-size: 11px; color: var(--text-secondary); + flex-shrink: 0; +} + +.thread-badge { + display: inline-block; + font-size: 9px; + font-weight: 600; + text-transform: uppercase; + letter-spacing: 0.5px; + padding: 1px 5px; + border-radius: 3px; + background: rgba(255, 255, 255, 0.08); + color: var(--text-secondary); + margin-right: 6px; + flex-shrink: 0; +} + +.thread-badge-routine { background: rgba(52, 211, 153, 0.15); color: var(--accent); } +.thread-badge-heartbeat { background: rgba(245, 166, 35, 0.15); color: var(--warning); } +.thread-badge-telegram { background: rgba(0, 136, 204, 0.15); color: #0088cc; } +.thread-badge-signal { background: rgba(59, 118, 240, 0.15); color: #3b76f0; } +.thread-badge-slack { background: rgba(74, 21, 75, 0.15); color: #e01e5a; } + +.thread-unread { + display: inline-flex; + align-items: center; + justify-content: center; + min-width: 16px; + height: 16px; + font-size: 10px; + font-weight: 700; + background: var(--accent); + color: var(--bg); + border-radius: 8px; + padding: 0 4px; + margin-left: auto; + flex-shrink: 0; } /* --- Memory editing --- */ @@ -3620,7 +3659,7 @@ mark { left: 0; top: 0; bottom: 0; - width: 200px; + width: 240px; z-index: 50; } diff --git a/src/channels/web/test_helpers.rs b/src/channels/web/test_helpers.rs index 9248f8f4..053dd84e 100644 --- a/src/channels/web/test_helpers.rs +++ b/src/channels/web/test_helpers.rs @@ -84,6 +84,7 @@ impl TestGatewayBuilder { chat_rate_limiter: RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), }) } diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index 18610d87..20844fc5 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -28,6 +28,8 @@ pub struct ThreadInfo { pub title: Option, #[serde(skip_serializing_if = "Option::is_none")] pub thread_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub channel: Option, } #[derive(Debug, Serialize)] @@ -1063,4 +1065,40 @@ mod tests { let req: AuthCancelRequest = serde_json::from_str(json).unwrap(); assert_eq!(req.extension_name, "telegram"); } + + // ---- ThreadInfo channel field tests ---- + + #[test] + fn test_thread_info_channel_serialized() { + let info = ThreadInfo { + id: Uuid::nil(), + state: "Idle".to_string(), + turn_count: 0, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + title: None, + thread_type: None, + channel: Some("telegram".to_string()), + }; + let json = serde_json::to_string(&info).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed["channel"], "telegram"); + } + + #[test] + fn test_thread_info_channel_omitted_when_none() { + let info = ThreadInfo { + id: Uuid::nil(), + state: "Idle".to_string(), + turn_count: 0, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + title: None, + thread_type: None, + channel: None, + }; + let json = serde_json::to_string(&info).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert!(parsed.get("channel").is_none()); + } } diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs index e1c242cf..81485b94 100644 --- a/src/channels/web/util.rs +++ b/src/channels/web/util.rs @@ -83,6 +83,19 @@ pub fn build_turns_from_db_messages( turns.push(turn); turn_number += 1; + } else if msg.role == "assistant" { + // Standalone assistant message (e.g. routine output, heartbeat) + // with no preceding user message — render as a turn with empty input. + turns.push(TurnInfo { + turn_number, + user_input: String::new(), + response: Some(msg.content.clone()), + state: "Completed".to_string(), + started_at: msg.created_at.to_rfc3339(), + completed_at: Some(msg.created_at.to_rfc3339()), + tool_calls: Vec::new(), + }); + turn_number += 1; } } @@ -220,6 +233,29 @@ mod tests { assert_eq!(turns[0].response.as_deref(), Some("Done")); } + #[test] + fn test_build_turns_standalone_assistant_messages() { + // Routine conversations only have assistant messages (no user messages). + let messages = vec![ + make_msg("assistant", "Routine executed: all checks passed", 0), + make_msg("assistant", "Routine executed: found 2 issues", 5000), + ]; + let turns = build_turns_from_db_messages(&messages); + assert_eq!(turns.len(), 2); + // Standalone assistant messages should have empty user_input + assert_eq!(turns[0].user_input, ""); + assert_eq!( + turns[0].response.as_deref(), + Some("Routine executed: all checks passed") + ); + assert_eq!(turns[0].state, "Completed"); + assert_eq!(turns[1].user_input, ""); + assert_eq!( + turns[1].response.as_deref(), + Some("Routine executed: found 2 issues") + ); + } + #[test] fn test_build_turns_backward_compatible() { let messages = vec![ diff --git a/src/channels/web/ws.rs b/src/channels/web/ws.rs index 2477217e..8a0caa54 100644 --- a/src/channels/web/ws.rs +++ b/src/channels/web/ws.rs @@ -493,6 +493,7 @@ mod tests { chat_rate_limiter: crate::channels::web::server::RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), } } diff --git a/src/db/libsql/conversations.rs b/src/db/libsql/conversations.rs index d9912805..2a7ef06c 100644 --- a/src/db/libsql/conversations.rs +++ b/src/db/libsql/conversations.rs @@ -20,9 +20,10 @@ impl ConversationStore for LibSqlBackend { ) -> Result { let conn = self.connect().await?; let id = Uuid::new_v4(); + let now = fmt_ts(&Utc::now()); conn.execute( - "INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, opt_text(thread_id)], + "INSERT INTO conversations (id, channel, user_id, thread_id, started_at, last_activity) VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + params![id.to_string(), channel, user_id, opt_text(thread_id), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -71,8 +72,8 @@ impl ConversationStore for LibSqlBackend { let now = fmt_ts(&Utc::now()); conn.execute( r#" - INSERT INTO conversations (id, channel, user_id, thread_id) - VALUES (?1, ?2, ?3, ?4) + INSERT INTO conversations (id, channel, user_id, thread_id, started_at, last_activity) + VALUES (?1, ?2, ?3, ?4, ?5, ?5) ON CONFLICT (id) DO UPDATE SET last_activity = ?5 "#, params![id.to_string(), channel, user_id, opt_text(thread_id), now], @@ -97,6 +98,7 @@ impl ConversationStore for LibSqlBackend { c.started_at, c.last_activity, c.metadata, + c.channel, (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 @@ -106,7 +108,7 @@ impl ConversationStore for LibSqlBackend { ) AS title FROM conversations c WHERE c.user_id = ?1 AND c.channel = ?2 - ORDER BY c.last_activity DESC + ORDER BY datetime(c.last_activity) DESC LIMIT ?3 "#, params![user_id, channel, limit], @@ -125,6 +127,13 @@ impl ConversationStore for LibSqlBackend { .get("thread_type") .and_then(|v| v.as_str()) .map(String::from); + let sql_title = get_opt_text(&row, 6); + let title = sql_title.or_else(|| { + metadata + .get("routine_name") + .and_then(|v| v.as_str()) + .map(String::from) + }); results.push(ConversationSummary { id: row .get::(0) @@ -133,14 +142,213 @@ impl ConversationStore for LibSqlBackend { .unwrap_or_default(), started_at: get_ts(&row, 1), last_activity: get_ts(&row, 2), - message_count: get_i64(&row, 4), - title: get_opt_text(&row, 5), + message_count: get_i64(&row, 5), + title, thread_type, + channel: get_text(&row, 4), }); } Ok(results) } + async fn list_conversations_all_channels( + &self, + user_id: &str, + limit: i64, + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + r#" + SELECT + c.id, + c.started_at, + c.last_activity, + c.metadata, + c.channel, + (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' + ORDER BY m2.created_at ASC, m2.rowid ASC + LIMIT 1 + ) AS title + FROM conversations c + WHERE c.user_id = ?1 + ORDER BY datetime(c.last_activity) DESC + LIMIT ?2 + "#, + params![user_id, limit], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let mut results = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let metadata = get_json(&row, 3); + let thread_type = metadata + .get("thread_type") + .and_then(|v| v.as_str()) + .map(String::from); + let sql_title = get_opt_text(&row, 6); + let title = sql_title.or_else(|| { + metadata + .get("routine_name") + .and_then(|v| v.as_str()) + .map(String::from) + }); + results.push(ConversationSummary { + id: row + .get::(0) + .unwrap_or_default() + .parse() + .unwrap_or_default(), + started_at: get_ts(&row, 1), + last_activity: get_ts(&row, 2), + message_count: get_i64(&row, 5), + title, + thread_type, + channel: get_text(&row, 4), + }); + } + Ok(results) + } + + /// Uses BEGIN IMMEDIATE to serialize concurrent writers and prevent + /// duplicate routine conversations (TOCTOU race). + async fn get_or_create_routine_conversation( + &self, + routine_id: Uuid, + routine_name: &str, + user_id: &str, + ) -> Result { + let conn = self.connect().await?; + let rid = routine_id.to_string(); + + conn.execute("BEGIN IMMEDIATE", params![]) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let result: Result = async { + let mut rows = conn + .query( + r#" + SELECT id FROM conversations + WHERE user_id = ?1 AND json_extract(metadata, '$.routine_id') = ?2 + LIMIT 1 + "#, + params![user_id, rid], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + if let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let id_str: String = row.get(0).unwrap_or_default(); + return id_str + .parse() + .map_err(|_| DatabaseError::Serialization("Invalid UUID".to_string())); + } + + let id = Uuid::new_v4(); + let now = fmt_ts(&Utc::now()); + let metadata = serde_json::json!({ + "thread_type": "routine", + "routine_id": routine_id.to_string(), + "routine_name": routine_name, + }); + conn.execute( + "INSERT INTO conversations (id, channel, user_id, metadata, started_at, last_activity) VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + params![id.to_string(), "routine", user_id, metadata.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(id) + } + .await; + + match &result { + Ok(_) => { + conn.execute("COMMIT", params![]) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + } + Err(_) => { + let _ = conn.execute("ROLLBACK", params![]).await; + } + } + result + } + + /// Uses BEGIN IMMEDIATE to serialize concurrent writers and prevent + /// duplicate heartbeat conversations (TOCTOU race). + async fn get_or_create_heartbeat_conversation( + &self, + user_id: &str, + ) -> Result { + let conn = self.connect().await?; + + conn.execute("BEGIN IMMEDIATE", params![]) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let result: Result = async { + let mut rows = conn + .query( + r#" + SELECT id FROM conversations + WHERE user_id = ?1 AND json_extract(metadata, '$.thread_type') = 'heartbeat' + LIMIT 1 + "#, + params![user_id], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + if let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let id_str: String = row.get(0).unwrap_or_default(); + return id_str + .parse() + .map_err(|_| DatabaseError::Serialization("Invalid UUID".to_string())); + } + + let id = Uuid::new_v4(); + let now = fmt_ts(&Utc::now()); + let metadata = serde_json::json!({ "thread_type": "heartbeat" }); + conn.execute( + "INSERT INTO conversations (id, channel, user_id, metadata, started_at, last_activity) VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + params![id.to_string(), "heartbeat", user_id, metadata.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(id) + } + .await; + + match &result { + Ok(_) => { + conn.execute("COMMIT", params![]) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + } + Err(_) => { + let _ = conn.execute("ROLLBACK", params![]).await; + } + } + result + } + async fn get_or_create_assistant_conversation( &self, user_id: &str, @@ -174,10 +382,11 @@ impl ConversationStore for LibSqlBackend { // Create new let id = Uuid::new_v4(); + let now = fmt_ts(&Utc::now()); let metadata = serde_json::json!({"thread_type": "assistant", "title": "Assistant"}); conn.execute( - "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, metadata.to_string()], + "INSERT INTO conversations (id, channel, user_id, metadata, started_at, last_activity) VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + params![id.to_string(), channel, user_id, metadata.to_string(), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -192,9 +401,10 @@ impl ConversationStore for LibSqlBackend { ) -> Result { let conn = self.connect().await?; let id = Uuid::new_v4(); + let now = fmt_ts(&Utc::now()); conn.execute( - "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, metadata.to_string()], + "INSERT INTO conversations (id, channel, user_id, metadata, started_at, last_activity) VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + params![id.to_string(), channel, user_id, metadata.to_string(), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -353,3 +563,128 @@ impl ConversationStore for LibSqlBackend { Ok(found.is_some()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::Database; + + #[tokio::test] + async fn test_get_or_create_routine_conversation_is_idempotent() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("test_routine_conv.db"); + let backend = LibSqlBackend::new_local(&db_path).await.unwrap(); + backend.run_migrations().await.unwrap(); + + let routine_id = Uuid::new_v4(); + let user_id = "test_user"; + + // First call — creates the conversation + let id1 = backend + .get_or_create_routine_conversation(routine_id, "my-routine", user_id) + .await + .unwrap(); + + // Second call — should return the SAME conversation + let id2 = backend + .get_or_create_routine_conversation(routine_id, "my-routine", user_id) + .await + .unwrap(); + + assert_eq!(id1, id2, "Expected same conversation ID on repeated calls"); + + // Third call — still the same + let id3 = backend + .get_or_create_routine_conversation(routine_id, "my-routine", user_id) + .await + .unwrap(); + + assert_eq!(id1, id3); + + // Different routine_id should get a different conversation + let other_routine_id = Uuid::new_v4(); + let id4 = backend + .get_or_create_routine_conversation(other_routine_id, "other-routine", user_id) + .await + .unwrap(); + + assert_ne!( + id1, id4, + "Different routines should get different conversations" + ); + } + + #[tokio::test] + async fn test_routine_conversation_persists_across_messages() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("test_routine_persist.db"); + let backend = LibSqlBackend::new_local(&db_path).await.unwrap(); + backend.run_migrations().await.unwrap(); + + let routine_id = Uuid::new_v4(); + let user_id = "test_user"; + + // First invocation: create conversation and add a message + let id1 = backend + .get_or_create_routine_conversation(routine_id, "my-routine", user_id) + .await + .unwrap(); + + backend + .add_conversation_message(id1, "assistant", "[cron] Completed: all good") + .await + .unwrap(); + + // Second invocation: should find existing conversation + let id2 = backend + .get_or_create_routine_conversation(routine_id, "my-routine", user_id) + .await + .unwrap(); + + assert_eq!(id1, id2, "Second invocation should reuse same conversation"); + + backend + .add_conversation_message(id2, "assistant", "[cron] Completed: still good") + .await + .unwrap(); + + // Verify only one routine conversation exists (not two) + let convs = backend + .list_conversations_all_channels(user_id, 50) + .await + .unwrap(); + + let routine_convs: Vec<_> = convs.iter().filter(|c| c.channel == "routine").collect(); + assert_eq!( + routine_convs.len(), + 1, + "Should have exactly 1 routine conversation, found {}", + routine_convs.len() + ); + } + + #[tokio::test] + async fn test_get_or_create_heartbeat_conversation_is_idempotent() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("test_heartbeat_conv.db"); + let backend = LibSqlBackend::new_local(&db_path).await.unwrap(); + backend.run_migrations().await.unwrap(); + + let user_id = "test_user"; + + let id1 = backend + .get_or_create_heartbeat_conversation(user_id) + .await + .unwrap(); + + let id2 = backend + .get_or_create_heartbeat_conversation(user_id) + .await + .unwrap(); + + assert_eq!( + id1, id2, + "Expected same heartbeat conversation on repeated calls" + ); + } +} diff --git a/src/db/libsql/mod.rs b/src/db/libsql/mod.rs index 0a813072..6ff8ca6b 100644 --- a/src/db/libsql/mod.rs +++ b/src/db/libsql/mod.rs @@ -118,15 +118,37 @@ impl LibSqlBackend { /// Sets `PRAGMA busy_timeout = 5000` on every connection so concurrent /// writers wait up to 5 seconds instead of failing instantly with /// "database is locked". + /// + /// Retries up to 3 times with exponential backoff to handle transient + /// "unable to open database file" errors from concurrent connection + /// creation (e.g. cron ticker vs main thread). pub async fn connect(&self) -> Result { - let conn = self - .db - .connect() - .map_err(|e| DatabaseError::Pool(format!("Failed to create connection: {}", e)))?; - conn.query("PRAGMA busy_timeout = 5000", ()) - .await - .map_err(|e| DatabaseError::Pool(format!("Failed to set busy_timeout: {}", e)))?; - Ok(conn) + let mut last_err = None; + for attempt in 0..3u32 { + match self.db.connect() { + Ok(conn) => { + conn.query("PRAGMA busy_timeout = 5000", ()) + .await + .map_err(|e| { + DatabaseError::Pool(format!("Failed to set busy_timeout: {}", e)) + })?; + return Ok(conn); + } + Err(e) => { + last_err = Some(e); + if attempt < 2 { + tokio::time::sleep(std::time::Duration::from_millis( + 50 * 2u64.pow(attempt), + )) + .await; + } + } + } + } + Err(DatabaseError::Pool(format!( + "Failed to create connection after 3 attempts: {}", + last_err.map(|e| e.to_string()).unwrap_or_default() + ))) } } @@ -459,4 +481,33 @@ mod tests { let count: i64 = row.get(0).unwrap(); assert_eq!(count, 20); } + + #[tokio::test] + async fn test_connect_retry_succeeds_on_valid_db() { + // Verify connect() works with retry logic on a file-backed DB + // (exercises the retry path even though transient failures are hard + // to reproduce deterministically). + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("test_retry.db"); + let backend = LibSqlBackend::new_local(&db_path).await.unwrap(); + backend.run_migrations().await.unwrap(); + + // Multiple concurrent connect() calls should all succeed + let mut handles = Vec::new(); + for _ in 0..10 { + let b = LibSqlBackend { + db: backend.shared_db(), + }; + handles.push(tokio::spawn(async move { b.connect().await })); + } + + for handle in handles { + let result = handle.await.unwrap(); + assert!( + result.is_ok(), + "concurrent connect failed: {:?}", + result.err() + ); + } + } } diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index 3006d61e..084ae53d 100644 --- a/src/db/libsql_migrations.rs +++ b/src/db/libsql_migrations.rs @@ -45,6 +45,15 @@ CREATE INDEX IF NOT EXISTS idx_conversations_channel ON conversations(channel); CREATE INDEX IF NOT EXISTS idx_conversations_user ON conversations(user_id); CREATE INDEX IF NOT EXISTS idx_conversations_last_activity ON conversations(last_activity); +-- Partial unique indexes to prevent duplicate singleton conversations. +CREATE UNIQUE INDEX IF NOT EXISTS uq_conv_routine +ON conversations (user_id, json_extract(metadata, '$.routine_id')) +WHERE json_extract(metadata, '$.routine_id') IS NOT NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_conv_heartbeat +ON conversations (user_id) +WHERE json_extract(metadata, '$.thread_type') = 'heartbeat'; + CREATE TABLE IF NOT EXISTS conversation_messages ( id TEXT PRIMARY KEY, conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE, diff --git a/src/db/mod.rs b/src/db/mod.rs index f065753a..560d682a 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -125,6 +125,21 @@ pub trait ConversationStore: Send + Sync { channel: &str, limit: i64, ) -> Result, DatabaseError>; + async fn list_conversations_all_channels( + &self, + user_id: &str, + limit: i64, + ) -> Result, DatabaseError>; + async fn get_or_create_routine_conversation( + &self, + routine_id: Uuid, + routine_name: &str, + user_id: &str, + ) -> Result; + async fn get_or_create_heartbeat_conversation( + &self, + user_id: &str, + ) -> Result; async fn get_or_create_assistant_conversation( &self, user_id: &str, diff --git a/src/db/postgres.rs b/src/db/postgres.rs index b73a81b4..9dd988bc 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -116,6 +116,36 @@ impl ConversationStore for PgBackend { .await } + async fn list_conversations_all_channels( + &self, + user_id: &str, + limit: i64, + ) -> Result, DatabaseError> { + self.store + .list_conversations_all_channels(user_id, limit) + .await + } + + async fn get_or_create_routine_conversation( + &self, + routine_id: Uuid, + routine_name: &str, + user_id: &str, + ) -> Result { + self.store + .get_or_create_routine_conversation(routine_id, routine_name, user_id) + .await + } + + async fn get_or_create_heartbeat_conversation( + &self, + user_id: &str, + ) -> Result { + self.store + .get_or_create_heartbeat_conversation(user_id) + .await + } + async fn get_or_create_assistant_conversation( &self, user_id: &str, diff --git a/src/error.rs b/src/error.rs index 973e0150..d9a01c83 100644 --- a/src/error.rs +++ b/src/error.rs @@ -401,6 +401,9 @@ pub enum RoutineError { #[error("Routine not found: {id}")] NotFound { id: Uuid }, + #[error("Not authorized to trigger routine {id}")] + NotAuthorized { id: Uuid }, + #[error("Routine {name} at max concurrent runs")] MaxConcurrent { name: String }, diff --git a/src/history/store.rs b/src/history/store.rs index 2ef121a3..2a46aaea 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1377,6 +1377,8 @@ pub struct ConversationSummary { pub last_activity: DateTime, /// Thread type extracted from metadata (e.g. "assistant", "thread"). pub thread_type: Option, + /// Channel that owns this conversation (e.g. "gateway", "telegram", "routine"). + pub channel: String, } /// A single message in a conversation. @@ -1429,6 +1431,7 @@ impl Store { c.started_at, c.last_activity, c.metadata, + c.channel, (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 @@ -1453,18 +1456,181 @@ impl Store { .get("thread_type") .and_then(|v| v.as_str()) .map(String::from); + let sql_title: Option = r.get("title"); + let title = sql_title.or_else(|| { + metadata + .get("routine_name") + .and_then(|v| v.as_str()) + .map(String::from) + }); ConversationSummary { id: r.get("id"), - title: r.get("title"), + title, message_count: r.get("message_count"), started_at: r.get("started_at"), last_activity: r.get("last_activity"), thread_type, + channel: r.get("channel"), } }) .collect()) } + /// List conversations across all channels with a title derived from the first user message. + pub async fn list_conversations_all_channels( + &self, + user_id: &str, + limit: i64, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = conn + .query( + r#" + SELECT + c.id, + c.started_at, + c.last_activity, + c.metadata, + c.channel, + (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' + ORDER BY m2.created_at ASC + LIMIT 1 + ) AS title + FROM conversations c + WHERE c.user_id = $1 + ORDER BY c.last_activity DESC + LIMIT $2 + "#, + &[&user_id, &limit], + ) + .await?; + + Ok(rows + .iter() + .map(|r| { + let metadata: serde_json::Value = r.get("metadata"); + let thread_type = metadata + .get("thread_type") + .and_then(|v| v.as_str()) + .map(String::from); + // For routine/heartbeat threads, derive title from metadata + // since they may have no user messages. + let sql_title: Option = r.get("title"); + let title = sql_title.or_else(|| { + metadata + .get("routine_name") + .and_then(|v| v.as_str()) + .map(String::from) + }); + ConversationSummary { + id: r.get("id"), + title, + message_count: r.get("message_count"), + started_at: r.get("started_at"), + last_activity: r.get("last_activity"), + thread_type, + channel: r.get("channel"), + } + }) + .collect()) + } + + /// Get or create a persistent conversation for a routine. + /// + /// Looks for a conversation where `metadata->>'routine_id' = routine_id`. + /// Creates one if it doesn't exist. Uses INSERT ON CONFLICT to avoid + /// TOCTOU races under concurrent routine executions. + pub async fn get_or_create_routine_conversation( + &self, + routine_id: Uuid, + routine_name: &str, + user_id: &str, + ) -> Result { + let conn = self.conn().await?; + let rid = routine_id.to_string(); + + // Attempt insert first; the partial unique index + // uq_conv_routine(user_id, (metadata->>'routine_id')) prevents duplicates. + let new_id = Uuid::new_v4(); + let metadata = serde_json::json!({ + "thread_type": "routine", + "routine_id": routine_id.to_string(), + "routine_name": routine_name, + }); + conn.execute( + r#" + INSERT INTO conversations (id, channel, user_id, metadata) + VALUES ($1, 'routine', $2, $3) + ON CONFLICT (user_id, (metadata->>'routine_id')) + WHERE metadata->>'routine_id' IS NOT NULL + DO NOTHING + "#, + &[&new_id, &user_id, &metadata], + ) + .await?; + + // Select back — always returns the winner. + let row = conn + .query_one( + r#" + SELECT id FROM conversations + WHERE user_id = $1 AND metadata->>'routine_id' = $2 + LIMIT 1 + "#, + &[&user_id, &rid], + ) + .await?; + + Ok(row.get("id")) + } + + /// Get or create the singleton heartbeat conversation for a user. + /// + /// Looks for a conversation where `metadata->>'thread_type' = 'heartbeat'`. + /// Creates one if it doesn't exist. Uses INSERT ON CONFLICT to avoid + /// TOCTOU races under concurrent heartbeat sends. + pub async fn get_or_create_heartbeat_conversation( + &self, + user_id: &str, + ) -> Result { + let conn = self.conn().await?; + + // Attempt insert; the partial unique index + // uq_conv_heartbeat(user_id) prevents duplicates. + let new_id = Uuid::new_v4(); + let metadata = serde_json::json!({ + "thread_type": "heartbeat", + }); + conn.execute( + r#" + INSERT INTO conversations (id, channel, user_id, metadata) + VALUES ($1, 'heartbeat', $2, $3) + ON CONFLICT (user_id) + WHERE metadata->>'thread_type' = 'heartbeat' + DO NOTHING + "#, + &[&new_id, &user_id, &metadata], + ) + .await?; + + // Select back — always returns the winner. + let row = conn + .query_one( + r#" + SELECT id FROM conversations + WHERE user_id = $1 AND metadata->>'thread_type' = 'heartbeat' + LIMIT 1 + "#, + &[&user_id], + ) + .await?; + + Ok(row.get("id")) + } + /// Get or create the singleton "assistant" conversation for a user+channel. /// /// Looks for a conversation where `metadata->>'thread_type' = 'assistant'`. @@ -1928,3 +2094,40 @@ impl Store { Ok(count > 0) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_conversation_summary_has_channel_field() { + // Regression: ConversationSummary must include a `channel` field + // so the gateway can distinguish thread origins. + let summary = ConversationSummary { + id: Uuid::nil(), + title: Some("Hello".to_string()), + message_count: 1, + started_at: Utc::now(), + last_activity: Utc::now(), + thread_type: Some("thread".to_string()), + channel: "telegram".to_string(), + }; + assert_eq!(summary.channel, "telegram"); + } + + #[test] + fn test_conversation_summary_channel_various_values() { + for ch in ["gateway", "routine", "heartbeat", "telegram", "signal"] { + let summary = ConversationSummary { + id: Uuid::nil(), + title: None, + message_count: 0, + started_at: Utc::now(), + last_activity: Utc::now(), + thread_type: None, + channel: ch.to_string(), + }; + assert_eq!(summary.channel, ch); + } + } +} diff --git a/src/llm/provider.rs b/src/llm/provider.rs index 83863573..c650f30b 100644 --- a/src/llm/provider.rs +++ b/src/llm/provider.rs @@ -522,4 +522,66 @@ mod tests { assert_eq!(messages[3].role, Role::User); // call_2 orphaned assert_eq!(messages[4].role, Role::User); // call_3 orphaned } + + /// Regression: worker's select_tools/execute_plan now emit + /// assistant_with_tool_calls before tool_result messages. + /// Verify sanitize_tool_messages preserves all tool_results when + /// each has a matching assistant tool_call. + #[test] + fn test_sanitize_preserves_tool_results_with_matching_assistant() { + let tc1 = ToolCall { + id: "call_sel_1".to_string(), + name: "search".to_string(), + arguments: serde_json::json!({"q": "test"}), + }; + let tc2 = ToolCall { + id: "call_sel_2".to_string(), + name: "http".to_string(), + arguments: serde_json::json!({"url": "https://example.com"}), + }; + let mut messages = vec![ + ChatMessage::system("You are a helpful assistant."), + ChatMessage::assistant_with_tool_calls(None, vec![tc1, tc2]), + ChatMessage::tool_result("call_sel_1", "search", "found 3 results"), + ChatMessage::tool_result("call_sel_2", "http", "200 OK"), + ]; + sanitize_tool_messages(&mut messages); + + // All tool_results must keep Role::Tool -- none should be rewritten. + assert_eq!(messages[2].role, Role::Tool); + assert_eq!(messages[2].tool_call_id, Some("call_sel_1".to_string())); + assert_eq!(messages[2].content, "found 3 results"); + + assert_eq!(messages[3].role, Role::Tool); + assert_eq!(messages[3].tool_call_id, Some("call_sel_2".to_string())); + assert_eq!(messages[3].content, "200 OK"); + } + + /// Regression: the OLD buggy worker code pushed tool_result messages + /// without a preceding assistant_with_tool_calls, causing + /// sanitize_tool_messages to rewrite them as orphaned user messages. + /// This test reproduces that buggy sequence and confirms the rewrite. + #[test] + fn test_sanitize_rewrites_orphaned_tool_results() { + let mut messages = vec![ + ChatMessage::system("You are a helpful assistant."), + // No assistant_with_tool_calls -- mimics the old bug. + ChatMessage::tool_result("call_bug_1", "search", "found 3 results"), + ChatMessage::tool_result("call_bug_2", "http", "200 OK"), + ]; + sanitize_tool_messages(&mut messages); + + // Both tool_results must be rewritten to Role::User. + assert_eq!(messages[1].role, Role::User); + assert!(messages[1].content.contains("[Tool `search` returned:")); + assert!(messages[1].content.contains("found 3 results")); + assert!(messages[1].tool_call_id.is_none()); + assert!(messages[1].name.is_none()); + + assert_eq!(messages[2].role, Role::User); + assert!(messages[2].content.contains("[Tool `http` returned:")); + assert!(messages[2].content.contains("200 OK")); + assert!(messages[2].tool_call_id.is_none()); + assert!(messages[2].name.is_none()); + } } diff --git a/src/llm/reasoning.rs b/src/llm/reasoning.rs index c75ffee3..79b0dcb2 100644 --- a/src/llm/reasoning.rs +++ b/src/llm/reasoning.rs @@ -8,7 +8,8 @@ use serde::{Deserialize, Serialize}; use crate::error::LlmError; use crate::llm::{ - ChatMessage, CompletionRequest, LlmProvider, ToolCall, ToolCompletionRequest, ToolDefinition, + ChatMessage, CompletionRequest, LlmProvider, Role, ToolCall, ToolCompletionRequest, + ToolDefinition, }; use crate::safety::SafetyLayer; @@ -460,8 +461,15 @@ impl Reasoning { pub async fn plan(&self, context: &ReasoningContext) -> Result { let system_prompt = self.build_planning_prompt(context); + let system_prompt = merge_system_messages(system_prompt, &context.messages); let mut messages = vec![ChatMessage::system(system_prompt)]; - messages.extend(context.messages.clone()); + messages.extend( + context + .messages + .iter() + .filter(|m| m.role != Role::System) + .cloned(), + ); if let Some(ref job) = context.job_description { messages.push(ChatMessage::user(format!( @@ -612,8 +620,15 @@ Respond in JSON format: None => self.build_system_prompt_with_tools(&context.available_tools), }; + let system_prompt = merge_system_messages(system_prompt, &context.messages); let mut messages = vec![ChatMessage::system(system_prompt)]; - messages.extend(context.messages.clone()); + messages.extend( + context + .messages + .iter() + .filter(|m| m.role != Role::System) + .cloned(), + ); let effective_tools = if context.force_text { Vec::new() @@ -1026,6 +1041,22 @@ pub struct SuccessEvaluation { pub suggestions: Vec, } +/// Merge the reasoning method's system prompt with any system messages already +/// present in the conversation context. Strict LLM providers (e.g. Qwen) +/// reject conversations with system messages that are not at the very +/// beginning, so we concatenate all system content into a single prompt. +fn merge_system_messages(primary: String, context_messages: &[ChatMessage]) -> String { + let extra: Vec<&str> = context_messages + .iter() + .filter(|m| m.role == Role::System) + .map(|m| m.content.as_str()) + .collect(); + if extra.is_empty() { + return primary; + } + format!("{}\n\n---\n\n{}", primary, extra.join("\n\n")) +} + /// Extract JSON from text that might contain other content. fn extract_json(text: &str) -> Option<&str> { // Find the first { and last } to extract JSON @@ -2198,6 +2229,58 @@ That's my plan."#; assert!(cleaned.contains("Here are the results.")); } + // ---- merge_system_messages: duplicate system message regression (Bug #597) ---- + + #[test] + fn test_merge_system_messages_no_system_in_context() { + let messages = vec![ + ChatMessage::user("Hello"), + ChatMessage::assistant("Hi there"), + ]; + let result = merge_system_messages("primary prompt".into(), &messages); + assert_eq!(result, "primary prompt"); + } + + #[test] + fn test_merge_system_messages_merges_worker_system() { + let messages = vec![ + ChatMessage::system("You are an autonomous agent working on a job.\n\nJob: Test Job"), + ChatMessage::user("Do the thing"), + ]; + let result = merge_system_messages("planning prompt".into(), &messages); + assert!( + result.contains("planning prompt"), + "must contain the primary prompt" + ); + assert!( + result.contains("autonomous agent"), + "must contain worker system text" + ); + assert!( + result.contains("Test Job"), + "must contain job description from worker system message" + ); + } + + #[test] + fn test_merge_system_messages_multiple_system() { + let messages = vec![ + ChatMessage::system("First system instruction"), + ChatMessage::system("Second system instruction"), + ChatMessage::user("Hello"), + ]; + let result = merge_system_messages("primary".into(), &messages); + assert!(result.contains("primary"), "must contain primary prompt"); + assert!( + result.contains("First system instruction"), + "must contain first system message" + ); + assert!( + result.contains("Second system instruction"), + "must contain second system message" + ); + } + #[test] fn test_system_prompt_without_tools_omits_tools_section() { let reasoning = make_test_reasoning(); diff --git a/src/main.rs b/src/main.rs index f50eb754..3a6f4ff4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -475,6 +475,7 @@ async fn async_main() -> anyhow::Result<()> { let mut sse_sender: Option< tokio::sync::broadcast::Sender, > = None; + let mut routine_engine_slot: Option = None; if let Some(ref gw_config) = config.channels.gateway { let mut gw = GatewayChannel::new(gw_config.clone()).with_llm_provider(Arc::clone(&components.llm)); @@ -528,10 +529,11 @@ async fn async_main() -> anyhow::Result<()> { tracing::info!("Web UI: http://{}:{}/", gw_config.host, gw_config.port); - // Capture SSE sender before moving gw into channels. + // Capture SSE sender and routine engine slot before moving gw into channels. // IMPORTANT: This must come after all `with_*` calls since `rebuild_state` // creates a new SseManager, which would orphan this sender. sse_sender = Some(gw.state().sse.sender()); + routine_engine_slot = Some(Arc::clone(&gw.state().routine_engine)); channel_names.push("gateway".to_string()); channels.add(Box::new(gw)).await; @@ -678,7 +680,7 @@ async fn async_main() -> anyhow::Result<()> { )), }; - let agent = Agent::new( + let mut agent = Agent::new( config.agent.clone(), deps, channels, @@ -692,6 +694,11 @@ async fn async_main() -> anyhow::Result<()> { // Fill the scheduler slot now that Agent (and its Scheduler) exist. *scheduler_slot.write().await = Some(agent.scheduler()); + // Give the agent the routine engine slot so it can expose the engine to the gateway. + if let Some(slot) = routine_engine_slot { + agent.set_routine_engine_slot(slot); + } + agent.run().await?; // ── Shutdown ──────────────────────────────────────────────────────── diff --git a/src/tools/builtin/routine.rs b/src/tools/builtin/routine.rs index 59a57e0c..2fddec29 100644 --- a/src/tools/builtin/routine.rs +++ b/src/tools/builtin/routine.rs @@ -620,9 +620,13 @@ impl Tool for RoutineFireTool { .map_err(|e| ToolError::ExecutionFailed(format!("DB error: {e}")))? .ok_or_else(|| ToolError::ExecutionFailed(format!("routine '{}' not found", name)))?; - let run_id = self.engine.fire_manual(routine.id).await.map_err(|e| { - ToolError::ExecutionFailed(format!("failed to fire routine '{}': {e}", name)) - })?; + let run_id = self + .engine + .fire_manual(routine.id, None) + .await + .map_err(|e| { + ToolError::ExecutionFailed(format!("failed to fire routine '{}': {e}", name)) + })?; let result = serde_json::json!({ "name": name, diff --git a/tests/openai_compat_integration.rs b/tests/openai_compat_integration.rs index d9dd3745..7f3f1e7f 100644 --- a/tests/openai_compat_integration.rs +++ b/tests/openai_compat_integration.rs @@ -211,6 +211,7 @@ async fn start_test_server_with_provider( chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), }); @@ -700,6 +701,7 @@ async fn test_no_llm_provider_returns_503() { chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), }); diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index 0016ba4e..da44f766 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -59,6 +59,7 @@ async fn start_test_server() -> ( chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), registry_entries: Vec::new(), cost_guard: None, + routine_engine: Arc::new(tokio::sync::RwLock::new(None)), startup_time: std::time::Instant::now(), });