feat(backend): replace messages/messages_context_length with history oneOf field

Replace the separate 'messages' array and 'messages_context_length' fields with
a single 'history' field that uses a oneOf discriminator.

The 'history' field can be either:
- 'auto' mode: automatically manages conversation history with memory, takes a
  'context_length' number parameter
- 'manual' mode: bypasses memory and uses explicitly provided messages array

Backward compatibility is maintained: if 'messages_context_length' is provided
in the old schema format, it is automatically converted to 'auto' mode with the
specified context_length.

Co-authored-by: centdix <centdix@users.noreply.github.com>

Author: github-actions[bot]

backend/windmill-worker/src/ai/types.rs

@@ -106,20 +106,37 @@ impl Default for OutputType {
     }
 }
 
+#[derive(Deserialize, Debug, Clone)]
+#[serde(tag = "mode", rename_all = "lowercase")]
+pub enum History {
+    Auto {
+        #[serde(default)]
+        context_length: usize,
+    },
+    Manual {
+        messages: Vec<OpenAIMessage>,
+    },
+}
+
 #[derive(Deserialize, Debug)]
 pub struct AIAgentArgs {
     pub provider: ProviderWithResource,
     pub system_prompt: Option<String>,
     pub user_message: Option<String>,
-    pub messages: Option<Vec<OpenAIMessage>>,
     pub temperature: Option<f32>,
     pub max_completion_tokens: Option<u32>,
     pub output_schema: Option<OpenAPISchema>,
     pub output_type: Option<OutputType>,
     pub user_images: Option<Vec<S3Object>>,
     pub streaming: Option<bool>,
-    pub messages_context_length: Option<usize>,
     pub max_iterations: Option<usize>,
+
+    // New history field (replaces messages and messages_context_length)
+    pub history: Option<History>,
+
+    // Deprecated: for backward compatibility only
+    #[serde(default)]
+    pub messages_context_length: Option<usize>,
 }
 
 #[derive(Deserialize, Debug)]

backend/windmill-worker/src/ai_executor.rs

@@ -394,12 +394,19 @@ pub async fn run_agent(
             vec![]
         };
 
-    // Check if explicit messages were provided
-    let use_explicit_messages = args
-        .messages
-        .as_ref()
-        .map(|m| !m.is_empty())
-        .unwrap_or(false);
+    // Fetch flow context for input transforms context, chat and memory
+    let mut flow_context = get_flow_context(db, job).await;
+
+    // Determine history mode with backward compatibility
+    let history = args.history.clone().or_else(|| {
+        // Backward compatibility: if messages_context_length is set, use auto mode
+        args.messages_context_length.map(|context_length| {
+            History::Auto { context_length }
+        })
+    });
+
+    // Determine if we're using manual messages (which bypasses memory)
+    let use_manual_messages = matches!(history, Some(History::Manual { .. }));
 
     // Check if user_message is provided and non-empty
     let has_user_message = args
@@ -408,55 +415,57 @@ pub async fn run_agent(
         .map(|m| !m.is_empty())
         .unwrap_or(false);
 
-    // Validate: at least one of messages or user_message must be provided
-    if !use_explicit_messages && !has_user_message {
+    // Validate: at least one of history with manual messages or user_message must be provided
+    if !use_manual_messages && !has_user_message {
         return Err(Error::internal_err(
-            "Either 'messages' or 'user_message' must be provided".to_string(),
+            "Either 'history' with manual messages or 'user_message' must be provided".to_string(),
         ));
     }
 
-    // Fetch flow context for input transforms context, chat and memory
-    let mut flow_context = get_flow_context(db, job).await;
-
-    // Load messages if given, this will bypass memory
-    if let Some(ref explicit_messages) = args.messages {
-        if !explicit_messages.is_empty() {
-            // Use explicitly provided messages (bypass memory)
-            messages.extend(explicit_messages.clone());
-        }
-    }
+    // Load messages based on history mode
+    if matches!(output_type, OutputType::Text) {
+        match history {
+            Some(History::Manual { messages: manual_messages }) => {
+                // Use explicitly provided messages (bypass memory)
+                if !manual_messages.is_empty() {
+                    messages.extend(manual_messages);
+                }
+            }
+            Some(History::Auto { context_length }) if context_length > 0 => {
+                // Auto mode: load from memory
+                if let Some(step_id) = job.flow_step_id.as_deref() {
+                    if let Some(memory_id) = flow_context
+                        .flow_status
+                        .as_ref()
+                        .and_then(|fs| fs.memory_id)
+                    {
+                        // Read messages from memory
+                        match read_from_memory(db, &job.workspace_id, memory_id, step_id).await {
+                            Ok(Some(loaded_messages)) => {
+                                // 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 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();
+                                }
 
-    if !use_explicit_messages && matches!(output_type, OutputType::Text) {
-        if let Some(context_length) = args.messages_context_length.filter(|&n| n > 0) {
-            if let Some(step_id) = job.flow_step_id.as_deref() {
-                if let Some(memory_id) = flow_context
-                    .flow_status
-                    .as_ref()
-                    .and_then(|fs| fs.memory_id)
-                {
-                    // Read messages from memory
-                    match read_from_memory(db, &job.workspace_id, memory_id, step_id).await {
-                        Ok(Some(loaded_messages)) => {
-                            // 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 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);
+                            }
+                            Ok(None) => {}
+                            Err(e) => {
+                                tracing::error!("Failed to read memory for step {}: {}", step_id, e);
                             }
-
-                            messages.extend(messages_to_load);
-                        }
-                        Ok(None) => {}
-                        Err(e) => {
-                            tracing::error!("Failed to read memory for step {}: {}", step_id, e);
                         }
                     }
                 }
             }
+            _ => {
+                // No history or context_length is 0 - don't load any messages
+            }
         }
     }
 
@@ -921,37 +930,39 @@ pub async fn run_agent(
         }
     }
 
-    // Persist complete conversation to memory at the end (only if context length is set)
-    // Skip memory persistence if explicit messages were provided (bypass memory entirely)
+    // Persist complete conversation to memory at the end (only if in auto mode with context length)
+    // Skip memory persistence if using manual messages (bypass memory entirely)
     // final_messages contains the complete history (old messages + new ones)
-    if matches!(output_type, OutputType::Text) && !use_explicit_messages {
-        if let Some(context_length) = args.messages_context_length.filter(|&n| n > 0) {
-            if let Some(step_id) = job.flow_step_id.as_deref() {
-                // Extract OpenAIMessages from final_messages
-                let all_messages: Vec<OpenAIMessage> =
-                    final_messages.iter().map(|m| m.message.clone()).collect();
-
-                if !all_messages.is_empty() {
-                    // Keep only the last n messages
-                    let start_idx = all_messages.len().saturating_sub(context_length);
-                    let messages_to_persist = all_messages[start_idx..].to_vec();
-
-                    if let Some(memory_id) = flow_context.flow_status.and_then(|fs| fs.memory_id) {
-                        if let Err(e) = write_to_memory(
-                            db,
-                            &job.workspace_id,
-                            memory_id,
-                            step_id,
-                            &messages_to_persist,
-                        )
-                        .await
-                        {
-                            tracing::error!(
-                                "Failed to persist {} messages to memory for step {}: {}",
-                                messages_to_persist.len(),
+    if matches!(output_type, OutputType::Text) && !use_manual_messages {
+        if let Some(History::Auto { context_length }) = history {
+            if context_length > 0 {
+                if let Some(step_id) = job.flow_step_id.as_deref() {
+                    // Extract OpenAIMessages from final_messages
+                    let all_messages: Vec<OpenAIMessage> =
+                        final_messages.iter().map(|m| m.message.clone()).collect();
+
+                    if !all_messages.is_empty() {
+                        // Keep only the last n messages
+                        let start_idx = all_messages.len().saturating_sub(context_length);
+                        let messages_to_persist = all_messages[start_idx..].to_vec();
+
+                        if let Some(memory_id) = flow_context.flow_status.and_then(|fs| fs.memory_id) {
+                            if let Err(e) = write_to_memory(
+                                db,
+                                &job.workspace_id,
+                                memory_id,
                                 step_id,
-                                e
-                            );
+                                &messages_to_persist,
+                            )
+                            .await
+                            {
+                                tracing::error!(
+                                    "Failed to persist {} messages to memory for step {}: {}",
+                                    messages_to_persist.len(),
+                                    step_id,
+                                    e
+                                );
+                            }
                         }
                     }
                 }