diff --git a/backend/windmill-api/src/flow_conversations.rs b/backend/windmill-api/src/flow_conversations.rs index 96ba5f3af6..efe11c2045 100644 --- a/backend/windmill-api/src/flow_conversations.rs +++ b/backend/windmill-api/src/flow_conversations.rs @@ -88,7 +88,7 @@ async fn list_conversations( .offset(offset as i64); let sql = sqlb.sql().map_err(|e| { - windmill_common::error::Error::InternalErr(format!("Failed to build SQL: {}", e)) + windmill_common::error::Error::internal_err(format!("Failed to build SQL: {}", e)) })?; let conversations = sqlx::query_as::(&sql) @@ -123,6 +123,12 @@ pub async fn get_or_create_conversation_with_id( return Ok(existing); } + // Truncate title to 25 char characters max + let title = if title.len() > 25 { + format!("{}...", &title[..25]) + } else { + title.to_string() + }; // Create new conversation with provided ID let conversation = sqlx::query_as!( FlowConversation, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e721facc6e..813e912d6e 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -3942,7 +3942,7 @@ async fn handle_chat_conversation_messages( w_id: &str, flow_path: &str, run_query: &RunJobQuery, - args: &PushArgsOwned, + user_message_raw: Option<&Box>, uuid: Uuid, ) -> error::Result<()> { let memory_id = run_query.memory_id.ok_or_else(|| { @@ -3951,13 +3951,17 @@ async fn handle_chat_conversation_messages( ) })?; - let user_msg_raw = args.args.get("user_message").ok_or_else(|| { + let user_message_raw = user_message_raw.ok_or_else(|| { windmill_common::error::Error::BadRequest( "user_message argument is required for chat-enabled flows".to_string(), ) })?; - let user_msg = serde_json::from_str::(user_msg_raw.get())?; + // Deserialize the RawValue to get the actual string without quotes + let user_message: String = serde_json::from_str(user_message_raw.get()) + .map_err(|e| windmill_common::error::Error::BadRequest( + format!("Failed to deserialize user_message: {}", e) + ))?; // Create conversation with provided ID (or get existing one) flow_conversations::get_or_create_conversation_with_id( @@ -3965,7 +3969,7 @@ async fn handle_chat_conversation_messages( w_id, flow_path, &authed.username, - &user_msg, + &user_message, memory_id, ) .await?; @@ -3975,7 +3979,7 @@ async fn handle_chat_conversation_messages( tx, memory_id, MessageType::User, - &user_msg, + &user_message, None, // No job_id for user message w_id, ) @@ -4082,7 +4086,7 @@ pub async fn run_flow_by_path_inner( apply_preprocessor: !run_query.skip_preprocessor.unwrap_or(false) && has_preprocessor.unwrap_or(false), }, - PushArgs { args: &args.args, extra: args.extra.clone() }, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), email, permissioned_as, @@ -4120,7 +4124,7 @@ pub async fn run_flow_by_path_inner( &w_id, &flow_path.to_string(), &run_query, - &args, + args.args.get("user_message"), uuid, ) .await?; @@ -5581,7 +5585,7 @@ pub async fn run_wait_result_flow_by_path_internal( apply_preprocessor: !run_query.skip_preprocessor.unwrap_or(false) && has_preprocessor.unwrap_or(false), }, - PushArgs { args: &args.args, extra: args.extra.clone() }, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), email, permissioned_as, @@ -5619,7 +5623,7 @@ pub async fn run_wait_result_flow_by_path_internal( &w_id, &flow_path.to_string(), &run_query, - &args, + args.args.get("user_message"), uuid, ) .await?; diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 912ba5c1f7..1aaecc2fe3 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -457,6 +457,7 @@ pub async fn store_pull_query(wc: &WorkerConfig) { pub const TMP_DIR: &str = "/tmp/windmill"; pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs"); +pub const TMP_MEMORY_DIR: &str = concatcp!(TMP_DIR, "/memory"); pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub"); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 3010874ac5..d0055bbe3a 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -746,11 +746,6 @@ lazy_static::lazy_static! { pub static ref MAX_RESULT_SIZE_MB: usize = std::env::var("MAX_RESULT_SIZE_MB").unwrap_or("500".to_string()).parse().unwrap_or(500); } -#[derive(Deserialize)] -struct OutputWrapper { - output: String, -} - pub async fn add_completed_job( db: &Pool, queued_job: &MiniPulledJob, @@ -831,53 +826,41 @@ pub async fn add_completed_job( // Update conversation message if it's a flow and it's done (both success and error cases) if !skipped && flow_is_done { let chat_input_enabled = queued_job.parse_chat_input_enabled(); + let value = serde_json::to_value(result.0) + .map_err(|e| Error::internal_err(format!("Failed to serialize result: {e}")))?; if chat_input_enabled.unwrap_or(false) { - let content = if let Ok(wrapper) = serde_json::from_value::( - serde_json::to_value(result.0).unwrap_or(serde_json::Value::Null), - ) { - // Successfully deserialized to OutputWrapper, use the output field - wrapper.output - } else { - // No string output field, use the whole result - serde_json::to_value(result.0) - .ok() - .and_then(|v| { - if let serde_json::Value::String(s) = v { - Some(s) - } else { - serde_json::to_string_pretty(&v).ok() - } - }) - .unwrap_or_else(|| { - if success { - "Job completed successfully".to_string() - } else { - "Job failed".to_string() - } - }) + let content = match value { + // If it's an Object with "output" key AND the output is a String, return it + serde_json::Value::Object(mut map) + if map.contains_key("output") + && matches!(map.get("output"), Some(serde_json::Value::String(_))) => + { + if let Some(serde_json::Value::String(s)) = map.remove("output") { + s + } else { + // prettify the whole result + serde_json::to_string_pretty(&map) + .unwrap_or_else(|e| format!("Failed to serialize result: {e}")) + } + } + // Otherwise, if the whole value is a String, return it + serde_json::Value::String(s) => s, + // Otherwise, prettify the whole result + v => serde_json::to_string_pretty(&v) + .unwrap_or_else(|e| format!("Failed to serialize result: {e}")), }; - // check if flow_conversation_message exists - let flow_conversation_message_exists = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM flow_conversation_message WHERE job_id = $1 AND message_type = 'assistant')", - queued_job.id - ) - .fetch_one(db) - .await?; - - if flow_conversation_message_exists.unwrap_or(false) { - // Update the assistant message using direct DB access - let _ = sqlx::query!( - "UPDATE flow_conversation_message + // Update the assistant message + let _ = sqlx::query!( + "UPDATE flow_conversation_message SET content = $1 WHERE job_id = $2 ", - content, - queued_job.id, - ) - .execute(db) - .await; - } + content, + queued_job.id, + ) + .execute(db) + .await; } } diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index c927c09354..7903e459c7 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -471,13 +471,12 @@ pub async fn run_agent( // Take the last n messages let start_idx = loaded_messages.len().saturating_sub(context_length); let mut messages_to_load = loaded_messages[start_idx..].to_vec(); + let first_non_tool_message_index = + messages_to_load.iter().position(|m| m.role != "tool"); - // Remove the first message if its role is "tool" to avoid OpenAI API error - // "messages with role 'tool' must be a response to a preceeding message with 'tool_calls'" - if let Some(first_msg) = messages_to_load.first() { - if first_msg.role == "tool" { - messages_to_load.remove(0); - } + // Remove the first messages if their role is "tool" to avoid OpenAI API error + if let Some(index) = first_non_tool_message_index { + messages_to_load = messages_to_load[index..].to_vec(); } messages.extend(messages_to_load); @@ -934,7 +933,7 @@ pub async fn run_agent( let (handle_result, updated_occupancy) = join_handle.await.map_err(|e| { Error::internal_err(format!( - "Tool execution task panicked: {}", + "Tool execution task failed: {}", e )) })?; diff --git a/backend/windmill-worker/src/memory_common.rs b/backend/windmill-worker/src/memory_common.rs index 3b934c2ee3..16821c555b 100644 --- a/backend/windmill-worker/src/memory_common.rs +++ b/backend/windmill-worker/src/memory_common.rs @@ -2,12 +2,11 @@ use crate::ai::types::OpenAIMessage; use std::path::PathBuf; use tokio::{fs, io::AsyncWriteExt}; use uuid::Uuid; -use windmill_common::worker::TMP_LOGS_DIR; +use windmill_common::worker::TMP_MEMORY_DIR; /// Get the file path for storing memory for a specific AI agent step pub fn path_for(workspace_id: &str, conversation_id: Uuid, step_id: &str) -> PathBuf { - PathBuf::from(TMP_LOGS_DIR) - .join("memory") + PathBuf::from(TMP_MEMORY_DIR) .join(workspace_id) .join(conversation_id.to_string()) .join(format!("{step_id}.json")) @@ -61,8 +60,7 @@ pub async fn delete_conversation_from_disk( workspace_id: &str, conversation_id: Uuid, ) -> anyhow::Result<()> { - let conversation_path = PathBuf::from(TMP_LOGS_DIR) - .join("memory") + let conversation_path = PathBuf::from(TMP_MEMORY_DIR) .join(workspace_id) .join(conversation_id.to_string()); diff --git a/docker-compose.yml b/docker-compose.yml index f8831523ae..1770a1e4db 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -46,6 +46,7 @@ services: condition: service_healthy volumes: - worker_logs:/tmp/windmill/logs + logging: *default-logging windmill_worker: @@ -72,6 +73,9 @@ services: - /var/run/docker.sock:/var/run/docker.sock - worker_dependency_cache:/tmp/windmill/cache - worker_logs:/tmp/windmill/logs + # for AI agent memory + - worker_memory:/tmp/windmill/memory + logging: *default-logging ## This worker is specialized for "native" jobs. Native jobs run in-process and thus are much more lightweight than other jobs @@ -188,6 +192,7 @@ volumes: db_data: null worker_dependency_cache: null worker_logs: null + worker_memory: null windmill_index: null lsp_cache: null caddy_data: null diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index f7e78a4449..69541052db 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -38,7 +38,7 @@ import { getStepHistoryLoaderContext } from './stepHistoryLoader.svelte' import { aiChatManager } from './copilot/chat/AIChatManager.svelte' import { stateSnapshot } from '$lib/svelte5Utils.svelte' - import FlowChatInterface from './flows/FlowChatInterface.svelte' + import FlowChatInterface from './flows/conversations/FlowChatInterface.svelte' interface Props { previewMode: 'upTo' | 'whole' diff --git a/frontend/src/lib/components/details/DetailPageLayout.svelte b/frontend/src/lib/components/details/DetailPageLayout.svelte index 57ecc6cff0..a1f4ea4734 100644 --- a/frontend/src/lib/components/details/DetailPageLayout.svelte +++ b/frontend/src/lib/components/details/DetailPageLayout.svelte @@ -2,12 +2,14 @@ import { Tabs, Tab, TabContent } from '$lib/components/common' import { Pane, Splitpanes } from 'svelte-splitpanes' import DetailPageDetailPanel from './DetailPageDetailPanel.svelte' + import FlowViewerInner from '../FlowViewerInner.svelte' interface Props { isOperator?: boolean flow_json?: any | undefined selected: string forceSmallScreen?: boolean + isChatMode?: boolean header?: import('svelte').Snippet form?: import('svelte').Snippet scriptRender?: import('svelte').Snippet @@ -21,6 +23,7 @@ flow_json = undefined, selected = $bindable(), forceSmallScreen = false, + isChatMode = false, header, form, scriptRender: script, @@ -74,8 +77,10 @@ {@render header?.()}
- Run form - Inputs + {isChatMode ? 'Chat' : 'Run form'} + {#if !isChatMode} + Inputs + {/if} {#if !isOperator} Triggers {/if} @@ -97,6 +102,11 @@ {@render triggers?.()} + {#if flow_json} + + + + {/if} {@render script?.()} diff --git a/frontend/src/lib/components/flows/content/FlowInput.svelte b/frontend/src/lib/components/flows/content/FlowInput.svelte index ea15d8f471..80d8c4c4b2 100644 --- a/frontend/src/lib/components/flows/content/FlowInput.svelte +++ b/frontend/src/lib/components/flows/content/FlowInput.svelte @@ -44,7 +44,7 @@ import { refreshStateStore } from '$lib/svelte5Utils.svelte' import type { ScriptLang } from '$lib/gen' import { deepEqual } from 'fast-equals' - import FlowChatInterface from '../FlowChatInterface.svelte' + import FlowChatInterface from '$lib/components/flows/conversations/FlowChatInterface.svelte' import Toggle from '$lib/components/Toggle.svelte' import { AI_AGENT_SCHEMA } from '../flowInfers' import { nextId } from '../flowModuleNextId' diff --git a/frontend/src/lib/components/flows/FlowChatInterface.svelte b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte similarity index 92% rename from frontend/src/lib/components/flows/FlowChatInterface.svelte rename to frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte index 3919627e50..7218f3caff 100644 --- a/frontend/src/lib/components/flows/FlowChatInterface.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte @@ -21,6 +21,7 @@ interface ChatMessage extends FlowConversationMessage { loading?: boolean + streaming?: boolean } let { @@ -48,17 +49,6 @@ const conversationsCache = $state>({}) - // Auto-scroll to bottom when messages change - $effect(() => { - const scroll = async () => { - if (messages.length > 0) { - await tick() - scrollToBottom() - } - } - scroll() - }) - // Cleanup EventSource on unmount $effect(() => { return () => { @@ -156,6 +146,14 @@ } } + function scrollToUserMessage(messageId: string) { + if (!messagesContainer) return + const messageElement = messagesContainer.querySelector(`[data-message-id="${messageId}"]`) + if (messageElement) { + messageElement.scrollIntoView({ behavior: 'smooth', block: 'start' }) + } + } + async function pollJobResult(jobId: string, messageId: string) { try { const result = await waitJob(jobId) @@ -270,11 +268,15 @@ message_type: 'assistant', conversation_id: currentConversationId, job_id: '', - loading: true + loading: true, + streaming: Boolean(useStreaming && path) } messages = [...messages, assistantMessage] + await tick() + scrollToUserMessage(userMessage.id) + if (useStreaming && path) { // Close any existing EventSource if (currentEventSource) { @@ -332,7 +334,8 @@ ? { ...msg, content: finalContent, - loading: false + loading: false, + streaming: false } : msg ) @@ -353,7 +356,8 @@ ? { ...msg, content: accumulatedContent || 'Stream error occurred', - loading: false + loading: false, + streaming: false } : msg ) @@ -369,7 +373,8 @@ ? { ...msg, content: 'Failed to connect to stream', - loading: false + loading: false, + streaming: false } : msg ) @@ -384,8 +389,6 @@ } pollJobResult(jobId, assistantMessageId) } - - scrollToBottom() } catch (error) { console.error('Error running flow:', error) sendUserToast('Failed to run flow: ' + error, true) @@ -411,31 +414,33 @@ } -
+
{#if deploymentInProgress} {/if} {#if isLoadingMessages} -
- +
+
{:else if messages.length === 0} -
+

Start a conversation

Send a message to run the flow and see the results

{:else} - {#each messages as message (message.id)} - - {/each} +
+ {#each messages as message (message.id)} + + {/each} +
{/if}
diff --git a/frontend/src/lib/components/flows/FlowChatMessage.svelte b/frontend/src/lib/components/flows/conversations/FlowChatMessage.svelte similarity index 80% rename from frontend/src/lib/components/flows/FlowChatMessage.svelte rename to frontend/src/lib/components/flows/conversations/FlowChatMessage.svelte index 44f5a0e625..9bb8172e2c 100644 --- a/frontend/src/lib/components/flows/FlowChatMessage.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChatMessage.svelte @@ -7,13 +7,16 @@ import LinkRenderer from '$lib/components/copilot/chat/LinkRenderer.svelte' interface Props { - message: FlowConversationMessage & { loading?: boolean } + message: FlowConversationMessage & { loading?: boolean; streaming?: boolean } } let { message }: Props = $props() -
+
+
{#if flow?.archived} This flow was archived @@ -612,8 +617,7 @@ {#if chatInputEnabled}
-
+
{/if}
-
- { - if (e.detail) { - stepDetail = e.detail - rightPaneSelected = 'flow_step' - } else { - stepDetail = undefined - rightPaneSelected = 'saved_inputs' - } - }} - on:triggerDetail={(e) => { - rightPaneSelected = 'triggers' - }} - /> -
+ {#if !chatInputEnabled} +
+ { + if (e.detail) { + stepDetail = e.detail + rightPaneSelected = 'flow_step' + } else { + stepDetail = undefined + rightPaneSelected = 'saved_inputs' + } + }} + on:triggerDetail={(e) => { + rightPaneSelected = 'triggers' + }} + /> +
+ {/if}
{/if} {/snippet}