chore(flow): clean flow conversation code (#6755)

* use own folder for memory * fixes * better chat interface * fix export tab * move in folder * dont show flow graph if chat mode * fix * fix too long title * fix user message * fix * fix * remove from server * cleaner * cleaning * cleaning * cleaning
2026-08-18 16:02:10 +00:00 · 2025-10-06 17:10:29 +02:00
parent 686cf8dd69
commit 820437c897
15 changed files with 148 additions and 128 deletions
@@ -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::<Postgres, FlowConversation>(&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,
@@ -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<serde_json::value::RawValue>>,
    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::<String>(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?;
@@ -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");

@@ -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<T: Serialize + Send + Sync + ValidableJson>(
    db: &Pool<Postgres>,
    queued_job: &MiniPulledJob,
@@ -831,53 +826,41 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
    // 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::<OutputWrapper>(
-                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;
        }
    }

@@ -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
                                        ))
                                    })?;
@@ -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());

@@ -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
@@ -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'
@@ -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?.()}
 			<div class="grow min-h-0 w-full flex flex-col">
 				<Tabs bind:selected={mobileTab} wrapperClass="flex-none">
-					<Tab value="form">Run form</Tab>
-					<Tab value="saved_inputs">Inputs</Tab>
+					<Tab value="form">{isChatMode ? 'Chat' : 'Run form'}</Tab>
+					{#if !isChatMode}
+						<Tab value="saved_inputs">Inputs</Tab>
+					{/if}
 					{#if !isOperator}
 						<Tab value="triggers">Triggers</Tab>
 					{/if}
@@ -97,6 +102,11 @@
 							<TabContent value="triggers" class="flex flex-col flex-1 h-full mt-[-2px]">
 								{@render triggers?.()}
 							</TabContent>
+							{#if flow_json}
+								<TabContent value="raw" class="flex flex-col flex-1 h-full overflow-auto p-2">
+									<FlowViewerInner flow={flow_json} />
+								</TabContent>
+							{/if}
 							<TabContent value="script" class="flex flex-col flex-1 h-full">
 								{@render script?.()}
 							</TabContent>
@@ -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'
@@ -21,6 +21,7 @@

 	interface ChatMessage extends FlowConversationMessage {
 		loading?: boolean
+		streaming?: boolean
 	}

 	let {
@@ -48,17 +49,6 @@

 	const conversationsCache = $state<Record<string, ChatMessage[]>>({})

-	// 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 @@
 	}
 </script>

-<div class="flex flex-col h-full w-full max-w-7xl mx-auto">
+<div class="flex flex-col h-full w-full">
 	<div class="flex-1 flex flex-col min-h-0 w-full">
 		<!-- Messages Container -->
 		<div
 			bind:this={messagesContainer}
-			class="flex-1 overflow-y-auto p-4 space-y-4 bg-background"
+			class="flex-1 overflow-y-auto p-4 bg-background"
 			onscroll={handleScroll}
 		>
 			{#if deploymentInProgress}
 				<Alert type="warning" title="Deployment in progress" size="xs" />
 			{/if}
 			{#if isLoadingMessages}
-				<div class="flex items-center justify-center">
-					<Loader2 size={24} class="animate-spin" />
+				<div class="flex items-center justify-center h-full">
+					<Loader2 size={32} class="animate-spin" />
 				</div>
 			{:else if messages.length === 0}
-				<div class="text-center text-tertiary py-8">
+				<div class="text-center text-tertiary flex items-center justify-center flex-col h-full">
 					<MessageCircle size={48} class="mx-auto mb-4 opacity-50" />
 					<p class="text-lg font-medium">Start a conversation</p>
 					<p class="text-sm">Send a message to run the flow and see the results</p>
 				</div>
 			{:else}
-				{#each messages as message (message.id)}
-					<FlowChatMessage {message} />
-				{/each}
+				<div class="max-w-7xl mx-auto space-y-4">
+					{#each messages as message (message.id)}
+						<FlowChatMessage {message} />
+					{/each}
+				</div>
 			{/if}
 		</div>

@@ -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()
 </script>

-<div class="flex {message.message_type === 'user' ? 'justify-end' : 'justify-start'}">
+<div
+	class={`flex ${message.message_type === 'user' ? 'justify-end' : 'justify-start'} ${message.loading || message.streaming ? 'min-h-[200px] items-start' : ''}`}
+	data-message-id={message.id}
+>
 	<div
 		class="max-w-[90%] min-w-0 rounded-lg p-3 {message.message_type === 'user'
 			? 'bg-surface-secondary'
@@ -27,7 +30,7 @@
 				<span>Processing...</span>
 			</div>
 		{:else if message.content}
-			<div class="prose prose-sm dark:prose-invert prose-ul:!pl-6 break-words">
+			<div class="prose prose-sm dark:prose-invert prose-ul:!pl-6 break-words whitespace-pre-wrap">
 				<Markdown
 					md={message.content}
 					plugins={[
@@ -30,7 +30,7 @@ export const AI_AGENT_SCHEMA = {
 		streaming: {
 			type: 'boolean',
 			description: 'Whether to stream the output of the AI agent.',
-			default: false,
+			default: true,
 			showExpr: "fields.output_type === 'text'"
 		},
 		messages_context_length: {
@@ -17,8 +17,8 @@
 	import { Badge as HeaderBadge, Alert } from '$lib/components/common'
 	import MoveDrawer from '$lib/components/MoveDrawer.svelte'
 	import RunForm from '$lib/components/RunForm.svelte'
-	import FlowChatInterface from '$lib/components/flows/FlowChatInterface.svelte'
-	import FlowConversationsSidebar from '$lib/components/flows/FlowConversationsSidebar.svelte'
+	import FlowChatInterface from '$lib/components/flows/conversations/FlowChatInterface.svelte'
+	import FlowConversationsSidebar from '$lib/components/flows/conversations/FlowConversationsSidebar.svelte'
 	import ShareModal from '$lib/components/ShareModal.svelte'
 	import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
@@ -424,6 +424,8 @@
 			flowChatInterface.clearMessages()
 		}

+		flowChatInterface?.focusInput()
+
 		return newConversationId
 	}

@@ -508,6 +510,7 @@
 	bind:selected={rightPaneSelected}
 	isOperator={$userStore?.operator}
 	forceSmallScreen={chatInputEnabled}
+	isChatMode={chatInputEnabled}
 	flow_json={{
 		value: flow?.value,
 		summary: flow?.summary,
@@ -583,7 +586,9 @@
 		{#if flow}
 			<div class="flex flex-col h-full justify-between">
 				<div
-					class="w-full {chatInputEnabled ? 'p-4 mt-8' : 'max-w-3xl p-8'} mx-auto gap-2 bg-surface"
+					class="w-full {chatInputEnabled
+						? 'p-3 flex flex-col h-full'
+						: 'max-w-3xl p-8'} mx-auto gap-2 bg-surface"
 				>
 					{#if flow?.archived}
 						<Alert type="error" title="Archived">This flow was archived</Alert>
@@ -612,8 +617,7 @@
 					{#if chatInputEnabled}
 						<!-- Chat Layout with Sidebar -->
 						<div
-							class="flex border border-gray-200 dark:border-gray-700 rounded-lg overflow-hidden"
-							style="height: 600px;"
+							class="flex border border-gray-200 dark:border-gray-700 rounded-lg overflow-hidden flex-1"
 						>
 							<div class="flex-shrink-0">
 								<FlowConversationsSidebar
@@ -625,7 +629,7 @@
 									onDeleteConversation={handleDeleteConversation}
 								/>
 							</div>
-							<div class="flex-1 min-h-0">
+							<div class="flex-1">
 								<FlowChatInterface
 									bind:this={flowChatInterface}
 									onRunFlow={runFlowForChat}
@@ -704,27 +708,29 @@
 						</div>
 					{/if}
 				</div>
-				<div class="mt-8">
-					<FlowGraphViewer
-						triggerNode={true}
-						download
-						{flow}
-						overflowAuto
-						noSide={true}
-						on:select={(e) => {
-							if (e.detail) {
-								stepDetail = e.detail
-								rightPaneSelected = 'flow_step'
-							} else {
-								stepDetail = undefined
-								rightPaneSelected = 'saved_inputs'
-							}
-						}}
-						on:triggerDetail={(e) => {
-							rightPaneSelected = 'triggers'
-						}}
-					/>
-				</div>
+				{#if !chatInputEnabled}
+					<div class="mt-8">
+						<FlowGraphViewer
+							triggerNode={true}
+							download
+							{flow}
+							overflowAuto
+							noSide={true}
+							on:select={(e) => {
+								if (e.detail) {
+									stepDetail = e.detail
+									rightPaneSelected = 'flow_step'
+								} else {
+									stepDetail = undefined
+									rightPaneSelected = 'saved_inputs'
+								}
+							}}
+							on:triggerDetail={(e) => {
+								rightPaneSelected = 'triggers'
+							}}
+						/>
+					</div>
+				{/if}
 			</div>
 		{/if}
 	{/snippet}