diff --git a/crates/agentic-server-core/migrations/0007_vector_store_lifecycle.sql b/crates/agentic-server-core/migrations/0007_vector_store_lifecycle.sql new file mode 100644 index 00000000..ed438b6f --- /dev/null +++ b/crates/agentic-server-core/migrations/0007_vector_store_lifecycle.sql @@ -0,0 +1,13 @@ +-- Lifecycle columns allow visibility and filtered pagination before row limits. +-- Legacy stores remain permanent; their creation time seeds first activity. +ALTER TABLE file_search_stores ADD COLUMN last_active_at BIGINT; +ALTER TABLE file_search_stores ADD COLUMN expires_after_days BIGINT; +ALTER TABLE file_search_stores ADD COLUMN expires_at BIGINT; +ALTER TABLE file_search_stores ADD COLUMN lifecycle_status TEXT NOT NULL DEFAULT 'completed'; +UPDATE file_search_stores SET last_active_at = created_at; +CREATE INDEX file_search_store_expiration ON file_search_stores(lifecycle_status, expires_at, id); + +-- Old successful attachments can recover parsed text from their original upload. +ALTER TABLE file_search_attachments ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'; +ALTER TABLE file_search_attachments ADD COLUMN parsed_content TEXT; +CREATE INDEX file_search_attachment_status ON file_search_attachments(store_id, status, created_at, file_id); diff --git a/crates/agentic-server-core/src/storage/file_search.rs b/crates/agentic-server-core/src/storage/file_search.rs index ccfa23ff..85ba5457 100644 --- a/crates/agentic-server-core/src/storage/file_search.rs +++ b/crates/agentic-server-core/src/storage/file_search.rs @@ -7,6 +7,8 @@ use sqlx::FromRow; use tokio_util::sync::CancellationToken; use super::{DbPool, DbTransaction}; +#[path = "vector_store_lifecycle.rs"] +mod lifecycle; use crate::types::file_search::{ FileObject, FileSearchError, ListOrder, ListParams, VectorStoreFileObject, VectorStoreObject, }; @@ -36,7 +38,6 @@ pub(crate) enum FilePublicationFailure { #[derive(FromRow)] pub(crate) struct StoredVectorStore { - pub data: String, pub embedding_identity: String, pub embedding_dimensions: i64, } @@ -57,6 +58,7 @@ pub(crate) struct PreparedAttachment { pub object: VectorStoreFileObject, pub chunks: Vec, pub dimensions: i64, + pub parsed_content: String, } #[derive(Clone, Copy)] @@ -109,9 +111,13 @@ impl FileSearchStorage { } async fn file_visibility(&self) -> Result { - let mut connection = self.pool.acquire().await?; - let now = database_now(&mut connection).await?; - Ok(format!("(expires_at IS NULL OR expires_at > {now})")) + let connection = self.pool.acquire().await?; + let clock = if connection.backend_name() == "PostgreSQL" { + "EXTRACT(EPOCH FROM clock_timestamp())" + } else { + "CAST(strftime('%s', 'now') AS BIGINT)" + }; + Ok(format!("(expires_at IS NULL OR expires_at > {clock})")) } pub(crate) async fn delete_file(&self, id: &str) -> Result<(), FileSearchError> { @@ -197,7 +203,8 @@ impl FileSearchStorage { .collect::>() .join(", "); let sql = format!( - "SELECT DISTINCT file_id FROM file_search_attachments WHERE store_id IN ({store_placeholders}) AND file_id IN ({file_placeholders}) AND file_id IN (SELECT id FROM file_search_files WHERE {})", + "SELECT DISTINCT file_id FROM file_search_attachments WHERE status = 'completed' AND store_id IN ({store_placeholders}) AND file_id IN ({file_placeholders}) AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND {}) AND file_id IN (SELECT id FROM file_search_files WHERE {})", + self.file_visibility().await?, self.file_visibility().await? ); let mut query = sqlx::query_scalar::<_, String>(&sql); @@ -217,7 +224,7 @@ impl FileSearchStorage { .join(", "); let visibility = self.file_visibility().await?; let sql = format!( - "SELECT chunk_index FROM file_search_chunks WHERE store_id IN ({placeholders}) AND file_id IN (SELECT id FROM file_search_files WHERE {visibility}) LIMIT 1" + "SELECT chunk_index FROM file_search_chunks WHERE (store_id, file_id) IN (SELECT store_id, file_id FROM file_search_attachments WHERE status = 'completed') AND store_id IN ({placeholders}) AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND {visibility}) AND file_id IN (SELECT id FROM file_search_files WHERE {visibility}) LIMIT 1" ); let mut query = sqlx::query_scalar::<_, i64>(&sql); for store in stores { @@ -305,22 +312,9 @@ impl FileSearchStorage { } pub(crate) async fn store(&self, id: &str) -> Result { - sqlx::query_as("SELECT data, embedding_identity, embedding_dimensions FROM file_search_stores WHERE id = $1") - .bind(id) - .fetch_optional(self.pool.as_ref()) - .await? - .ok_or_else(|| FileSearchError::NotFound("Vector store not found".into())) - } - - pub(crate) async fn store_object(&self, id: &str) -> Result { - let row = self.store(id).await?; - let mut store: VectorStoreObject = serde_json::from_str(&row.data)?; - let (count, bytes): (i64, i64) = sqlx::query_as(&format!("SELECT COUNT(*), CAST(COALESCE(SUM(usage_bytes), 0) AS BIGINT) FROM file_search_attachments WHERE store_id = $1 AND file_id IN (SELECT id FROM file_search_files WHERE {})", self.file_visibility().await?)) - .bind(id).fetch_one(self.pool.as_ref()).await?; - store.file_counts.completed = count; - store.file_counts.total = count; - store.usage_bytes = bytes; - Ok(store) + sqlx::query_as(&format!("SELECT embedding_identity, embedding_dimensions FROM file_search_stores WHERE id = $1 AND lifecycle_status != 'expired' AND {}", self.file_visibility().await?)) + .bind(id).fetch_optional(self.pool.as_ref()).await? + .ok_or_else(|| FileSearchError::NotFound("Vector store not found or expired".into())) } pub(crate) async fn create_store( @@ -339,8 +333,10 @@ impl FileSearchStorage { encoded.push(chunks); } let mut tx = self.pool.begin().await?; - sqlx::query("INSERT INTO file_search_stores (id, created_at, data, embedding_identity, embedding_dimensions) VALUES ($1, $2, $3, $4, 0)") - .bind(&object.id).bind(object.created_at).bind(serde_json::to_string(object)?).bind(identity) + let now = database_now(&mut tx).await?; + let days = object.expires_after.as_ref().map(|policy| i64::from(policy.days)); + sqlx::query("INSERT INTO file_search_stores (id, created_at, data, embedding_identity, embedding_dimensions, last_active_at, expires_after_days, expires_at) VALUES ($1, $2, $3, $4, 0, $5, $6, $7)") + .bind(&object.id).bind(object.created_at).bind(serde_json::to_string(object)?).bind(identity).bind(now).bind(days).bind(days.map(|days| now + days * 86400)) .execute(&mut *tx).await?; for (attachment, (chunks, storage_bytes)) in attachments.iter().zip(encoded) { publish_attachment(&mut tx, &object.id, identity, attachment, &chunks, storage_bytes).await?; @@ -369,7 +365,7 @@ impl FileSearchStorage { file_id: &str, ) -> Result, FileSearchError> { let data: Option = - sqlx::query_scalar(&format!("SELECT data FROM file_search_attachments WHERE store_id = $1 AND file_id = $2 AND file_id IN (SELECT id FROM file_search_files WHERE {})", self.file_visibility().await?)) + sqlx::query_scalar(&format!("SELECT data FROM file_search_attachments WHERE store_id = $1 AND file_id = $2 AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND {}) AND file_id IN (SELECT id FROM file_search_files WHERE {})", self.file_visibility().await?, self.file_visibility().await?)) .bind(store_id) .bind(file_id) .fetch_optional(self.pool.as_ref()) @@ -387,6 +383,25 @@ impl FileSearchStorage { store_id: Option<&str>, ) -> Result<(), FileSearchError> { let mut tx = self.pool.begin().await?; + if let Some(store_id) = store_id { + sqlx::query("UPDATE file_search_files SET id = id WHERE id = $1") + .bind(id) + .execute(&mut *tx) + .await?; + lifecycle::lock_store(&mut tx, store_id).await?; + let now = database_now(&mut tx).await?; + lifecycle::require_live_store(&mut tx, store_id, now).await?; + let live: Option = sqlx::query_scalar( + "SELECT id FROM file_search_files WHERE id = $1 AND (expires_at IS NULL OR expires_at > $2)", + ) + .bind(id) + .bind(now) + .fetch_optional(&mut *tx) + .await?; + if live.is_none() { + return Err(FileSearchError::NotFound("File not found or expired".into())); + } + } let filter = if store_id.is_some() { " AND store_id = $2" } else { "" }; let sql = format!( "DELETE FROM {} WHERE {} = $1{filter}", @@ -448,7 +463,9 @@ impl FileSearchStorage { } ), Collection::Attachments => { - format!("{filter} AND file_id IN (SELECT id FROM file_search_files WHERE {visibility})") + format!( + "{filter} AND ($5 = '' OR status = $5) AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND {visibility}) AND file_id IN (SELECT id FROM file_search_files WHERE {visibility})" + ) } Collection::Stores => filter.to_owned(), }; @@ -463,6 +480,13 @@ impl FileSearchStorage { if let Some(store_id) = store_id { query = query.bind(store_id); } + if matches!(collection, Collection::Attachments) { + query = query.bind( + params + .filter + .map_or("", crate::types::file_search::AttachmentStatus::as_str), + ); + } if matches!(collection, Collection::Files) { query = query.bind(params.purpose.as_deref().unwrap_or("")); } @@ -481,7 +505,7 @@ impl FileSearchStorage { .join(", "); let visibility = self.file_visibility().await?; let sql = format!( - "SELECT data FROM file_search_chunks WHERE store_id IN ({placeholders}) AND file_id IN (SELECT id FROM file_search_files WHERE {visibility}) ORDER BY store_id, file_id, chunk_index" + "SELECT data FROM file_search_chunks WHERE (store_id, file_id) IN (SELECT store_id, file_id FROM file_search_attachments WHERE status = 'completed') AND store_id IN ({placeholders}) AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND {visibility}) AND file_id IN (SELECT id FROM file_search_files WHERE {visibility}) ORDER BY store_id, file_id, chunk_index" ); let mut query = sqlx::query_scalar::<_, String>(&sql); for id in store_ids { @@ -590,6 +614,7 @@ async fn publish_attachment( if live.is_none() { return Err(FileSearchError::NotFound("File expired during ingestion".into())); } + lifecycle::require_live_store(tx, store_id, now).await?; let bytes: i64 = sqlx::query_scalar( "SELECT CAST(COALESCE(SUM(storage_bytes), 0) AS BIGINT) FROM file_search_attachments WHERE store_id = $1", ) @@ -605,9 +630,9 @@ async fn publish_attachment( count.saturating_add(i64::try_from(chunks.len()).unwrap_or(i64::MAX)), )?; let object = &attachment.object; - let inserted = sqlx::query("INSERT INTO file_search_attachments (store_id, file_id, created_at, usage_bytes, data, storage_bytes) VALUES ($1, $2, $3, $4, $5, $6)") + let inserted = sqlx::query("INSERT INTO file_search_attachments (store_id, file_id, created_at, usage_bytes, data, storage_bytes, status, parsed_content) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)") .bind(store_id).bind(&object.id).bind(object.created_at).bind(object.usage_bytes).bind(serde_json::to_string(object)?) - .bind(storage_bytes) + .bind(storage_bytes).bind(object.status.as_str()).bind(&attachment.parsed_content) .execute(&mut **tx).await; if let Err(sqlx::Error::Database(error)) = &inserted { if error.is_unique_violation() { @@ -620,6 +645,7 @@ async fn publish_attachment( } } inserted?; + lifecycle::touch_store(tx, store_id, now).await?; for (chunk, data) in attachment.chunks.iter().zip(chunks) { let index = i64::try_from(chunk.chunk_index).map_err(|_| FileSearchError::InvalidRequest("Too many chunks".into()))?; @@ -640,7 +666,9 @@ mod tests { use crate::{ storage::create_pool_with_schema, tool::file_search::FileSearchService, - types::file_search::{ChunkingStrategy, CreateVectorStoreRequest, FileAttributes, FileSearchConfig}, + types::file_search::{ + CreateVectorStoreRequest, FileAttributes, FileSearchConfig, VectorStoreFileChunkingStrategy, + }, }; async fn prepared( @@ -654,15 +682,16 @@ mod tests { .await .unwrap(); PreparedAttachment { + parsed_content: "capacity".into(), object: VectorStoreFileObject { id: file.id.clone(), object: "vector_store.file".into(), created_at: 0, vector_store_id: store_id.into(), - status: "completed".into(), + status: crate::types::file_search::AttachmentStatus::Completed, usage_bytes: 0, attributes: FileAttributes::default(), - chunking_strategy: ChunkingStrategy::Auto, + chunking_strategy: VectorStoreFileChunkingStrategy::Other, last_error: None, }, dimensions: i64::try_from(dimensions).unwrap(), diff --git a/crates/agentic-server-core/src/storage/pgvector.rs b/crates/agentic-server-core/src/storage/pgvector.rs index 36bfa45f..2e5868f1 100644 --- a/crates/agentic-server-core/src/storage/pgvector.rs +++ b/crates/agentic-server-core/src/storage/pgvector.rs @@ -187,14 +187,16 @@ impl PgvectorStorage { if (semantic && mode == SearchMode::Keyword) || (!semantic && mode == SearchMode::Semantic) { continue; } - let mut sql = CandidateSql::new("SELECT data FROM file_search_chunks WHERE store_id IN ("); + let mut sql = CandidateSql::new( + "SELECT data FROM file_search_chunks WHERE (store_id, file_id) IN (SELECT store_id, file_id FROM file_search_attachments WHERE status = 'completed') AND store_id IN (", + ); for (i, store) in stores.iter().enumerate() { if i > 0 { sql.push(", "); } sql.push_bind(store); } - sql.push(") AND file_id IN (SELECT id FROM file_search_files WHERE expires_at IS NULL OR expires_at > EXTRACT(EPOCH FROM clock_timestamp()))"); + sql.push(") AND store_id IN (SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND (expires_at IS NULL OR expires_at > EXTRACT(EPOCH FROM clock_timestamp()))) AND file_id IN (SELECT id FROM file_search_files WHERE expires_at IS NULL OR expires_at > EXTRACT(EPOCH FROM clock_timestamp()))"); if let Some(filter) = filter { sql.push(" AND "); push_filter(&mut sql, filter)?; diff --git a/crates/agentic-server-core/src/storage/schema.rs b/crates/agentic-server-core/src/storage/schema.rs index 5c5821b0..bc943360 100644 --- a/crates/agentic-server-core/src/storage/schema.rs +++ b/crates/agentic-server-core/src/storage/schema.rs @@ -15,7 +15,7 @@ use crate::config::DEFAULT_POSTGRES_MIGRATION_TIMEOUT_SECONDS; type DbResult = Result; const POSTGRES_SCHEMA_ADVISORY_LOCK: i64 = 7_194_963_546_799_751; -const REQUIRED_POSTGRES_SCHEMA_COLUMN_COUNT: i64 = 43; +const REQUIRED_POSTGRES_SCHEMA_COLUMN_COUNT: i64 = 49; const REQUIRED_POSTGRES_CONSTRAINT_COUNT: i64 = 15; const REQUIRED_POSTGRES_INTEGER_COLUMN_COUNT: i64 = 4; const POSTGRES_INTEGER_WIDENING_SQL: &str = " @@ -99,12 +99,18 @@ where ('file_search_stores', 'data', 'text', 'NO'), \ ('file_search_stores', 'embedding_identity', 'text', 'NO'), \ ('file_search_stores', 'embedding_dimensions', 'bigint', 'NO'), \ + ('file_search_stores', 'last_active_at', 'bigint', 'YES'), \ + ('file_search_stores', 'expires_after_days', 'bigint', 'YES'), \ + ('file_search_stores', 'expires_at', 'bigint', 'YES'), \ + ('file_search_stores', 'lifecycle_status', 'text', 'NO'), \ ('file_search_attachments', 'store_id', 'text', 'NO'), \ ('file_search_attachments', 'file_id', 'text', 'NO'), \ ('file_search_attachments', 'created_at', 'bigint', 'NO'), \ ('file_search_attachments', 'usage_bytes', 'bigint', 'NO'), \ ('file_search_attachments', 'storage_bytes', 'bigint', 'NO'), \ ('file_search_attachments', 'data', 'text', 'NO'), \ + ('file_search_attachments', 'status', 'text', 'NO'), \ + ('file_search_attachments', 'parsed_content', 'text', 'YES'), \ ('file_search_chunks', 'store_id', 'text', 'NO'), \ ('file_search_chunks', 'file_id', 'text', 'NO'), \ ('file_search_chunks', 'chunk_index', 'bigint', 'NO'), \ @@ -416,10 +422,12 @@ pub(crate) async fn verify_persistence_ready(pool: &DbPool) -> DbResult<()> { ('file_search_attachments', 'INSERT'), \ ('file_search_attachments', 'DELETE'), \ ('file_search_chunks', 'SELECT'), \ + ('file_search_attachments', 'UPDATE'), \ + ('file_search_chunks', 'UPDATE'), \ ('file_search_chunks', 'INSERT') \ ) \ SELECT current_setting('transaction_read_only') = 'off' \ - AND COUNT(table_relation.oid) = 23 \ + AND COUNT(table_relation.oid) = 25 \ AND COALESCE(BOOL_AND( \ has_table_privilege(current_user, table_relation.oid, required.privilege) \ ), false) \ @@ -450,8 +458,8 @@ pub(crate) async fn verify_persistence_ready(pool: &DbPool) -> DbResult<()> { "SELECT id FROM responses LIMIT 0", "SELECT id, created_at, data, content_type, content_base64, expires_at, purpose FROM file_search_files LIMIT 0", "SELECT file_id FROM file_search_blob_cleanup LIMIT 0", - "SELECT id, created_at, data, embedding_identity, embedding_dimensions FROM file_search_stores LIMIT 0", - "SELECT store_id, file_id, created_at, usage_bytes, storage_bytes, data FROM file_search_attachments LIMIT 0", + "SELECT id, created_at, data, embedding_identity, embedding_dimensions, last_active_at, expires_after_days, expires_at, lifecycle_status FROM file_search_stores LIMIT 0", + "SELECT store_id, file_id, created_at, usage_bytes, storage_bytes, data, status, parsed_content FROM file_search_attachments LIMIT 0", "SELECT store_id, file_id, chunk_index, data FROM file_search_chunks LIMIT 0", ] { sqlx::query(statement).execute(&mut *connection).await?; @@ -714,6 +722,15 @@ mod tests { .execute(pool.as_ref()) .await .unwrap(); + assert!( + verify_persistence_ready(pool.as_ref()).await.is_err(), + "store lifecycle migration is required" + ); + assert!(wrapper.ensure_schema_ready_with_marker(true).await.is_err()); + sqlx::raw_sql(include_str!("../../migrations/0007_vector_store_lifecycle.sql")) + .execute(pool.as_ref()) + .await + .unwrap(); verify_persistence_ready(pool.as_ref()).await.unwrap(); wrapper.ensure_schema_ready_with_marker(true).await.unwrap(); } @@ -824,6 +841,14 @@ mod tests { .execute(&mut *connection) .await .unwrap(); + assert!( + supervisor.ensure_schema_ready_with_marker(true).await.is_err(), + "store lifecycle migration is required" + ); + sqlx::raw_sql(include_str!("../../migrations/0007_vector_store_lifecycle.sql")) + .execute(&mut *connection) + .await + .unwrap(); supervisor.ensure_schema_ready_with_marker(true).await.unwrap(); supervisor.pool.close().await; sqlx::query("SET search_path TO public") @@ -866,6 +891,7 @@ mod tests { include_str!("../../migrations/0004_link_conversation_latest_response.sql"), include_str!("../../migrations/0005_file_search.sql"), include_str!("../../migrations/0006_file_expiration.sql"), + include_str!("../../migrations/0007_vector_store_lifecycle.sql"), ] { sqlx::raw_sql(migration) .execute(&mut *connection) diff --git a/crates/agentic-server-core/src/storage/vector_store_lifecycle.rs b/crates/agentic-server-core/src/storage/vector_store_lifecycle.rs new file mode 100644 index 00000000..c48b9828 --- /dev/null +++ b/crates/agentic-server-core/src/storage/vector_store_lifecycle.rs @@ -0,0 +1,278 @@ +//! Store lifecycle transactions share the publication store guard. +use super::{DbTransaction, FileSearchStorage, StoredChunk, database_now}; +use crate::types::file_search::{ + AttachmentStatus, FileAttributes, FileCounts, FileSearchError, UpdateVectorStoreRequest, + VectorStoreExpirationAnchor, VectorStoreExpiresAfter, VectorStoreFileObject, VectorStoreObject, VectorStoreStatus, +}; + +#[derive(sqlx::FromRow)] +struct StoreLifecycle { + data: String, + last_active_at: Option, + expires_after_days: Option, + expires_at: Option, + lifecycle_status: String, +} + +impl StoreLifecycle { + fn expired(&self, now: i64) -> bool { + self.lifecycle_status == "expired" || self.expires_at.is_some_and(|deadline| deadline <= now) + } + + fn object(&self, now: i64) -> Result { + let mut object: VectorStoreObject = serde_json::from_str(&self.data)?; + object.last_active_at = self.last_active_at; + object.expires_at = self.expires_at; + object.expires_after = self + .expires_after_days + .map(|days| { + u16::try_from(days) + .map(|days| VectorStoreExpiresAfter { + anchor: VectorStoreExpirationAnchor::LastActiveAt, + days, + }) + .map_err(|_| FileSearchError::Unavailable("Stored expiration policy is invalid".into())) + }) + .transpose()?; + if self.expired(now) { + object.status = VectorStoreStatus::Expired; + } + Ok(object) + } +} + +pub(super) async fn lock_store(tx: &mut DbTransaction<'_>, id: &str) -> Result<(), FileSearchError> { + if sqlx::query("UPDATE file_search_stores SET id = id WHERE id = $1") + .bind(id) + .execute(&mut **tx) + .await? + .rows_affected() + != 1 + { + return Err(FileSearchError::NotFound("Vector store not found".into())); + } + Ok(()) +} + +pub(super) async fn require_live_store(tx: &mut DbTransaction<'_>, id: &str, now: i64) -> Result<(), FileSearchError> { + let live: Option = sqlx::query_scalar("SELECT id FROM file_search_stores WHERE id = $1 AND lifecycle_status != 'expired' AND (expires_at IS NULL OR expires_at > $2)") + .bind(id).bind(now).fetch_optional(&mut **tx).await?; + if live.is_none() { + return Err(FileSearchError::NotFound("Vector store not found or expired".into())); + } + Ok(()) +} + +pub(super) async fn touch_store(tx: &mut DbTransaction<'_>, id: &str, now: i64) -> Result<(), FileSearchError> { + sqlx::query( + "UPDATE file_search_stores SET last_active_at = $2, expires_at = $2 + expires_after_days * 86400 WHERE id = $1", + ) + .bind(id) + .bind(now) + .execute(&mut **tx) + .await?; + Ok(()) +} + +impl FileSearchStorage { + pub(crate) async fn store_object(&self, id: &str) -> Result { + let mut connection = self.pool.acquire().await?; + let row: StoreLifecycle = sqlx::query_as("SELECT data, last_active_at, expires_after_days, expires_at, lifecycle_status FROM file_search_stores WHERE id = $1") + .bind(id).fetch_optional(&mut *connection).await?.ok_or_else(|| FileSearchError::NotFound("Vector store not found".into()))?; + let now = database_now(&mut connection).await?; + let mut object = row.object(now)?; + object.file_counts = FileCounts::default(); + object.usage_bytes = 0; + if object.status != VectorStoreStatus::Expired { + let counts: Vec<(String, i64, i64)> = sqlx::query_as("SELECT status, COUNT(*), CAST(COALESCE(SUM(usage_bytes), 0) AS BIGINT) FROM file_search_attachments WHERE store_id = $1 AND file_id IN (SELECT id FROM file_search_files WHERE expires_at IS NULL OR expires_at > $2) GROUP BY status") + .bind(id).bind(now).fetch_all(&mut *connection).await?; + for (status, count, bytes) in counts { + match status.as_str() { + "in_progress" => object.file_counts.in_progress = count, + "completed" => object.file_counts.completed = count, + "failed" => object.file_counts.failed = count, + "cancelled" => object.file_counts.cancelled = count, + _ => { + return Err(FileSearchError::Unavailable( + "Stored attachment status is invalid".into(), + )); + } + } + object.file_counts.total += count; + object.usage_bytes += bytes; + } + object.status = if object.file_counts.in_progress > 0 { + VectorStoreStatus::InProgress + } else { + VectorStoreStatus::Completed + }; + } + Ok(object) + } + + pub(crate) async fn update_store( + &self, + id: &str, + request: UpdateVectorStoreRequest, + ) -> Result<(), FileSearchError> { + let mut tx = self.pool.begin().await?; + lock_store(&mut tx, id).await?; + let now = database_now(&mut tx).await?; + require_live_store(&mut tx, id, now).await?; + let row: StoreLifecycle = sqlx::query_as("SELECT data, last_active_at, expires_after_days, expires_at, lifecycle_status FROM file_search_stores WHERE id = $1") + .bind(id).fetch_one(&mut *tx).await?; + let mut object = row.object(now)?; + if let Some(name) = request.name.0 { + object.name = name.unwrap_or_default(); + } + if let Some(metadata) = request.metadata.0 { + object.metadata = metadata; + } + if let Some(policy) = request.expires_after.0 { + object.expires_after = policy; + object.expires_at = object.expires_after.as_ref().map(|policy| { + object + .last_active_at + .unwrap_or(object.created_at) + .saturating_add(i64::from(policy.days) * 86400) + }); + } + sqlx::query("UPDATE file_search_stores SET data = $2, expires_after_days = $3, expires_at = $4 WHERE id = $1") + .bind(id) + .bind(serde_json::to_string(&object)?) + .bind(object.expires_after.as_ref().map(|policy| i64::from(policy.days))) + .bind(object.expires_at) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + pub(crate) async fn refresh_activity(&self, ids: &[String]) -> Result<(), FileSearchError> { + let mut ids = ids.iter().collect::>(); + ids.sort_unstable(); + ids.dedup(); + let mut tx = self.pool.begin().await?; + for id in &ids { + lock_store(&mut tx, id).await?; + } + let now = database_now(&mut tx).await?; + for id in ids { + require_live_store(&mut tx, id, now).await?; + touch_store(&mut tx, id, now).await?; + } + tx.commit().await?; + Ok(()) + } + + pub(crate) async fn expire_stores(&self, limit: usize) -> Result { + // Scan outside each write transaction so SQLite never upgrades a stale read snapshot. + let visibility = self.file_visibility().await?; + let ids: Vec = sqlx::query_scalar(&format!("SELECT id FROM file_search_stores WHERE lifecycle_status != 'expired' AND NOT {visibility} ORDER BY expires_at, id LIMIT $1")) + .bind(i64::try_from(limit).unwrap_or(1000)).fetch_all(self.pool.as_ref()).await?; + let mut expired = 0; + for id in ids { + let mut tx = self.pool.begin().await?; + match lock_store(&mut tx, &id).await { + Ok(()) => {} + Err(FileSearchError::NotFound(_)) => continue, + Err(error) => return Err(error), + } + let now = database_now(&mut tx).await?; + let changed = sqlx::query("UPDATE file_search_stores SET lifecycle_status = 'expired' WHERE id = $1 AND lifecycle_status != 'expired' AND expires_at <= $2") + .bind(&id).bind(now).execute(&mut *tx).await?.rows_affected(); + if changed == 1 { + sqlx::query("DELETE FROM file_search_attachments WHERE store_id = $1") + .bind(&id) + .execute(&mut *tx) + .await?; + expired += 1; + } + tx.commit().await?; + } + Ok(expired) + } + + pub(crate) async fn update_attachment( + &self, + store_id: &str, + file_id: &str, + attributes: FileAttributes, + ) -> Result { + let mut tx = self.pool.begin().await?; + sqlx::query("UPDATE file_search_files SET id = id WHERE id = $1") + .bind(file_id) + .execute(&mut *tx) + .await?; + lock_store(&mut tx, store_id).await?; + let now = database_now(&mut tx).await?; + require_live_store(&mut tx, store_id, now).await?; + let data: String = sqlx::query_scalar("SELECT data FROM file_search_attachments WHERE store_id = $1 AND file_id = $2 AND file_id IN (SELECT id FROM file_search_files WHERE expires_at IS NULL OR expires_at > $3)") + .bind(store_id).bind(file_id).bind(now).fetch_optional(&mut *tx).await?.ok_or_else(|| FileSearchError::NotFound("Vector store file not found".into()))?; + let mut object: VectorStoreFileObject = serde_json::from_str(&data)?; + object.attributes = attributes; + sqlx::query("UPDATE file_search_attachments SET data = $3 WHERE store_id = $1 AND file_id = $2") + .bind(store_id) + .bind(file_id) + .bind(serde_json::to_string(&object)?) + .execute(&mut *tx) + .await?; + // Bounded by the store's existing serialized corpus budget; preserve every other chunk field. + let rows: Vec<(i64, String)> = + sqlx::query_as("SELECT chunk_index, data FROM file_search_chunks WHERE store_id = $1 AND file_id = $2") + .bind(store_id) + .bind(file_id) + .fetch_all(&mut *tx) + .await?; + let mut bytes = 0i64; + let mut chunks = Vec::with_capacity(rows.len()); + for (index, data) in rows { + let mut chunk: StoredChunk = serde_json::from_str(&data)?; + chunk.attributes.clone_from(&object.attributes); + let data = serde_json::to_string(&chunk)?; + bytes = bytes.saturating_add(i64::try_from(data.len()).unwrap_or(i64::MAX)); + chunks.push((index, data)); + if chunks.len() % 32 == 0 { + tokio::task::yield_now().await; + } + } + let other_bytes: i64 = sqlx::query_scalar("SELECT CAST(COALESCE(SUM(storage_bytes), 0) AS BIGINT) FROM file_search_attachments WHERE store_id = $1 AND file_id != $2") + .bind(store_id).bind(file_id).fetch_one(&mut *tx).await?; + super::validate_capacity(other_bytes.saturating_add(bytes), 0)?; + for (index, data) in chunks { + sqlx::query( + "UPDATE file_search_chunks SET data = $4 WHERE store_id = $1 AND file_id = $2 AND chunk_index = $3", + ) + .bind(store_id) + .bind(file_id) + .bind(index) + .bind(data) + .execute(&mut *tx) + .await?; + } + sqlx::query("UPDATE file_search_attachments SET storage_bytes = $3 WHERE store_id = $1 AND file_id = $2") + .bind(store_id) + .bind(file_id) + .bind(bytes) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(object) + } + + pub(crate) async fn parsed_content( + &self, + store_id: &str, + file_id: &str, + ) -> Result, FileSearchError> { + let row: Option> = sqlx::query_scalar( + "SELECT parsed_content FROM file_search_attachments WHERE store_id = $1 AND file_id = $2 AND status = $3", + ) + .bind(store_id) + .bind(file_id) + .bind(AttachmentStatus::Completed.as_str()) + .fetch_optional(self.pool.as_ref()) + .await?; + row.ok_or_else(|| FileSearchError::NotFound("Completed vector store file not found".into())) + } +} diff --git a/crates/agentic-server-core/src/tool/file_search/ingest.rs b/crates/agentic-server-core/src/tool/file_search/ingest.rs index c0ce48b1..158418f7 100644 --- a/crates/agentic-server-core/src/tool/file_search/ingest.rs +++ b/crates/agentic-server-core/src/tool/file_search/ingest.rs @@ -96,6 +96,21 @@ pub(super) fn extract_and_chunk( chunking: &StaticChunking, cancelled: &AtomicBool, ) -> Result { + let text = extract_text(bytes, filename, content_type, cancelled)?; + let chunks = chunks(&text, chunking, cancelled)?; + Ok(ExtractedDocument { text, chunks }) +} + +/// Extracts bounded original text without applying a new chunking policy. +pub(super) fn extract_text( + bytes: Vec, + filename: &str, + content_type: &str, + cancelled: &AtomicBool, +) -> Result { + if cancelled.load(Ordering::Relaxed) { + return Err(FileSearchError::Unavailable("File ingestion was cancelled".into())); + } validate_content_type(filename, content_type)?; let content_type = content_type.split(';').next().unwrap_or_default().trim(); let text = if is_pdf(filename, content_type) { @@ -113,8 +128,10 @@ pub(super) fn extract_and_chunk( if text.contains('\0') { return invalid("The file contains binary content instead of text"); } - let chunks = chunks(&text, chunking, cancelled)?; - Ok(ExtractedDocument { text, chunks }) + if cancelled.load(Ordering::Relaxed) { + return Err(FileSearchError::Unavailable("File ingestion was cancelled".into())); + } + Ok(text) } #[cfg(not(feature = "file-search-pdf"))] @@ -276,6 +293,32 @@ pub(super) fn limit_context( mod tests { use super::*; + #[test] + fn original_text_extraction_retains_validation_limits_and_cancellation() { + let active = AtomicBool::new(false); + let text = b"Original text.\n".to_vec(); + assert_eq!( + extract_text(text.clone(), "source.txt", "text/plain", &active).unwrap(), + "Original text.\n" + ); + assert!(matches!( + extract_text(text, "source.txt", "text/plain", &AtomicBool::new(true)), + Err(FileSearchError::Unavailable(_)) + )); + for bytes in [ + vec![0xff], + vec![0], + Vec::new(), + b" \n ".to_vec(), + vec![b'a'; MAX_EXTRACTED_BYTES + 1], + ] { + assert!(matches!( + extract_text(bytes, "source.txt", "text/plain", &active), + Err(FileSearchError::InvalidRequest(_)) + )); + } + } + #[test] fn cancelled_context_preparation_exits_before_tokenizing() { let cancelled = AtomicBool::new(true); diff --git a/crates/agentic-server-core/src/tool/file_search/service.rs b/crates/agentic-server-core/src/tool/file_search/service.rs index 40ac7e90..9148e0cd 100644 --- a/crates/agentic-server-core/src/tool/file_search/service.rs +++ b/crates/agentic-server-core/src/tool/file_search/service.rs @@ -13,6 +13,9 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore}; #[path = "files.rs"] mod files; pub use files::{FileDownload, FileUpload}; +#[path = "stores.rs"] +mod stores; +use stores::validate_store_fields; use super::{embeddings::Embeddings, ingest, models::Models, ranking}; use crate::{ @@ -22,10 +25,10 @@ use crate::{ local_files::LocalFiles, }, types::file_search::{ - AttachFileRequest, ChunkingStrategy, CreateVectorStoreRequest, DeleteObject, FileAttributes, FileCounts, - FileObject, FileSearchConfig, FileSearchError, ListParams, ListResponse, Ranker, RankingOptions, SearchMode, - SearchQuery, SearchRequest, SearchResponse, VectorStoreFileObject, VectorStoreObject, invalid, - validate_attributes, + AttachFileRequest, AttachmentStatus, ChunkingStrategy, CreateVectorStoreRequest, DeleteObject, FileAttributes, + FileCounts, FileObject, FileSearchConfig, FileSearchError, ListParams, ListResponse, Ranker, RankingOptions, + SearchMode, SearchQuery, SearchRequest, SearchResponse, VectorStoreFileChunkingStrategy, VectorStoreFileObject, + VectorStoreObject, VectorStoreStatus, invalid, validate_attributes, }, }; @@ -162,6 +165,9 @@ impl FileSearchService { let mut params = params.clone(); params.limit = Some(params.limit.unwrap_or(10000)); validate_pagination(¶ms, 10000)?; + if params.filter.is_some() { + return invalid("filter applies only to vector store file lists"); + } if let Some(purpose) = ¶ms.purpose { files::validate_purpose(purpose)?; } @@ -243,16 +249,13 @@ impl FileSearchService { request: CreateVectorStoreRequest, ) -> Result { let permit = self.permit()?; - if request.name.as_ref().is_some_and(|name| name.len() > 256) { - return invalid("vector store name must not exceed 256 bytes"); - } - if request.metadata.len() > 16 - || request - .metadata - .iter() - .any(|(key, value)| key.is_empty() || key.len() > 64 || value.len() > 512) - { - return invalid("metadata accepts at most 16 entries, with 1 to 64 byte keys and values up to 512 bytes"); + validate_store_fields( + request.name.as_deref(), + request.metadata.as_ref(), + request.expires_after.as_ref(), + )?; + if request.description.as_ref().is_some_and(|value| value.len() > 512) { + return invalid("description must not exceed 512 bytes"); } if request.file_ids.len() > 16 || request.file_ids.iter().collect::>().len() != request.file_ids.len() @@ -268,7 +271,11 @@ impl FileSearchService { name: request.name.unwrap_or_default(), usage_bytes: 0, file_counts: FileCounts::default(), - status: "completed".into(), + status: VectorStoreStatus::Completed, + description: request.description, + last_active_at: None, + expires_at: None, + expires_after: request.expires_after, metadata: request.metadata, }; let mut attachments = Vec::new(); @@ -315,6 +322,9 @@ impl FileSearchService { params: &ListParams, ) -> Result, FileSearchError> { validate_list(params)?; + if params.filter.is_some() { + return invalid("filter applies only to vector store file lists"); + } let stores: Vec = self.storage.list(Collection::Stores, None, params).await?; let mut page = page(stores, params, |store| &store.id); for store in &mut page.data { @@ -363,6 +373,10 @@ impl FileSearchService { Ok(prepared.object) } + #[allow( + clippy::too_many_lines, + reason = "keeps bounded parsing, contextualization, and atomic attachment preparation together" + )] async fn prepare( &self, store_id: &str, @@ -395,6 +409,9 @@ impl FileSearchService { self.config.contextual_retrieval_params.model.as_ref(), )?; } + let resolved_chunking = VectorStoreFileChunkingStrategy::Static { + config: chunking.clone(), + }; let filename = file.filename.clone(); let worker_permit = permit.clone(); let cancelled = Arc::new(AtomicBool::new(false)); @@ -457,13 +474,14 @@ impl FileSearchService { object: "vector_store.file".into(), created_at: chrono::Utc::now().timestamp(), vector_store_id: store_id.into(), - status: "completed".into(), + status: AttachmentStatus::Completed, usage_bytes: size_i64(usage_bytes)?, attributes: request.attributes, - chunking_strategy: strategy, + chunking_strategy: resolved_chunking, last_error: None, }; Ok(PreparedAttachment { + parsed_content: document.text, object, chunks, dimensions: size_i64(embedding_dimensions)?, @@ -704,6 +722,7 @@ impl FileSearchService { if prepare_context { data = self.prepare_context(data, permit).await?; } + self.storage.refresh_activity(store_ids).await?; let visible = self.storage.visible_result_files(store_ids, &data).await?; data.retain(|result| visible.contains(&result.file_id)); Ok(SearchResponse { diff --git a/crates/agentic-server-core/src/tool/file_search/stores.rs b/crates/agentic-server-core/src/tool/file_search/stores.rs new file mode 100644 index 00000000..6288e789 --- /dev/null +++ b/crates/agentic-server-core/src/tool/file_search/stores.rs @@ -0,0 +1,109 @@ +//! Vector Store lifecycle operations, separate from model preparation and transport. +use super::{ + Arc, AtomicBool, CancelIngestionOnDrop, FileObject, FileSearchError, FileSearchService, VectorStoreFileObject, + VectorStoreObject, ingest, invalid, validate_attributes, +}; + +pub(super) fn validate_store_fields( + name: Option<&str>, + metadata: Option<&std::collections::BTreeMap>, + expiration: Option<&crate::types::file_search::VectorStoreExpiresAfter>, +) -> Result<(), FileSearchError> { + if name.is_some_and(|name| name.len() > 256) { + return invalid("vector store name must not exceed 256 bytes"); + } + if metadata.is_some_and(|metadata| { + metadata.len() > 16 + || metadata + .iter() + .any(|(key, value)| key.is_empty() || key.len() > 64 || value.len() > 512) + }) { + return invalid("metadata accepts at most 16 entries, with 1 to 64 byte keys and values up to 512 bytes"); + } + if let Some(expiration) = expiration { + expiration.validate()?; + } + Ok(()) +} + +impl FileSearchService { + /// Updates only provided fields. Expired stores cannot be revived. + /// # Errors + /// Returns validation, not-found, or storage errors. + pub async fn update_vector_store( + &self, + id: &str, + request: crate::types::file_search::UpdateVectorStoreRequest, + ) -> Result { + validate_store_fields( + request.name.0.as_ref().and_then(Option::as_deref), + request.metadata.0.as_ref().and_then(Option::as_ref), + request.expires_after.0.as_ref().and_then(Option::as_ref), + )?; + self.storage.update_store(id, request).await?; + self.storage.store_object(id).await + } + + /// Replaces attachment attributes, including the attributes used for retrieval. + /// # Errors + /// Returns validation, not-found, or storage errors. + pub async fn update_vector_store_file( + &self, + store_id: &str, + file_id: &str, + request: crate::types::file_search::UpdateVectorStoreFileRequest, + ) -> Result { + let _permit = self.permit()?; + validate_attributes(&request.attributes)?; + self.storage + .update_attachment(store_id, file_id, request.attributes) + .await + } + + /// Returns extracted original text as a single bounded page, without chunk overlap or context hints. + /// Legacy attachments reparse the original upload without calling model providers. + /// # Errors + /// Returns not-found, parser, resource-limit, or storage errors. + pub async fn vector_store_file_content( + &self, + store_id: &str, + file_id: &str, + ) -> Result { + let permit = self.permit()?; + self.get_vector_store_file(store_id, file_id).await?; + let text = if let Some(text) = self.storage.parsed_content(store_id, file_id).await? { + text + } else { + let uploaded = self.storage.file(file_id).await?; + let file: FileObject = serde_json::from_str(&uploaded.data)?; + let bytes = self + .read_content(file_id, file.bytes, uploaded.content_base64, permit.clone()) + .await?; + let cancelled = Arc::new(AtomicBool::new(false)); + let _cancel_on_drop = CancelIngestionOnDrop(cancelled.clone()); + tokio::task::spawn_blocking(move || { + let _permit = permit; + ingest::extract_text(bytes, &file.filename, &uploaded.content_type, &cancelled) + }) + .await?? + }; + self.get_vector_store_file(store_id, file_id).await?; + Ok(crate::types::file_search::VectorStoreFileContentPage { + object: "vector_store.file_content.page".into(), + data: vec![crate::types::file_search::ParsedFileContent::Text { text }], + has_more: false, + next_page: None, + }) + } + + /// Removes attachments and search data from at most `limit` expired stores, preserving uploads and store metadata. + /// Explicit cleanup is restart-safe and does not start background tasks. + /// # Errors + /// Returns invalid-request unless limit is 1..1000, or storage errors. + pub async fn cleanup_expired_vector_stores(&self, limit: usize) -> Result { + if !(1..=1000).contains(&limit) { + return invalid("cleanup limit must be between 1 and 1000"); + } + self.storage.expire_stores(limit).await + } +} diff --git a/crates/agentic-server-core/src/types/file_search.rs b/crates/agentic-server-core/src/types/file_search.rs index 461b2632..9d6b38a7 100644 --- a/crates/agentic-server-core/src/types/file_search.rs +++ b/crates/agentic-server-core/src/types/file_search.rs @@ -461,10 +461,12 @@ impl Default for StaticChunking { #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] pub struct CreateVectorStoreRequest { pub name: Option, + pub description: Option, + pub expires_after: Option, #[serde(default)] pub file_ids: Vec, #[serde(default)] - pub metadata: BTreeMap, + pub metadata: Option>, pub chunking_strategy: Option, } @@ -473,7 +475,8 @@ pub struct CreateVectorStoreRequest { #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] pub struct AttachFileRequest { pub file_id: String, - #[serde(default)] + #[serde(default, deserialize_with = "null_default")] + #[cfg_attr(feature = "openapi", schema(nullable = true))] pub attributes: FileAttributes, pub chunking_strategy: Option, } @@ -497,8 +500,18 @@ pub struct VectorStoreObject { pub name: String, pub usage_bytes: i64, pub file_counts: FileCounts, - pub status: String, - pub metadata: BTreeMap, + pub status: VectorStoreStatus, + #[serde(default)] + pub description: Option, + #[serde(default)] + #[cfg_attr(feature = "openapi", schema(required = true))] + pub last_active_at: Option, + #[serde(default)] + pub expires_after: Option, + #[serde(default)] + pub expires_at: Option, + #[cfg_attr(feature = "openapi", schema(required = true))] + pub metadata: Option>, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -508,17 +521,17 @@ pub struct VectorStoreFileObject { pub object: String, pub created_at: i64, pub vector_store_id: String, - pub status: String, + pub status: AttachmentStatus, pub usage_bytes: i64, pub attributes: FileAttributes, - pub chunking_strategy: ChunkingStrategy, + pub chunking_strategy: VectorStoreFileChunkingStrategy, pub last_error: Option, } #[derive(Clone, Debug, Serialize, Deserialize)] #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] pub struct VectorStoreFileError { - pub code: String, + pub code: VectorStoreFileErrorCode, pub message: String, } @@ -553,6 +566,7 @@ pub enum ListOrder { #[serde(deny_unknown_fields)] #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] pub struct ListParams { + pub filter: Option, pub purpose: Option, pub limit: Option, pub after: Option, @@ -761,3 +775,172 @@ impl SearchFilter { fn same_type(left: &AttributeValue, right: &AttributeValue) -> bool { std::mem::discriminant(left) == std::mem::discriminant(right) } + +/// A nullable patch distinguishes an omitted member from explicit JSON null. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(transparent)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct NullablePatch(pub Option>); + +impl Default for NullablePatch { + fn default() -> Self { + Self(None) + } +} + +impl NullablePatch { + #[must_use] + pub const fn is_missing(&self) -> bool { + self.0.is_none() + } +} + +fn deserialize_patch<'de, D, T>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, + T: Deserialize<'de>, +{ + Option::::deserialize(deserializer).map(|value| NullablePatch(Some(value))) +} + +fn null_default<'de, D, T>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, + T: Deserialize<'de> + Default, +{ + Ok(Option::::deserialize(deserializer)?.unwrap_or_default()) +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct UpdateVectorStoreRequest { + #[serde( + default, + deserialize_with = "deserialize_patch", + skip_serializing_if = "NullablePatch::is_missing" + )] + #[cfg_attr(feature = "openapi", schema(value_type = Option))] + pub name: NullablePatch, + #[serde( + default, + deserialize_with = "deserialize_patch", + skip_serializing_if = "NullablePatch::is_missing" + )] + #[cfg_attr(feature = "openapi", schema(value_type = Option>))] + pub metadata: NullablePatch>, + #[serde( + default, + deserialize_with = "deserialize_patch", + skip_serializing_if = "NullablePatch::is_missing" + )] + #[cfg_attr(feature = "openapi", schema(value_type = Option))] + pub expires_after: NullablePatch, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct UpdateVectorStoreFileRequest { + #[serde(deserialize_with = "null_default")] + #[cfg_attr(feature = "openapi", schema(nullable = true))] + pub attributes: FileAttributes, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum VectorStoreExpirationAnchor { + LastActiveAt, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct VectorStoreExpiresAfter { + pub anchor: VectorStoreExpirationAnchor, + pub days: u16, +} + +impl VectorStoreExpiresAfter { + /// # Errors + /// Rejects policies outside the supported one to 365 day window. + pub fn validate(&self) -> Result<(), FileSearchError> { + if !(1..=365).contains(&self.days) { + return invalid("expires_after.days must be between 1 and 365"); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum VectorStoreStatus { + InProgress, + #[default] + Completed, + Expired, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum AttachmentStatus { + InProgress, + #[default] + Completed, + Cancelled, + Failed, +} + +impl AttachmentStatus { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::InProgress => "in_progress", + Self::Completed => "completed", + Self::Cancelled => "cancelled", + Self::Failed => "failed", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum VectorStoreFileErrorCode { + ServerError, + UnsupportedFile, + InvalidFile, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct VectorStoreFileContentPage { + pub object: String, + pub data: Vec, + pub has_more: bool, + pub next_page: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum ParsedFileContent { + Text { text: String }, +} + +/// Reported chunk boundaries, distinct from request-only auto/contextual configuration. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub enum VectorStoreFileChunkingStrategy { + Static { + #[serde(rename = "static")] + config: StaticChunking, + }, + // Legacy rows retained request settings rather than resolved chunk boundaries. + #[serde(alias = "auto", alias = "contextual")] + Other, +} diff --git a/crates/agentic-server-core/tests/file_search_models.rs b/crates/agentic-server-core/tests/file_search_models.rs index ddf1d7b8..f6593d28 100644 --- a/crates/agentic-server-core/tests/file_search_models.rs +++ b/crates/agentic-server-core/tests/file_search_models.rs @@ -195,6 +195,16 @@ async fn contextual_embedding_changes_retrieval_but_preserves_original_source() .await .unwrap(); let id = attach(&setup.service, &store.id, "unadorned source", Some(contextual())).await; + let attachment = setup.service.get_vector_store_file(&store.id, &id).await.unwrap(); + assert_eq!( + serde_json::to_value(attachment).unwrap()["chunking_strategy"], + json!({"type":"static","static":{"max_chunk_size_tokens":700,"chunk_overlap_tokens":400}}) + ); + let content = setup.service.vector_store_file_content(&store.id, &id).await.unwrap(); + assert_eq!( + serde_json::to_value(content).unwrap()["data"][0]["text"], + "unadorned source" + ); let result = setup .service .search( @@ -901,3 +911,32 @@ async fn expiration_during_reranking_discards_cached_candidates() { setup.state.rerank_resume.notify_one(); assert!(search.await.unwrap().unwrap().data.is_empty()); } + +#[tokio::test] +async fn store_expiration_during_reranking_rejects_cached_candidates() { + let setup = setup(false).await; + let store = setup + .service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + attach(&setup.service, &store.id, "coral preferred", None).await; + *setup.state.rerank_pause.lock().unwrap() = true; + let service = setup.service.clone(); + let store_id = store.id.clone(); + let request: SearchRequest = + serde_json::from_value(json!({"query":"coral","ranking_options":{"ranker":"neural"}})).unwrap(); + let search = tokio::spawn(async move { service.search(&[store_id], &request).await }); + setup.state.rerank_started.notified().await; + sqlx::query("UPDATE file_search_stores SET expires_at = 1 WHERE id = $1") + .bind(&store.id) + .execute(setup.pool.as_ref()) + .await + .unwrap(); + setup.state.rerank_resume.notify_one(); + assert_eq!(search.await.unwrap().unwrap_err().status_code(), 404); + assert_eq!( + setup.service.get_vector_store(&store.id).await.unwrap().expires_at, + Some(1) + ); +} diff --git a/crates/agentic-server-core/tests/file_search_service.rs b/crates/agentic-server-core/tests/file_search_service.rs index 9297931a..ef9ce10b 100644 --- a/crates/agentic-server-core/tests/file_search_service.rs +++ b/crates/agentic-server-core/tests/file_search_service.rs @@ -340,6 +340,7 @@ async fn before_pagination_returns_adjacent_files_stores_and_attachments() { order: Some(order), after: None, purpose: None, + filter: None, }; let page = service.list_files(&before(&file_ids[5])).await.unwrap(); assert!(page.has_more); @@ -779,6 +780,18 @@ async fn embedding_service() -> ( tokio::task::JoinHandle<()>, Arc, FileSearchConfig, +) { + embedding_service_database("sqlite::memory:").await +} + +async fn embedding_service_database( + database: &str, +) -> ( + TestService, + ProviderState, + tokio::task::JoinHandle<()>, + Arc, + FileSearchConfig, ) { let state = ProviderState { mode: Arc::new(std::sync::Mutex::new(ProviderMode::Good)), @@ -801,7 +814,7 @@ async fn embedding_service() -> ( embedding_api_key: Some("never-log-this-key".into()), ..FileSearchConfig::default() }; - let pool = create_pool_with_schema(Some("sqlite::memory:")).await.unwrap(); + let pool = create_pool_with_schema(Some(database)).await.unwrap(); let service = FileSearchService::new(pool.clone(), Arc::new(reqwest::Client::new()), config.clone()).unwrap(); (TestService { service, files }, state, task, pool, config) } @@ -2324,3 +2337,90 @@ async fn postgres_file_expiring_while_publication_waits_for_store_lock_is_not_at ); assert_eq!(chunks, 0); } + +async fn store_expiration_during_models(database: &str) { + let (service, state, task, pool, _) = embedding_service_database(database).await; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let file = service + .upload_file("expire.txt", "text/plain", "assistants", b"coral reefs".to_vec()) + .await + .unwrap(); + *state.mode.lock().unwrap() = ProviderMode::Pause; + let worker = service.service.clone(); + let store_id = store.id.clone(); + let file_id = file.id.clone(); + let ingestion = tokio::spawn(async move { + worker + .attach_file( + &store_id, + AttachFileRequest { + file_id, + ..Default::default() + }, + ) + .await + }); + state.started.notified().await; + sqlx::query("UPDATE file_search_stores SET expires_at = 1 WHERE id = $1") + .bind(&store.id) + .execute(pool.as_ref()) + .await + .unwrap(); + state.resume.notify_one(); + let result = ingestion.await.unwrap(); + let chunks: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM file_search_chunks WHERE store_id = $1") + .bind(&store.id) + .fetch_one(pool.as_ref()) + .await + .unwrap(); + service.delete_vector_store(&store.id).await.unwrap(); + service.delete_file(&file.id).await.unwrap(); + assert_eq!(result.unwrap_err().status_code(), 404); + assert_eq!(chunks, 0, "late model completion must not publish expired search data"); + *state.mode.lock().unwrap() = ProviderMode::Good; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let file = attach_text( + &service, + &store.id, + "source.txt", + "coral reefs", + FileAttributes::default(), + ) + .await; + *state.mode.lock().unwrap() = ProviderMode::Pause; + let worker = service.service.clone(); + let store_id = store.id.clone(); + let search = tokio::spawn(async move { worker.search(&[store_id], &query("aquatic")).await }); + state.started.notified().await; + sqlx::query("UPDATE file_search_stores SET expires_at = 1 WHERE id = $1") + .bind(&store.id) + .execute(pool.as_ref()) + .await + .unwrap(); + state.resume.notify_one(); + let result = search.await.unwrap(); + let expired = service.get_vector_store(&store.id).await.unwrap(); + service.delete_vector_store(&store.id).await.unwrap(); + service.delete_file(&file.id).await.unwrap(); + task.abort(); + assert_eq!(result.unwrap_err().status_code(), 404); + assert_eq!(expired.status, VectorStoreStatus::Expired); + assert_eq!(expired.expires_at, Some(1), "late search must not revive expiry"); +} + +#[tokio::test] +async fn sqlite_store_expiration_during_model_calls_cannot_publish_or_revive() { + store_expiration_during_models("sqlite::memory:").await; +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_store_expiration_during_model_calls_cannot_publish_or_revive() { + store_expiration_during_models(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; +} diff --git a/crates/agentic-server-core/tests/vector_store_lifecycle.rs b/crates/agentic-server-core/tests/vector_store_lifecycle.rs new file mode 100644 index 00000000..2805cadb --- /dev/null +++ b/crates/agentic-server-core/tests/vector_store_lifecycle.rs @@ -0,0 +1,754 @@ +use std::sync::Arc; + +use agentic_core::{ + storage::{DbPool, create_pool_with_schema}, + tool::file_search::FileSearchService, + types::file_search::*, +}; +use serde_json::json; + +async fn fixture(database: &str) -> (FileSearchService, Arc, tempfile::TempDir) { + let pool = create_pool_with_schema(Some(database)).await.unwrap(); + let files = tempfile::tempdir().unwrap(); + let service = FileSearchService::new( + pool.clone(), + Arc::new(reqwest::Client::new()), + FileSearchConfig { + files_storage_dir: Some(files.path().to_owned()), + ..Default::default() + }, + ) + .unwrap(); + (service, pool, files) +} + +fn request(value: serde_json::Value) -> UpdateVectorStoreRequest { + serde_json::from_value(value).unwrap() +} +fn query() -> SearchRequest { + serde_json::from_value(json!({"query":"lunar"})).unwrap() +} + +#[allow( + clippy::too_many_lines, + reason = "keeps expiry visibility, cleanup, and independent source assertions in one database fixture" +)] +async fn lifecycle(database: &str) { + let (service, pool, _files) = fixture(database).await; + let file = service + .upload_file( + "original.txt", + "text/plain", + "assistants", + b"Original lunar policy.\n".repeat(100), + ) + .await + .unwrap(); + let first = service + .create_vector_store( + serde_json::from_value(json!({"file_ids":[file.id],"expires_after":{"anchor":"last_active_at","days":1}})) + .unwrap(), + ) + .await + .unwrap(); + let second = service + .create_vector_store(serde_json::from_value(json!({"file_ids":[file.id]})).unwrap()) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET expires_at = 1 WHERE id = $1") + .bind(&first.id) + .execute(pool.as_ref()) + .await + .unwrap(); + let expired = service.get_vector_store(&first.id).await.unwrap(); + assert_eq!(expired.status, VectorStoreStatus::Expired); + assert_eq!(expired.file_counts.total, 0); + assert_eq!(expired.usage_bytes, 0); + assert_eq!( + service + .search(std::slice::from_ref(&first.id), &query()) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!( + service + .get_vector_store_file(&first.id, &file.id) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!( + service + .vector_store_file_content(&first.id, &file.id) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!( + service + .update_vector_store(&first.id, request(json!({"expires_after":null}))) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!( + service + .update_vector_store_file( + &first.id, + &file.id, + UpdateVectorStoreFileRequest { + attributes: FileAttributes::default() + } + ) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!( + service + .detach_file(&first.id, &file.id) + .await + .unwrap_err() + .status_code(), + 404 + ); + assert_eq!(service.cleanup_expired_vector_stores(1).await.unwrap(), 1); + assert_eq!(service.cleanup_expired_vector_stores(1).await.unwrap(), 0); + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM file_search_chunks WHERE store_id = $1") + .bind(&first.id) + .fetch_one(pool.as_ref()) + .await + .unwrap(); + assert_eq!(count, 0); + assert_eq!(service.get_file(&file.id).await.unwrap().bytes, file.bytes); + assert!( + !service + .search(std::slice::from_ref(&second.id), &query()) + .await + .unwrap() + .data + .is_empty() + ); + let text = service.vector_store_file_content(&second.id, &file.id).await.unwrap(); + assert_eq!( + serde_json::to_value(&text).unwrap()["data"][0]["text"], + "Original lunar policy.\n".repeat(100) + ); + // Legacy request-shaped strategies remain readable but are never echoed as response options. + let mut legacy = serde_json::to_value(service.get_vector_store_file(&second.id, &file.id).await.unwrap()).unwrap(); + for strategy in [ + json!({"type":"auto"}), + json!({"type":"contextual","contextual":{"max_chunk_size_tokens":700,"chunk_overlap_tokens":400}}), + ] { + legacy["chunking_strategy"] = strategy; + sqlx::query("UPDATE file_search_attachments SET data = $3 WHERE store_id = $1 AND file_id = $2") + .bind(&second.id) + .bind(&file.id) + .bind(serde_json::to_string(&legacy).unwrap()) + .execute(pool.as_ref()) + .await + .unwrap(); + let attachment = service.get_vector_store_file(&second.id, &file.id).await.unwrap(); + assert_eq!( + serde_json::to_value(attachment).unwrap()["chunking_strategy"], + json!({"type":"other"}) + ); + } + // Legacy attachments recover original text from the uploaded bytes, never overlap chunks. + sqlx::query("UPDATE file_search_attachments SET parsed_content = NULL WHERE store_id = $1") + .bind(&second.id) + .execute(pool.as_ref()) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(service.vector_store_file_content(&second.id, &file.id).await.unwrap()).unwrap(), + serde_json::to_value(text).unwrap() + ); + service.delete_vector_store(&first.id).await.unwrap(); + service.delete_vector_store(&second.id).await.unwrap(); + service.delete_file(&file.id).await.unwrap(); +} + +#[tokio::test] +async fn sqlite_store_expiration_preserves_uploads_and_other_stores() { + lifecycle("sqlite::memory:").await; +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_store_expiration_preserves_uploads_and_other_stores() { + lifecycle(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; +} + +#[allow( + clippy::too_many_lines, + reason = "exercises one ordered mixed-status corpus through filtered updates and idle-store expiration" +)] +async fn updates_and_pages(database: &str) { + let (service, pool, _files) = fixture(database).await; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let mut ids = Vec::new(); + for status in [ + AttachmentStatus::Failed, + AttachmentStatus::Completed, + AttachmentStatus::Failed, + AttachmentStatus::Completed, + AttachmentStatus::Cancelled, + ] { + let file = service + .upload_file("page.txt", "text/plain", "assistants", b"lunar page".to_vec()) + .await + .unwrap(); + let mut attachment = service + .attach_file( + &store.id, + AttachFileRequest { + file_id: file.id.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + attachment.status = status; + sqlx::query("UPDATE file_search_attachments SET created_at = 100, status = $3, data = $4 WHERE store_id = $1 AND file_id = $2").bind(&store.id).bind(&file.id).bind(status.as_str()).bind(serde_json::to_string(&attachment).unwrap()).execute(pool.as_ref()).await.unwrap(); + ids.push(file.id); + } + let counts = service.get_vector_store(&store.id).await.unwrap().file_counts; + assert_eq!( + (counts.total, counts.completed, counts.failed, counts.cancelled), + (5, 2, 2, 1) + ); + let params = ListParams { + filter: Some(AttachmentStatus::Completed), + limit: Some(1), + order: Some(ListOrder::Asc), + ..Default::default() + }; + let first = service.list_vector_store_files(&store.id, ¶ms).await.unwrap(); + assert_eq!(first.data[0].id, ids[1]); + assert!(first.has_more); + let next = service + .list_vector_store_files( + &store.id, + &ListParams { + after: Some(ids[1].clone()), + ..params.clone() + }, + ) + .await + .unwrap(); + assert_eq!(next.data[0].id, ids[3]); + assert!(!next.has_more); + let previous = service + .list_vector_store_files( + &store.id, + &ListParams { + before: Some(ids[3].clone()), + ..params + }, + ) + .await + .unwrap(); + assert_eq!(previous.data[0].id, ids[1]); + assert!(!previous.has_more); + let hits = service.search(std::slice::from_ref(&store.id), &query()).await.unwrap(); + assert_eq!(hits.data.len(), 2, "only completed attachments are searchable"); + let attributes: FileAttributes = serde_json::from_value(json!({"kind":"selected"})).unwrap(); + service + .update_vector_store_file(&store.id, &ids[1], UpdateVectorStoreFileRequest { attributes }) + .await + .unwrap(); + let filter_query: SearchRequest = + serde_json::from_value(json!({"query":"lunar","filters":{"type":"eq","key":"kind","value":"selected"}})) + .unwrap(); + assert_eq!( + service + .search(std::slice::from_ref(&store.id), &filter_query) + .await + .unwrap() + .data[0] + .file_id, + ids[1] + ); + service + .update_vector_store_file( + &store.id, + &ids[1], + serde_json::from_value(json!({"attributes":null})).unwrap(), + ) + .await + .unwrap(); + assert!( + service + .search(std::slice::from_ref(&store.id), &filter_query) + .await + .unwrap() + .data + .is_empty() + ); + // Policy updates use prior activity rather than refreshing an idle store. + sqlx::query("UPDATE file_search_stores SET last_active_at = 100 WHERE id = $1") + .bind(&store.id) + .execute(pool.as_ref()) + .await + .unwrap(); + assert_eq!( + service.get_vector_store(&store.id).await.unwrap().last_active_at, + Some(100) + ); + let unchanged = service + .update_vector_store(&store.id, request(json!({"name":"renamed","metadata":{"team":"docs"}}))) + .await + .unwrap(); + assert_eq!(unchanged.last_active_at, Some(100)); + let expired = service + .update_vector_store( + &store.id, + request(json!({"expires_after":{"anchor":"last_active_at","days":1}})), + ) + .await + .unwrap(); + assert_eq!(expired.expires_at, Some(86500)); + assert_eq!(expired.status, VectorStoreStatus::Expired); + assert_eq!( + service + .update_vector_store(&store.id, request(json!({"expires_after":null}))) + .await + .unwrap_err() + .status_code(), + 404 + ); + service.delete_vector_store(&store.id).await.unwrap(); + for id in ids { + service.delete_file(&id).await.unwrap(); + } +} + +#[tokio::test] +async fn sqlite_status_filter_precedes_pagination_and_updates_do_not_refresh_activity() { + updates_and_pages("sqlite::memory:").await; +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_status_filter_precedes_pagination_and_updates_do_not_refresh_activity() { + updates_and_pages(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; +} + +async fn database_clock(pool: &DbPool) -> i64 { + sqlx::query_scalar("SELECT CAST(FLOOR(EXTRACT(EPOCH FROM clock_timestamp())) AS BIGINT)") + .fetch_one(pool) + .await + .unwrap() +} + +async fn wait_for_lock(pool: &DbPool, blocker: i64) { + tokio::time::timeout(std::time::Duration::from_secs(4), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE wait_event_type = 'Lock' AND CAST($1 AS INTEGER) = ANY(pg_blocking_pids(pid)))").bind(blocker).fetch_one(pool).await.unwrap(); + if waiting { break; } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.expect("operation must reach the contended store lock"); +} + +async fn wait_until(pool: &DbPool, deadline: i64) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while database_clock(pool).await < deadline { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_expiry_is_rechecked_after_contended_store_lock_for_publication_and_policy() { + let (service, pool, _files) = fixture(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; + for publication in [true, false] { + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let file = service + .upload_file("wait.txt", "text/plain", "assistants", b"lunar".to_vec()) + .await + .unwrap(); + let deadline = database_clock(&pool).await + 3; + sqlx::query("UPDATE file_search_stores SET expires_at = $2 WHERE id = $1") + .bind(&store.id) + .bind(deadline) + .execute(pool.as_ref()) + .await + .unwrap(); + let mut blocker = pool.begin().await.unwrap(); + let pid: i64 = sqlx::query_scalar("SELECT CAST(pg_backend_pid() AS BIGINT)") + .fetch_one(&mut *blocker) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET id = id WHERE id = $1") + .bind(&store.id) + .execute(&mut *blocker) + .await + .unwrap(); + let worker = service.clone(); + let store_id = store.id.clone(); + let file_id = file.id.clone(); + let operation = tokio::spawn(async move { + if publication { + worker + .attach_file( + &store_id, + AttachFileRequest { + file_id, + ..Default::default() + }, + ) + .await + .map(|_| ()) + } else { + worker + .update_vector_store(&store_id, request(json!({"expires_after":null}))) + .await + .map(|_| ()) + } + }); + wait_for_lock(&pool, pid).await; + assert!(database_clock(&pool).await < deadline); + wait_until(&pool, deadline).await; + blocker.commit().await.unwrap(); + let result = operation.await.unwrap(); + let chunks: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM file_search_chunks WHERE store_id = $1") + .bind(&store.id) + .fetch_one(pool.as_ref()) + .await + .unwrap(); + let expired = service.get_vector_store(&store.id).await.unwrap(); + service.delete_vector_store(&store.id).await.unwrap(); + service.delete_file(&file.id).await.unwrap(); + assert_eq!(result.unwrap_err().status_code(), 404); + assert_eq!(expired.expires_at, Some(deadline)); + assert_eq!(expired.status, VectorStoreStatus::Expired); + assert_eq!(chunks, 0); + } +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_cleanup_rechecks_policy_after_waiting_for_committed_extension() { + let (service, pool, _files) = fixture(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let file = service + .upload_file("policy.txt", "text/plain", "assistants", b"lunar".to_vec()) + .await + .unwrap(); + service + .attach_file( + &store.id, + AttachFileRequest { + file_id: file.id.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + let deadline = database_clock(&pool).await + 2; + sqlx::query("UPDATE file_search_stores SET expires_at = $2 WHERE id = $1") + .bind(&store.id) + .bind(deadline) + .execute(pool.as_ref()) + .await + .unwrap(); + // Hold the same store-row policy write as update_store before COMMIT. A sweeper sees + // the previously committed deadline, waits, then must recheck the new policy. + let mut policy = pool.begin().await.unwrap(); + let pid: i64 = sqlx::query_scalar("SELECT CAST(pg_backend_pid() AS BIGINT)") + .fetch_one(&mut *policy) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET expires_after_days = 1, expires_at = $2 WHERE id = $1") + .bind(&store.id) + .bind(deadline + 86400) + .execute(&mut *policy) + .await + .unwrap(); + wait_until(&pool, deadline).await; + let worker = service.clone(); + let cleanup = tokio::spawn(async move { worker.cleanup_expired_vector_stores(1000).await }); + wait_for_lock(&pool, pid).await; + policy.commit().await.unwrap(); + let count = cleanup.await.unwrap().unwrap(); + let current = service.get_vector_store(&store.id).await.unwrap(); + let content = service.vector_store_file_content(&store.id, &file.id).await; + service.delete_vector_store(&store.id).await.unwrap(); + service.delete_file(&file.id).await.unwrap(); + assert_eq!(count, 0); + assert_eq!(current.status, VectorStoreStatus::Completed); + assert_eq!(current.file_counts.completed, 1); + assert!(content.is_ok()); +} + +#[tokio::test] +async fn sqlite_ingestion_and_empty_search_refresh_activity_and_cleanup_is_bounded() { + let (service, pool, _files) = fixture("sqlite::memory:").await; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET last_active_at = 1 WHERE id = $1") + .bind(&store.id) + .execute(pool.as_ref()) + .await + .unwrap(); + let file = service + .upload_file("activity.txt", "text/plain", "assistants", b"lunar activity".to_vec()) + .await + .unwrap(); + service + .attach_file( + &store.id, + AttachFileRequest { + file_id: file.id, + ..Default::default() + }, + ) + .await + .unwrap(); + let active = service.get_vector_store(&store.id).await.unwrap(); + assert!(active.last_active_at.unwrap() > 1); + sqlx::query("UPDATE file_search_stores SET last_active_at = 2 WHERE id = $1") + .bind(&store.id) + .execute(pool.as_ref()) + .await + .unwrap(); + let miss = service + .search( + std::slice::from_ref(&store.id), + &serde_json::from_value(json!({"query":"absent"})).unwrap(), + ) + .await + .unwrap(); + assert!(miss.data.is_empty()); + assert!( + service + .get_vector_store(&store.id) + .await + .unwrap() + .last_active_at + .unwrap() + > 2 + ); + let second = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET expires_at = 1 WHERE id IN ($1, $2)") + .bind(&store.id) + .bind(&second.id) + .execute(pool.as_ref()) + .await + .unwrap(); + assert!(service.cleanup_expired_vector_stores(0).await.is_err()); + assert!(service.cleanup_expired_vector_stores(1001).await.is_err()); + assert_eq!(service.cleanup_expired_vector_stores(1).await.unwrap(), 1); + assert_eq!(service.cleanup_expired_vector_stores(1).await.unwrap(), 1); + assert_eq!(service.cleanup_expired_vector_stores(1).await.unwrap(), 0); +} + +#[tokio::test] +async fn sqlite_cleanup_waiting_for_policy_extension_rechecks_the_deadline() { + let directory = tempfile::tempdir().unwrap(); + let (service, pool, _files) = fixture(&format!( + "sqlite://{}?mode=rwc", + directory.path().join("lifecycle.db").display() + )) + .await; + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + let deadline: i64 = sqlx::query_scalar("SELECT CAST(strftime('%s', 'now') AS BIGINT) + 1") + .fetch_one(pool.as_ref()) + .await + .unwrap(); + sqlx::query("UPDATE file_search_stores SET expires_at = $2 WHERE id = $1") + .bind(&store.id) + .bind(deadline) + .execute(pool.as_ref()) + .await + .unwrap(); + let mut policy = pool.begin().await.unwrap(); + sqlx::query("UPDATE file_search_stores SET expires_after_days = 1, expires_at = $2 WHERE id = $1") + .bind(&store.id) + .bind(deadline + 86400) + .execute(&mut *policy) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(1100)).await; + let mut cleanup = Box::pin(service.cleanup_expired_vector_stores(1)); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(100), &mut cleanup) + .await + .is_err() + ); + policy.commit().await.unwrap(); + assert_eq!(cleanup.await.unwrap(), 0); + assert_eq!( + service.get_vector_store(&store.id).await.unwrap().status, + VectorStoreStatus::Completed + ); + pool.close().await; +} + +#[tokio::test] +async fn legacy_large_chunk_content_recovery_does_not_rechunk() { + let (service, pool, _files) = fixture("sqlite::memory:").await; + // Each repeated word requires a cl100k token, exceeding the default 819,600-token + // budget while remaining well inside the 4096/0 ingestion and extracted-byte limits. + let text = "a ".repeat(900_000); + let file = service + .upload_file("large-legacy.txt", "text/plain", "assistants", text.as_bytes().to_vec()) + .await + .unwrap(); + let store = service + .create_vector_store(CreateVectorStoreRequest::default()) + .await + .unwrap(); + service + .attach_file( + &store.id, + AttachFileRequest { + file_id: file.id.clone(), + chunking_strategy: Some(ChunkingStrategy::Static { + config: StaticChunking { + max_chunk_size_tokens: 4096, + chunk_overlap_tokens: 0, + }, + }), + ..Default::default() + }, + ) + .await + .unwrap(); + // Migration 0007 leaves this column null for existing attachments. + sqlx::query("UPDATE file_search_attachments SET parsed_content = NULL WHERE store_id = $1 AND file_id = $2") + .bind(&store.id) + .bind(&file.id) + .execute(pool.as_ref()) + .await + .unwrap(); + let content = service.vector_store_file_content(&store.id, &file.id).await.unwrap(); + assert!(!content.has_more); + assert!(content.next_page.is_none()); + assert_eq!(content.data.len(), 1); + let ParsedFileContent::Text { text: recovered } = &content.data[0]; + assert_eq!(recovered, &text); +} + +#[tokio::test] +#[ignore = "requires isolated TEST_POSTGRES_URL"] +async fn postgres_store_updates_round_trip_omitted_null_and_value() { + let (service, _pool, _files) = fixture(&std::env::var("TEST_POSTGRES_URL").unwrap()).await; + let store = service + .create_vector_store( + serde_json::from_value(json!({ + "name":"original", "metadata":{"purpose":"docs"}, + "expires_after":{"anchor":"last_active_at","days":2} + })) + .unwrap(), + ) + .await + .unwrap(); + service + .update_vector_store(&store.id, request(json!({}))) + .await + .unwrap(); + let omitted = service.get_vector_store(&store.id).await.unwrap(); + service + .update_vector_store( + &store.id, + request(json!({ + "name":"changed", "metadata":{"team":"search"}, + "expires_after":{"anchor":"last_active_at","days":3} + })), + ) + .await + .unwrap(); + let replaced = service.get_vector_store(&store.id).await.unwrap(); + service + .update_vector_store(&store.id, request(json!({"metadata":null}))) + .await + .unwrap(); + let cleared_metadata = service.get_vector_store(&store.id).await.unwrap(); + service + .update_vector_store(&store.id, request(json!({"name":null,"expires_after":null}))) + .await + .unwrap(); + let cleared = service.get_vector_store(&store.id).await.unwrap(); + service + .update_vector_store(&store.id, request(json!({}))) + .await + .unwrap(); + let omitted_after_null = service.get_vector_store(&store.id).await.unwrap(); + service + .update_vector_store( + &store.id, + request(json!({ + "name":"restored", "metadata":{"revision":"2"}, + "expires_after":{"anchor":"last_active_at","days":1} + })), + ) + .await + .unwrap(); + let restored = service.get_vector_store(&store.id).await.unwrap(); + service.delete_vector_store(&store.id).await.unwrap(); + + let activity = store.last_active_at.unwrap(); + assert_eq!(omitted.name, "original"); + assert_eq!( + serde_json::to_value(&omitted.metadata).unwrap(), + json!({"purpose":"docs"}) + ); + assert_eq!(omitted.expires_after.unwrap().days, 2); + assert_eq!(omitted.expires_at, Some(activity + 172_800)); + assert_eq!(replaced.name, "changed"); + assert_eq!( + serde_json::to_value(&replaced.metadata).unwrap(), + json!({"team":"search"}) + ); + assert_eq!(replaced.expires_after.unwrap().days, 3); + assert_eq!(replaced.expires_at, Some(activity + 259_200)); + assert_eq!(cleared_metadata.name, "changed"); + assert!(cleared_metadata.metadata.is_none()); + assert_eq!(cleared_metadata.expires_after.unwrap().days, 3); + assert_eq!(cleared_metadata.expires_at, Some(activity + 259_200)); + for object in [cleared, omitted_after_null] { + assert_eq!(object.name, ""); + assert!(object.metadata.is_none()); + assert!(object.expires_after.is_none()); + assert!(object.expires_at.is_none()); + assert_eq!(object.last_active_at, Some(activity)); + assert_eq!(object.status, VectorStoreStatus::Completed); + } + assert_eq!(restored.name, "restored"); + assert_eq!( + serde_json::to_value(&restored.metadata).unwrap(), + json!({"revision":"2"}) + ); + assert_eq!(restored.expires_after.unwrap().days, 1); + assert_eq!(restored.expires_at, Some(activity + 86400)); + assert_eq!(restored.last_active_at, Some(activity)); +} diff --git a/crates/agentic-server/src/handler/http/file_search.rs b/crates/agentic-server/src/handler/http/file_search.rs index becc88da..3229b7b0 100644 --- a/crates/agentic-server/src/handler/http/file_search.rs +++ b/crates/agentic-server/src/handler/http/file_search.rs @@ -5,7 +5,7 @@ use agentic_core::tool::ToolError; use agentic_core::tool::file_search::{FileSearchService, MAX_FILE_BYTES}; use agentic_core::types::file_search::{ AttachFileRequest, CreateVectorStoreRequest, FileExpirationAnchor, FileExpiresAfter, FileSearchError, ListParams, - SearchRequest, + SearchRequest, UpdateVectorStoreFileRequest, UpdateVectorStoreRequest, }; #[path = "multipart_limits.rs"] mod multipart_limits; @@ -34,7 +34,9 @@ pub(crate) fn router() -> Router { .route("/v1/vector_stores", post(create_vector_store).get(list_vector_stores)) .route( "/v1/vector_stores/{store_id}", - get(get_vector_store).delete(delete_vector_store), + get(get_vector_store) + .post(update_vector_store) + .delete(delete_vector_store), ) .route( "/v1/vector_stores/{store_id}/files", @@ -42,7 +44,13 @@ pub(crate) fn router() -> Router { ) .route( "/v1/vector_stores/{store_id}/files/{file_id}", - get(get_vector_store_file).delete(detach_file), + get(get_vector_store_file) + .post(update_vector_store_file) + .delete(detach_file), + ) + .route( + "/v1/vector_stores/{store_id}/files/{file_id}/content", + get(vector_store_file_content), ) .route("/v1/vector_stores/{store_id}/search", post(search)) } @@ -376,7 +384,7 @@ pub(crate) async fn attach_file( #[cfg_attr(feature = "openapi", utoipa::path( get, path = "/v1/vector_stores/{store_id}/files", - params(("store_id" = String, Path, description = "Object identifier"), ("limit" = Option, Query, description = "Page size, 1 to 100"), ("after" = Option, Query), ("before" = Option, Query), ("order" = Option, Query)), + params(("store_id" = String, Path, description = "Object identifier"), ("filter" = Option, Query), ("limit" = Option, Query, description = "Page size, 1 to 100"), ("after" = Option, Query), ("before" = Option, Query), ("order" = Option, Query)), responses((status = 200, description = "Success", body = agentic_core::types::file_search::ListResponse), (status = 400, description = "Invalid request", body = crate::openapi::ApiErrorResponse), (status = 404, description = "Object not found", body = crate::openapi::ApiErrorResponse)), security(("bearer_auth" = [])), tag = "file_search", ))] @@ -452,3 +460,66 @@ pub(crate) async fn search( }; result(search.search(&[id], &request).await) } + +#[cfg_attr(feature = "openapi", utoipa::path( + post, path = "/v1/vector_stores/{store_id}", + params(("store_id" = String, Path, description = "Object identifier")), + request_body = agentic_core::types::file_search::UpdateVectorStoreRequest, + responses((status = 200, description = "Success", body = agentic_core::types::file_search::VectorStoreObject), (status = 400, description = "Invalid request", body = crate::openapi::ApiErrorResponse), (status = 404, description = "Object not found", body = crate::openapi::ApiErrorResponse)), + security(("bearer_auth" = [])), tag = "file_search", +))] +pub(crate) async fn update_vector_store( + State(state): State, + Path(id): Path, + request: Result, JsonRejection>, +) -> Response { + let search = match service(&state) { + Ok(service) => service, + Err(error) => return *error, + }; + let request = match body(request) { + Ok(request) => request, + Err(error) => return *error, + }; + result(search.update_vector_store(&id, request).await) +} + +#[cfg_attr(feature = "openapi", utoipa::path( + post, path = "/v1/vector_stores/{store_id}/files/{file_id}", + params(("store_id" = String, Path, description = "Object identifier"), ("file_id" = String, Path, description = "Object identifier")), + request_body = agentic_core::types::file_search::UpdateVectorStoreFileRequest, + responses((status = 200, description = "Success", body = agentic_core::types::file_search::VectorStoreFileObject), (status = 400, description = "Invalid request", body = crate::openapi::ApiErrorResponse), (status = 404, description = "Object not found", body = crate::openapi::ApiErrorResponse)), + security(("bearer_auth" = [])), tag = "file_search", +))] +pub(crate) async fn update_vector_store_file( + State(state): State, + Path((store_id, file_id)): Path<(String, String)>, + request: Result, JsonRejection>, +) -> Response { + let search = match service(&state) { + Ok(service) => service, + Err(error) => return *error, + }; + let request = match body(request) { + Ok(request) => request, + Err(error) => return *error, + }; + result(search.update_vector_store_file(&store_id, &file_id, request).await) +} + +#[cfg_attr(feature = "openapi", utoipa::path( + get, path = "/v1/vector_stores/{store_id}/files/{file_id}/content", + params(("store_id" = String, Path, description = "Object identifier"), ("file_id" = String, Path, description = "Object identifier")), + responses((status = 200, description = "Success", body = agentic_core::types::file_search::VectorStoreFileContentPage), (status = 400, description = "Invalid request", body = crate::openapi::ApiErrorResponse), (status = 404, description = "Object not found", body = crate::openapi::ApiErrorResponse)), + security(("bearer_auth" = [])), tag = "file_search", +))] +pub(crate) async fn vector_store_file_content( + State(state): State, + Path((store_id, file_id)): Path<(String, String)>, +) -> Response { + let search = match service(&state) { + Ok(service) => service, + Err(error) => return *error, + }; + result(search.vector_store_file_content(&store_id, &file_id).await) +} diff --git a/crates/agentic-server/src/openapi.rs b/crates/agentic-server/src/openapi.rs index 65bbfc0c..6e06f289 100644 --- a/crates/agentic-server/src/openapi.rs +++ b/crates/agentic-server/src/openapi.rs @@ -19,6 +19,9 @@ use utoipa::OpenApi; crate::handler::http::file_search::create_vector_store, crate::handler::http::file_search::list_vector_stores, crate::handler::http::file_search::get_vector_store, + crate::handler::http::file_search::update_vector_store, + crate::handler::http::file_search::update_vector_store_file, + crate::handler::http::file_search::vector_store_file_content, crate::handler::http::file_search::delete_vector_store, crate::handler::http::file_search::attach_file, crate::handler::http::file_search::list_vector_store_files, diff --git a/crates/agentic-server/tests/file_search_http_test.rs b/crates/agentic-server/tests/file_search_http_test.rs index 0ffd9681..d5b82633 100644 --- a/crates/agentic-server/tests/file_search_http_test.rs +++ b/crates/agentic-server/tests/file_search_http_test.rs @@ -82,7 +82,12 @@ async fn detaching_or_deleting_a_vector_store_preserves_the_uploaded_file() { store ); let attachment_url = format!("{store_url}/files/{file_id}"); - assert_eq!(api_json(client.get(&attachment_url)).await["id"], file_id); + let attachment = api_json(client.get(&attachment_url)).await; + assert_eq!(attachment["id"], file_id); + assert_eq!( + attachment["chunking_strategy"], + json!({"type":"static","static":{"max_chunk_size_tokens":800,"chunk_overlap_tokens":400}}) + ); assert_eq!( api_json(client.get(format!("{store_url}/files"))).await["data"][0]["id"], file_id @@ -526,3 +531,88 @@ async fn oversized_multipart_part_headers_are_rejected_early() { ) .await; } + +#[tokio::test] +async fn store_nullable_updates_and_expiration_contract() { + let server = gateway().await; + let client = reqwest::Client::new(); + let store = api_json(client.post(format!("{}/v1/vector_stores", server.url)).json(&json!({ + "name":"original", "description":"A collection", "metadata":null, + "expires_after":{"anchor":"last_active_at","days":1} + }))) + .await; + assert_eq!( + store["expires_at"].as_i64(), + Some(store["last_active_at"].as_i64().unwrap() + 86400) + ); + let url = format!("{}/v1/vector_stores/{}", server.url, store["id"].as_str().unwrap()); + let updated = api_json( + client + .post(&url) + .json(&json!({"name":null,"metadata":{"project":"test"},"expires_after":null})), + ) + .await; + assert_eq!(updated["name"], ""); + assert!(updated["expires_after"].is_null()); + assert!(updated["expires_at"].is_null()); + let retained = api_json(client.post(&url).json(&json!({}))).await; + assert_eq!(retained["metadata"], json!({"project":"test"})); + let cleared = api_json(client.post(&url).json(&json!({"metadata":null}))).await; + assert!(cleared["metadata"].is_null()); + for days in [0, 366] { + assert_eq!( + client + .post(&url) + .json(&json!({"expires_after":{"anchor":"last_active_at","days":days}})) + .send() + .await + .unwrap() + .status(), + StatusCode::BAD_REQUEST + ); + } +} + +#[tokio::test] +async fn attachment_nullable_attributes_original_content_and_status_filter() { + let server = gateway().await; + let client = reqwest::Client::new(); + let text = "Original lunar policy.\n".repeat(100); + let file = api_json( + client + .post(format!("{}/v1/files", server.url)) + .header("content-type", "multipart/form-data; boundary=upload") + .body(multipart("original.txt", &text)), + ) + .await; + let store = api_json(client.post(format!("{}/v1/vector_stores", server.url)).json(&json!({}))).await; + let url = format!( + "{}/v1/vector_stores/{}/files", + server.url, + store["id"].as_str().unwrap() + ); + let file_id = file["id"].as_str().unwrap(); + api_json(client.post(&url).json(&json!({"file_id":file_id,"attributes":null,"chunking_strategy":{"type":"static","static":{"max_chunk_size_tokens":100,"chunk_overlap_tokens":50}}}))).await; + let attachment = format!("{url}/{file_id}"); + assert_eq!( + client.post(&attachment).json(&json!({})).send().await.unwrap().status(), + StatusCode::BAD_REQUEST + ); + let updated = api_json(client.post(&attachment).json(&json!({"attributes":{"kind":"lunar"}}))).await; + assert_eq!(updated["attributes"]["kind"], "lunar"); + let cleared = api_json(client.post(&attachment).json(&json!({"attributes":null}))).await; + assert_eq!(cleared["attributes"], json!({})); + let content = api_json(client.get(format!("{attachment}/content"))).await; + assert_eq!( + content, + json!({"object":"vector_store.file_content.page","data":[{"type":"text","text":text}],"has_more":false,"next_page":null}) + ); + assert_eq!( + api_json(client.get(format!("{url}?filter=completed&limit=1"))).await["data"][0]["id"], + file_id + ); + assert_eq!( + api_json(client.get(format!("{url}?filter=failed&limit=1"))).await["data"], + json!([]) + ); +} diff --git a/crates/agentic-server/tests/openapi_test.rs b/crates/agentic-server/tests/openapi_test.rs index dca17c62..55f101c0 100644 --- a/crates/agentic-server/tests/openapi_test.rs +++ b/crates/agentic-server/tests/openapi_test.rs @@ -217,3 +217,23 @@ async fn swagger_ui_returns_html() { "swagger-ui should return HTML, got: {content_type}" ); } + +#[tokio::test] +async fn vector_store_schema_preserves_required_nullable_contracts() { + let spec = fetch_spec().await; + let schemas = &spec["components"]["schemas"]; + let required = schemas["VectorStoreObject"]["required"].as_array().unwrap(); + assert!(required.iter().any(|field| field == "last_active_at")); + assert!(required.iter().any(|field| field == "metadata")); + assert_eq!( + schemas["UpdateVectorStoreFileRequest"]["required"], + serde_json::json!(["attributes"]) + ); + assert!(schemas["UpdateVectorStoreRequest"]["required"].is_null()); + let chunking = schemas["VectorStoreFileChunkingStrategy"]["oneOf"].as_array().unwrap(); + let kinds: Vec<_> = chunking + .iter() + .filter_map(|variant| variant["properties"]["type"]["enum"][0].as_str()) + .collect(); + assert_eq!(kinds, ["static", "other"]); +} diff --git a/docs/api/file-search.md b/docs/api/file-search.md index 964ab370..c1278a07 100644 --- a/docs/api/file-search.md +++ b/docs/api/file-search.md @@ -397,9 +397,10 @@ parts, output items, and the terminal response. | Retrieve/delete file metadata | `GET` / `DELETE /v1/files/{file_id}` | | Download original bytes | `GET /v1/files/{file_id}/content` | | Create/list vector stores | `POST` / `GET /v1/vector_stores` | -| Retrieve/delete a vector store | `GET` / `DELETE /v1/vector_stores/{store_id}` | +| Retrieve/update/delete a vector store | `GET` / `POST` / `DELETE /v1/vector_stores/{store_id}` | | Attach/list files in a store | `POST` / `GET /v1/vector_stores/{store_id}/files` | -| Retrieve/detach a store file | `GET` / `DELETE /v1/vector_stores/{store_id}/files/{file_id}` | +| Retrieve/update/detach a store file | `GET` / `POST` / `DELETE /v1/vector_stores/{store_id}/files/{file_id}` | +| Retrieve parsed original text | `GET /v1/vector_stores/{store_id}/files/{file_id}/content` | | Search a store | `POST /v1/vector_stores/{store_id}/search` | Lists accept `limit`, `after`, `before`, and `order`. Files lists additionally accept @@ -409,6 +410,76 @@ original upload. Deleting an upload removes its metadata, attachments, and chunk from all stores, then removes its local file bytes. Deleting a vector store preserves uploaded files. + +### Store updates, activity, and expiration + +Create accepts optional `name`, `description` (up to 512 bytes), nullable `metadata`, +and `expires_after: {"anchor": "last_active_at", "days": 1}`. Days must be 1–365. +Store responses include nullable `last_active_at`, `expires_after`, and `expires_at`. +`POST /v1/vector_stores/{store_id}` updates `name`, `metadata`, and `expires_after`: +omitted fields are preserved; null clears them. Clearing `name` produces an empty +string; clearing metadata produces null; clearing the policy removes the deadline. + +In this implementation, creation initializes activity using the database clock. +Successful attachment publication and successful search completion refresh +`last_active_at` and recompute the policy deadline. This applies to both the search +endpoint and the Responses built-in tool, including searches with no matches. +Reads, lists, attribute changes, and store metadata/policy updates do not refresh +activity. Setting a policy on an idle permanent store uses its previous activity +and can expire it immediately. This refresh-event policy describes this server; +it is not a claim about undocumented hosted-service behavior. + +An expired store remains retrievable and listable with `status: "expired"`, zero +visible file counts, and zero usage bytes. Its attachments, parsed content, and +search are unavailable immediately, before any cleanup. Updates and late model +completions cannot revive it. Explicit store deletion still works. Search checks +all selected stores after model calls; if any expired or was deleted, it returns +not-found. A successful search linearizes activity at this final database check; +a subsequent deletion or expiration cannot retract an already returned response. +Publication locks the source upload before the store, then checks fresh database +time after both locks. Policy updates and cleanup share the same store guard. + +`FileSearchService::cleanup_expired_vector_stores(limit)` processes 1–1000 due +stores per call and returns the number expired. It rechecks each deadline after +acquiring its store lock, commits expired status and removal of attachments/chunks +atomically, and preserves uploaded files, including those used by other stores. +Repeated cleanup is safe after a restart. This layer exposes explicit cleanup; +it does not start a worker from service construction or cloning. + +### Attachment updates and parsed content + +Attachment creation accepts null `attributes`, equivalent to an empty map. +`POST /v1/vector_stores/{store_id}/files/{file_id}` requires the `attributes` member; +an object replaces the attributes and null clears them. Updates also change the +attributes used by exact and indexed search in the same transaction. +Attachment responses report `chunking_strategy` separately from request options: +new auto, static, and contextual ingestion report `type: "static"` with the actual +resolved chunk size and overlap. Legacy auto/contextual records report +`type: "other"`; their original request settings did not persist resolved boundaries. +Legacy static settings remain static. Contextual ingestion retains its extension +semantics, including its default 700-token size and 400-token overlap; reporting +those boundaries does not make that overlap valid for an OpenAI-style static request. +Store-file lists accept `filter=in_progress|completed|cancelled|failed`. Filtering +happens before keyset pagination and only completed attachments are searchable. +`filter` is rejected on Files and vector store lists. + +The content endpoint returns one bounded page: + +```json +{ + "object": "vector_store.file_content.page", + "data": [{"type": "text", "text": "Original extracted document text"}], + "has_more": false, + "next_page": null +} +``` + +Parsed text is saved during ingestion, before overlapping chunks or contextual +hints are generated. It is bounded by the existing 16 MiB extracted-document limit. +Legacy attachments reparse their original uploaded bytes without model calls. +This endpoint returns extracted text; the Files content endpoint returns the +original bytes, including a PDF's binary representation. + Uploads publish complete, synced files before committing metadata. Failures and cancellation before commit clean up the upload. Filesystem and SQL commits are separate: a process crash or uncertain upload commit can leave unreferenced files. diff --git a/docs/superpowers/plans/2026-09-10-file-search-parity.md b/docs/superpowers/plans/2026-09-10-file-search-parity.md index c153529a..bb3826da 100644 --- a/docs/superpowers/plans/2026-09-10-file-search-parity.md +++ b/docs/superpowers/plans/2026-09-10-file-search-parity.md @@ -57,23 +57,37 @@ - [ ] Implement nullable/optional expiration timestamps and batch-purpose default expiration. Expired files must disappear from read/list/search and their attachments must be removed; expose cleanup for the lifecycle worker without deleting unrelated uploads. - [ ] Test real multipart and download bytes, disconnect/error cleanup, expiration visibility/deletion, legacy reads, and boundary errors on SQLite and PostgreSQL. Run covering service/HTTP tests, changed-target clippy, formatting, and pre-commit; commit signed off and review. -## Task 4: Vector Stores lifecycle and durable file batches +## Task 4: Vector Stores contracts and expiration **Branch:** `codex/vector-store-lifecycle`, based on the Files API branch. -**Files:** vector store/file/batch types, storage lifecycle modules/migrations, service lifecycle modules, server vector-store handlers and startup/shutdown, OpenAPI schemas, HTTP/service/model-fixture tests, compatibility documentation. +**Files:** store/file update and content types, storage/service lifecycle modules and migration, server Vector Stores/file handlers, OpenAPI schemas, HTTP/service/model tests, compatibility documentation. -**Interfaces:** Extend existing `FileSearchService` with update/expiration/content/batch operations. Workers have explicit start/shutdown ownership. Batch ingestion uses Tasks 1–2's atomic vector/model pipeline and Task 3's file visibility rules. Store expiration removes search data without deleting independent uploaded files. +**Interfaces:** Extend FileSearchService with typed store updates, expiration policies/status, attachment attributes/status filtering, and parsed original content. Expose bounded explicit expiration cleanup for the next layer's runtime. Preserve existing atomic ingestion and immediate visibility rules. -- [ ] Use official Vector Stores/file/batch schemas and the SDK contract cases. Add red tests for store and attribute updates, nullable metadata/expiry, attachment status filtering, parsed content, and per-file batch options/pagination. -- [ ] Implement typed updates, expiration policies/timestamps and parsed content. Preserve omitted-versus-null semantics. Expired stores must stop search and attachment reads and clean associated state. -- [ ] Implement batch create/retrieve/cancel/list-files with durable portable state, bounded workers, per-file status/counts, restart recovery, and explicit shutdown. Prevent cancelled/replaced workers from publishing stale results; support multiple server instances safely. -- [ ] Test partial failure, cancellation during an active model call, restart recovery, expiration, pagination and literal response fields. Include real PostgreSQL lifecycle tests and SQLite tests, not only internal serialization tests. -- [ ] Run covering tests, changed-target clippy, formatting, and pre-commit; commit signed off and review. +- [ ] Add red tests for nullable store/attribute updates, expiry, filtered pagination and parsed original content, using official schemas and strict SDK cases. +- [ ] Implement typed omitted/null/value updates, store expiration/activity, attachment attribute updates and content pages. Expired stores stop serving search data while preserving uploaded files and expired metadata. +- [ ] Serialize policy updates, expiration cleanup and publication on SQLite/PostgreSQL; recheck visibility after slow model calls and prevent expired stores from reviving. +- [ ] Test expiration during model calls, policy update versus cleanup, original content without duplicated overlap/context, filtered keyset pagination and independent upload preservation on SQLite and real PostgreSQL. +- [ ] Run covering service/HTTP/SDK tests, changed-target clippy, formatting and pre-commit; commit signed off and review. -## Task 5: Responses citation events and SDK conformance verification +## Task 5: Durable Vector Store file batches and worker runtime -**Branch:** `codex/file-search-openai-api`, based on the lifecycle branch. +**Branch:** `codex/vector-store-batches`, based on the Vector Stores lifecycle branch. + +**Files:** typed batch/job contracts, portable durable storage migration/modules, service publication integration and worker runtime, server batch handlers/startup/shutdown, OpenAPI schemas, lifecycle/runtime/HTTP/model tests, documentation. + +**Interfaces:** Consume Tasks3–4's file/store visibility and cleanup. Reuse the existing prepared attachment and atomic publication path. Runtime ownership stays separate from the cloneable request service. + +- [ ] Add red batch/SDK tests for create/retrieve/cancel/list-files, per-file options/counts, in-progress visibility, partial failure and pagination. +- [ ] Implement durable membership/jobs, bounded admission, claims/leases/generation fencing and restart recovery. Prevent stale publication after cancellation, claim loss, detach/re-attach, deletion or expiry across multiple server instances. +- [ ] Wire explicit worker start/shutdown into every server exit path. Shutdown joins owned work and preserves resumability; it does not cancel the API batch. Run prior layers' bounded expiration and durable blob cleanup from this runtime. +- [ ] Test competing runtimes, lease expiry, blocked-model cancellation, restart, parent mutation, counts/pagination, cleanup replay and shutdown on SQLite and real PostgreSQL. +- [ ] Run covering service/HTTP/SDK tests, changed-target clippy, formatting and pre-commit; commit signed off and review. + +## Task 6: Responses citation events and SDK conformance verification + +**Branch:** `codex/file-search-openai-api`, based on the durable batches branch. **Files:** typed Responses annotation events, executor citation event handling, HTTP/tool tests, SDK contract test/script and CI invocation, OpenAPI schemas, compatibility documentation. diff --git a/docs/superpowers/specs/2026-09-10-file-search-parity.md b/docs/superpowers/specs/2026-09-10-file-search-parity.md index a8c9d891..263010d2 100644 --- a/docs/superpowers/specs/2026-09-10-file-search-parity.md +++ b/docs/superpowers/specs/2026-09-10-file-search-parity.md @@ -16,8 +16,10 @@ the missing OGX capabilities and OpenAI API operations as stacked PRs. 4. `codex/files-api`, based on the retrieval branch: streaming Files API, documented upload purposes, expiration and list/delete contracts. 5. `codex/vector-store-lifecycle`, based on the Files API branch: store/file - updates, expiration, asynchronous durable batches and parsed content. -6. `codex/file-search-openai-api`, based on the lifecycle branch: Responses + updates, expiration and parsed content. +6. `codex/vector-store-batches`, based on the lifecycle branch: asynchronous + durable batches, restart recovery, cancellation and explicit worker shutdown. +7. `codex/file-search-openai-api`, based on the batches branch: Responses annotation events, SDK contract verification and compatibility documentation. Each layer must build and pass its relevant tests independently. Open PRs only