diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index 419a8ad582..ce20220c4f 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -653,9 +653,11 @@ pub async fn set_flow_memory_id( pub async fn process_flow_run_query_params( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, job_id: Uuid, + w_id: &str, + flow_path: &str, run_query: &RunJobQuery, ) -> error::Result<()> { - if let Some(memory_id) = run_query.memory_key() { + if let Some(memory_id) = run_query.memory_key(w_id, flow_path) { set_flow_memory_id(tx, job_id, memory_id).await?; } Ok(()) @@ -669,7 +671,7 @@ pub async fn handle_chat_conversation_messages( run_query: &RunJobQuery, user_message_raw: Option<&Box>, ) -> error::Result<()> { - let memory_id = run_query.memory_key().ok_or_else(|| { + let memory_id = run_query.memory_key(w_id, flow_path).ok_or_else(|| { windmill_common::error::Error::BadRequest( "memory_id is required for chat-enabled flows".to_string(), ) @@ -813,7 +815,7 @@ pub async fn run_flow<'c>( .await?; // Set memory_id if provided (for agent memory) - if let Some(memory_id) = run_query.memory_key() { + if let Some(memory_id) = run_query.memory_key(w_id, flow_path) { set_flow_memory_id(&mut tx, uuid, memory_id).await?; } diff --git a/backend/windmill-api-jobs/src/types.rs b/backend/windmill-api-jobs/src/types.rs index dbde9d32ef..f37040bd11 100644 --- a/backend/windmill-api-jobs/src/types.rs +++ b/backend/windmill-api-jobs/src/types.rs @@ -55,12 +55,15 @@ pub struct RunJobQuery { } impl RunJobQuery { - /// The memory id as stored in `flow_status.memory_id`: a uuid is kept, any other string hashed. - pub fn memory_key(&self) -> Option { + /// The memory id as stored in `flow_status.memory_id`: a uuid is kept, any other string hashed + /// within the workspace and the flow being run. + pub fn memory_key(&self, workspace_id: &str, flow_path: &str) -> Option { self.memory_id .as_deref() .filter(|memory_id| !memory_id.trim().is_empty()) - .map(windmill_common::flow_conversations::memory_key) + .map(|memory_id| { + windmill_common::flow_conversations::memory_key(workspace_id, flow_path, memory_id) + }) } pub async fn get_scheduled_for( diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c48e7b0503..0094f60a18 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -10937,7 +10937,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -10977,7 +10977,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -11017,7 +11017,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -11043,7 +11043,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -11084,7 +11084,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -11123,7 +11123,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -11170,7 +11170,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/SkipPreprocessor" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -14223,7 +14223,7 @@ paths: schema: type: boolean - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -14280,7 +14280,7 @@ paths: schema: type: boolean - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -14771,7 +14771,7 @@ paths: type: boolean - $ref: "#/components/parameters/NewJobId" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string @@ -14802,7 +14802,7 @@ paths: parameters: - $ref: "#/components/parameters/WorkspaceId" - name: memory_id - description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed, so the same string always names the same memory. + description: Memory id for the flow's AI agent steps. A uuid is used as is; any other string is hashed within the workspace and flow, so the same string always names the same memory of that flow. in: query schema: type: string diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 25178ebd80..67b3918804 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -4267,11 +4267,11 @@ async fn execute_component( } } - let is_flow = payload + let flow_path = payload .path - .as_ref() - .map(|p| p.starts_with("flow/")) - .unwrap_or(false); + .as_deref() + .and_then(|path| path.strip_prefix("flow/")) + .map(str::to_string); // Tag for inline-script jobs is read from the deployed policy in run mode; // only preview mode (editor) honors the client-supplied tag. This applies to @@ -4377,8 +4377,9 @@ async fn execute_component( // Apply runnable query parameters if provided if let Some(ref run_query) = payload.run_query_params { - if is_flow { - crate::jobs::process_flow_run_query_params(&mut tx, uuid, run_query).await?; + if let Some(flow_path) = flow_path.as_deref() { + crate::jobs::process_flow_run_query_params(&mut tx, uuid, &w_id, flow_path, run_query) + .await?; } } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 82086f9bac..91338ae867 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -9540,7 +9540,7 @@ async fn run_preview_flow_job( .await?; // Set memory_id if provided (for agent memory) - if let Some(memory_id) = run_query.memory_key() { + if let Some(memory_id) = run_query.memory_key(&w_id, &flow_path) { set_flow_memory_id(&mut tx, uuid, memory_id).await?; } diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index 21a49946c8..93e818436d 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/backend/windmill-common/src/flow_conversations.rs @@ -10,15 +10,20 @@ use crate::utils::truncate_with_ellipsis; /// Changing it detaches every memory stored under a string memory id. const MEMORY_ID_NAMESPACE: Uuid = Uuid::from_u128(0x6f1c2d4e_8a3b_5c7d_9e0f_1a2b3c4d5e6f); -/// Memory is stored and carried in `flow_status.memory_id` as a uuid. Any other string names a -/// memory through a name-based (v5) uuid, so the same string reaches the same memory on every run. -pub fn memory_key(memory_id: &str) -> Uuid { +/// Memory is stored and carried in `flow_status.memory_id` as a uuid, which names the same memory +/// wherever it is passed, as a chat conversation id must. Any other string names a memory through a +/// name-based (v5) uuid scoped to its workspace and flow: a guessable key such as a customer id must +/// not reach another flow's memory, and chat conversation ids are unique across workspaces. +pub fn memory_key(workspace_id: &str, flow_path: &str, memory_id: &str) -> Uuid { let memory_id = memory_id.trim(); Uuid::parse_str(memory_id).unwrap_or_else(|_| { use sha1::{Digest, Sha1}; let mut hasher = Sha1::new(); hasher.update(MEMORY_ID_NAMESPACE.as_bytes()); - hasher.update(memory_id.as_bytes()); + for part in [workspace_id, flow_path, memory_id] { + hasher.update(part.as_bytes()); + hasher.update([0u8]); + } let mut bytes = [0u8; 16]; bytes.copy_from_slice(&hasher.finalize()[..16]); uuid::Builder::from_sha1_bytes(bytes).into_uuid() @@ -161,3 +166,26 @@ pub async fn delete_conversation_memory( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// A string names a memory only within its workspace and flow; a uuid is used as is. + #[test] + fn memory_key_scopes_strings_but_not_uuids() { + let key = memory_key("ws", "f/support/triage", " customer-1 "); + assert_eq!(key, memory_key("ws", "f/support/triage", "customer-1")); + assert_ne!( + key, + memory_key("other_ws", "f/support/triage", "customer-1") + ); + assert_ne!(key, memory_key("ws", "f/sales/triage", "customer-1")); + let conversation = Uuid::from_u128(7).to_string(); + assert_eq!(memory_key("ws", "f/a", &conversation), Uuid::from_u128(7)); + assert_eq!( + memory_key("other_ws", "f/b", &conversation), + Uuid::from_u128(7) + ); + } +} diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index 51a00e65a7..e717f2172e 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -154,6 +154,8 @@ pub async fn get_flow_job_runnable_and_raw_flow( pub struct FlowContext { pub flow_inputs: Option>>, pub flow_status: Option, + /// Path of the flow the run started from, which scopes a string memory id. + pub flow_path: Option, } /// Get flow context (chat settings + args + flow_status) from root flow's job data @@ -171,7 +173,8 @@ pub async fn get_flow_context(db: &DB, job: &MiniPulledJob) -> FlowContext { r#" SELECT j.args as "args: Json>>", - js.flow_status as "flow_status: Json" + js.flow_status as "flow_status: Json", + j.runnable_path FROM v2_job_status js INNER JOIN v2_job j ON j.id = js.id WHERE js.id = $1 @@ -184,6 +187,7 @@ pub async fn get_flow_context(db: &DB, job: &MiniPulledJob) -> FlowContext { Ok(Some(row)) => FlowContext { flow_inputs: row.args.map(|j| j.0), flow_status: row.flow_status.map(|j| j.0), + flow_path: row.runnable_path, }, Ok(None) => { tracing::warn!( diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 88962a7ec3..1fbc157d86 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -142,10 +142,12 @@ fn keep_authored_history_args( /// Reconciles the step's history inputs, the agent's memory policy and the run's memory id, for /// every shape a flow or agent resource may still carry. Also returns a line for the job log when a /// policy that remembers ends up stateless. -fn resolve_history_source( - args: &AIAgentArgs, +fn resolve_history_source<'a>( + args: &'a AIAgentArgs, run_memory_id: Option, -) -> (HistorySource<'_>, Option<&'static str>) { + workspace_id: &str, + flow_path: &str, +) -> (HistorySource<'a>, Option<&'static str>) { if let Some(messages) = &args.messages { return (HistorySource::Messages(messages), None); } @@ -162,7 +164,7 @@ fn resolve_history_source( HistorySource::Stateless, Some("This step's memory id evaluated to an empty value, so the agent runs without memory."), ), - Some(step_memory_id) => memory_key(step_memory_id), + Some(step_memory_id) => memory_key(workspace_id, flow_path, step_memory_id), None => match run_memory_id.or(legacy_memory_id) { Some(memory_id) => memory_id, None => return ( @@ -199,14 +201,16 @@ async fn find_ai_agent_tool_module_in_parent_agent( return Ok(None); }; - let FlowModuleValue::AIAgent { tools, agent, .. } = parent_agent_module.get_value()? else { + let FlowModuleValue::AIAgent { tools, agent, tool_inputs, .. } = + parent_agent_module.get_value()? + else { return Ok(None); }; // A linked parent carries no tools on the module (they live in the resource, resolved only in // the main execution branch). Resolve them from the resource here too, so a nested agent tool // of a saved+linked agent can still be located when it runs as its own job. - let tools = if let Some(agent_ref) = agent.as_deref() { + let mut tools = if let Some(agent_ref) = agent.as_deref() { let agent_path = agent_ref .trim_start_matches("$res:") .trim_start_matches("res://"); @@ -233,6 +237,9 @@ async fn find_ai_agent_tool_module_in_parent_agent( } else { tools }; + // The nested job reads its history inputs from the tool's transforms, which must carry the + // host flow's bindings as the parent evaluated them. + overlay_tool_inputs(&mut tools, &tool_inputs); for tool in tools { if tool.id == tool_module_id { @@ -975,7 +982,12 @@ pub async fn run_agent( .flow_status .as_ref() .and_then(|fs| fs.memory_id); - let (history, history_note) = resolve_history_source(args, conversation_id); + let (history, history_note) = resolve_history_source( + args, + conversation_id, + &job.workspace_id, + flow_context.flow_path.as_deref().unwrap_or_default(), + ); // Check if user_message is provided and non-empty let has_user_message = args @@ -1784,7 +1796,7 @@ mod tests { use serde_json::json; let run = Uuid::from_u128(1); let baked = Uuid::from_u128(2); - let cust_1 = Uuid::parse_str("4e201797-1cea-50b8-bcf8-c11c0f487f67").unwrap(); + let cust_1 = Uuid::parse_str("0168fcea-ffa7-5c15-bdb0-7709bb5f540d").unwrap(); let window = json!({ "kind": "window", "context_length": 10 }); let message = json!([{ "role": "user", "content": "earlier" }]); let cases = [ @@ -1873,7 +1885,7 @@ mod tests { .unwrap() .extend(history.as_object().unwrap().clone()); let args: AIAgentArgs = serde_json::from_value(raw).unwrap(); - let resolved = match resolve_history_source(&args, run_memory_id) { + let resolved = match resolve_history_source(&args, run_memory_id, "ws", "f/flow") { (HistorySource::Messages(m), _) => Resolved::Messages(m.len()), (HistorySource::Window { memory_id, context_length }, _) => { Resolved::Window(memory_id, context_length) diff --git a/docs/reusable-ai-agents.md b/docs/reusable-ai-agents.md index b92205cf3f..5e8c360b50 100644 --- a/docs/reusable-ai-agents.md +++ b/docs/reusable-ai-agents.md @@ -44,11 +44,12 @@ which memory it is: An absent `memory` means off. `auto` and `manual` are the older spellings and are still read. - **Run: memory id.** `flow_status.memory_id`, set when the run is queued: the chat conversation id, an app chat session id, or the `memory_id` run parameter. Any string is accepted, and one - that is not a uuid is hashed to a v5 uuid (`memory_key` in - `windmill-common/src/flow_conversations.rs`). Nothing is generated at save time, so schedules, + that is not a uuid is hashed to a v5 uuid scoped to the workspace and the flow the run started + from (`memory_key` in `windmill-common/src/flow_conversations.rs`), so a guessable key never + reaches another flow's memory. A uuid is used as is. Nothing is generated at save time, so schedules, webhooks, evals and plain runs pass no id and run stateless. - **Step: history inputs.** Flow-local, so they stay on a linked step. `memory_id` overrides the - run's id: a fixed value is one memory shared by every run, an expression such as + run's id, hashed the same way: a fixed value is one memory shared by every run, an expression such as `flow_input.customer_id` one memory per key, and an expression that evaluates to nothing runs stateless rather than falling back to the run's id. `messages` supplies the history itself and bypasses memory. The editor writes at most one of them and never seeds a placeholder for diff --git a/frontend/src/lib/components/flows/content/AgentResourceBar.svelte b/frontend/src/lib/components/flows/content/AgentResourceBar.svelte index b7d4f02234..8507bf41a6 100644 --- a/frontend/src/lib/components/flows/content/AgentResourceBar.svelte +++ b/frontend/src/lib/components/flows/content/AgentResourceBar.svelte @@ -50,6 +50,7 @@ opWorkspace = undefined, flowPath = '', fromAgentEditor = false, + chatInputEnabled = false, linkedMemory = $bindable() }: { agent: string | undefined @@ -68,6 +69,7 @@ // backend supports it, but only a flow can author it, and a second editor over a second draft // is the wrong way in. fromAgentEditor?: boolean + chatInputEnabled?: boolean // The linked agent's memory once its config has loaded, for the step's history row. linkedMemory?: { memory: unknown } | undefined } = $props() @@ -291,6 +293,20 @@ // saved without a complete one fails on every linked run. Block saving when the provider is // computed/connected (only a static value can be captured into the resource) or when the static // value is incomplete (a fresh step defaults to empty resource/model, which is still static). + // A saved agent never carries a memory id, so saving would drop the id this step's runs still fall + // back to and leave them without memory. The author picks what replaces it first. In chat mode the + // conversation id always won, so there the id was never read. + let legacyMemorySaveError = $derived.by(() => { + const memory = inputTransforms?.memory as + | { type?: string; value?: { kind?: string; context_length?: number; memory_id?: string } } + | undefined + const value = memory?.type === 'static' ? memory.value : undefined + if (chatInputEnabled || value?.kind !== 'auto' || !value.memory_id || !value.context_length) { + return undefined + } + return "This step still uses a fixed memory id from an earlier version. In Memory, choose Keep as override or Use the run's memory id, then save it as an agent." + }) + let providerSaveError = $derived.by(() => { const t = inputTransforms?.provider as | { type?: string; value?: { resource?: string; model?: string } } @@ -322,8 +338,8 @@ // the success toast that would otherwise bury the explanation. async function persist(path: string, description?: string): Promise { const dropped = nonStaticBrainKeys(inputTransforms) - if (providerSaveError) { - throw new Error(providerSaveError) + if (providerSaveError ?? legacyMemorySaveError) { + throw new Error(providerSaveError ?? legacyMemorySaveError) } if (dropped.length > 0) { sendUserToast( @@ -640,9 +656,9 @@ size="sm" /> - {#if providerSaveError} + {#if providerSaveError ?? legacyMemorySaveError}

- {providerSaveError} + {providerSaveError ?? legacyMemorySaveError}

{/if} @@ -650,7 +666,10 @@