diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 3d7fce4d07..588b1938bd 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -2059,6 +2059,7 @@ pub async fn run_agent( *memory_id, step_id, &messages_to_persist, + true, ) .await { diff --git a/backend/windmill-worker/src/memory_common.rs b/backend/windmill-worker/src/memory_common.rs index e024acea69..52aaa4fafa 100644 --- a/backend/windmill-worker/src/memory_common.rs +++ b/backend/windmill-worker/src/memory_common.rs @@ -30,16 +30,21 @@ pub async fn read_from_db( } } -/// Write AI agent memory to database with size checking and truncation +/// Write AI agent memory to database with size checking and truncation. +/// +/// Returns how many of the oldest messages were dropped to fit, so the step can say so +/// on the run: from the flow's side a truncation is invisible, the agent simply having +/// forgotten the start of its conversation by the next one. pub async fn write_to_db( db: &DB, workspace_id: &str, conversation_id: Uuid, step_id: &str, messages: &[OpenAIMessage], -) -> Result<(), Error> { + allow_truncation: bool, +) -> Result { if messages.is_empty() { - return Ok(()); + return Ok(0); } // Serialize messages and check size @@ -49,6 +54,13 @@ pub async fn write_to_db( // Truncate if necessary if size_bytes > MAX_MEMORY_SIZE_BYTES { + // Compaction owns its exchange boundaries, including after an object-store failure. + if !allow_truncation { + return Err(Error::ExecutionErr(format!( + "Memory exceeds database capacity ({} > {} bytes); existing memory was not updated", + size_bytes, MAX_MEMORY_SIZE_BYTES + ))); + } tracing::warn!( "Memory size ({} bytes) exceeds limit ({} bytes) for workspace={} conversation={} step={}. Truncating messages. Use S3 storage in workspace settings to store full conversation history.", size_bytes, @@ -78,7 +90,7 @@ pub async fn write_to_db( .execute(db) .await?; - Ok(()) + Ok(messages.len() - messages_to_store.len()) } /// Delete all memory for a conversation from database diff --git a/backend/windmill-worker/src/memory_oss.rs b/backend/windmill-worker/src/memory_oss.rs index dcdab14e2a..64d6c2a0e5 100644 --- a/backend/windmill-worker/src/memory_oss.rs +++ b/backend/windmill-worker/src/memory_oss.rs @@ -30,14 +30,29 @@ pub async fn write_to_memory( conversation_id: Uuid, step_id: &str, messages: &[OpenAIMessage], -) -> anyhow::Result<()> { + allow_truncation: bool, +) -> anyhow::Result { if messages.is_empty() { - return Ok(()); + return Ok(0); } - memory_common::write_to_db(db, workspace_id, conversation_id, step_id, messages) - .await - .map_err(|e| anyhow::anyhow!("Database write failed: {e:?}")) + memory_common::write_to_db( + db, + workspace_id, + conversation_id, + step_id, + messages, + allow_truncation, + ) + .await + .map_err(|e| anyhow::anyhow!("Database write failed: {e:?}")) +} + +/// How many bytes a persisted memory can hold, `None` where nothing bounds it. +/// In OSS: always the database limit +#[cfg(not(all(feature = "private", feature = "enterprise")))] +pub async fn memory_storage_capacity_bytes() -> Option { + Some(memory_common::MAX_MEMORY_SIZE_BYTES) } /// Delete all memory for a conversation from storage