diff --git a/src/ai/insight_generator.rs b/src/ai/insight_generator.rs index e396ec9..be2c246 100644 --- a/src/ai/insight_generator.rs +++ b/src/ai/insight_generator.rs @@ -19,20 +19,32 @@ use crate::ai::sms_client::{SmsApiClient, SmsSearchHit, SmsSearchParams}; use crate::ai::user_display_name; use crate::database::models::InsertPhotoInsight; use crate::database::{ - CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, KnowledgeDao, LocationHistoryDao, - SearchHistoryDao, + CalendarEventDao, DailySummaryDao, DbError, ExifDao, InsightDao, KnowledgeDao, + LocationHistoryDao, SearchHistoryDao, }; use crate::libraries::Library; use crate::memories::extract_date_from_filename; use crate::otel::global_tracer; use crate::tags::TagDao; -use crate::utils::{earliest_fs_time, normalize_path, retry_with_backoff}; +use crate::utils::{earliest_fs_time, normalize_path, retry_with_backoff, retry_with_backoff_if}; /// Max location records rendered by `tool_get_location_history`. The DAO /// query is range-bounded, not limited, so the tool caps the rendered list /// and labels the truncation via `found_header`. const LOCATION_HISTORY_DISPLAY_LIMIT: usize = 20; +/// SQLite transient-lock detector for knowledge-tool writes. WAL + +/// busy_timeout absorb most reader/writer contention, but a parallel +/// insight job or the face-watch write storm can still surface SQLITE_BUSY +/// ("database is locked" / "database is busy"). Those errors are transient +/// and worth a backoff retry; everything else surfaces immediately. +fn is_sqlite_busy(e: &DbError) -> bool { + match &e.source { + Some(s) => s.contains("database is locked") || s.contains("database is busy"), + None => false, + } +} + /// Strip common markdown decoration (bold/italic markers, heading hashes, /// backticks, quotes) from both ends of a model-emitted title. Models wrap /// the line despite the prompt: `**Title: A Day in the Woods**`, @@ -3004,11 +3016,13 @@ Return ONLY the summary, nothing else."#, updated_at: now, }; - let mut kdao = self - .knowledge_dao - .lock() - .expect("Unable to lock KnowledgeDao"); - match kdao.upsert_entity(cx, insert) { + match retry_with_backoff_if("tool_store_entity/upsert_entity", 4, is_sqlite_busy, || { + let mut kdao = self + .knowledge_dao + .lock() + .expect("Unable to lock KnowledgeDao"); + kdao.upsert_entity(cx, insert.clone()) + }) { Ok(entity) => { let mut response = format!( "Entity stored: ID:{} | {} | {} | confidence:{:.2}", @@ -3116,16 +3130,17 @@ Return ONLY the summary, nothing else."#, last_modified_at: None, }; - let mut kdao = self - .knowledge_dao - .lock() - .expect("Unable to lock KnowledgeDao"); - - // Upsert the fact (corroboration bumps confidence if duplicate) - let (stored_fact, is_new) = match kdao.upsert_fact(cx, fact) { - Ok(r) => r, - Err(e) => return format!("Error storing fact: {:?}", e), - }; + let (stored_fact, is_new) = + match retry_with_backoff_if("tool_store_fact/upsert_fact", 4, is_sqlite_busy, || { + let mut kdao = self + .knowledge_dao + .lock() + .expect("Unable to lock KnowledgeDao"); + kdao.upsert_fact(cx, fact.clone()) + }) { + Ok((f, is_new)) => (f, is_new), + Err(e) => return format!("Error storing fact: {:?}", e), + }; // Upsert a photo link so this entity is associated with this photo let link = InsertEntityPhotoLink { @@ -3134,7 +3149,20 @@ Return ONLY the summary, nothing else."#, file_path: file_path.to_string(), role: photo_role, }; - if let Err(e) = kdao.upsert_photo_link(cx, link) { + // Best-effort; a failed link shouldn't fail the fact write, but + // give SQLite contention the same retry treatment. + if let Err(e) = retry_with_backoff_if( + "tool_store_fact/upsert_photo_link", + 4, + is_sqlite_busy, + || { + let mut kdao = self + .knowledge_dao + .lock() + .expect("Unable to lock KnowledgeDao"); + kdao.upsert_photo_link(cx, link.clone()) + }, + ) { log::warn!( "Failed to upsert photo link for entity {}: {:?}", subject_entity_id, @@ -3220,12 +3248,13 @@ Return ONLY the summary, nothing else."#, log::info!("tool_update_fact: fact_id={}", fact_id); - let mut kdao = self - .knowledge_dao - .lock() - .expect("Unable to lock KnowledgeDao"); - - match kdao.update_fact(cx, fact_id, patch, Some((model, backend))) { + match retry_with_backoff_if("tool_update_fact/update_fact", 4, is_sqlite_busy, || { + let mut kdao = self + .knowledge_dao + .lock() + .expect("Unable to lock KnowledgeDao"); + kdao.update_fact(cx, fact_id, patch.clone(), Some((model, backend))) + }) { Ok(Some(f)) => format!( "Updated fact ID:{} (status={}, confidence={:.2})", f.id, f.status, f.confidence @@ -3276,12 +3305,18 @@ Return ONLY the summary, nothing else."#, new_fact_id ); - let mut kdao = self - .knowledge_dao - .lock() - .expect("Unable to lock KnowledgeDao"); - - match kdao.supersede_fact(cx, old_fact_id, new_fact_id, Some((model, backend))) { + match retry_with_backoff_if( + "tool_supersede_fact/supersede_fact", + 4, + is_sqlite_busy, + || { + let mut kdao = self + .knowledge_dao + .lock() + .expect("Unable to lock KnowledgeDao"); + kdao.supersede_fact(cx, old_fact_id, new_fact_id, Some((model, backend))) + }, + ) { Ok(Some(f)) => format!( "Superseded fact ID:{} (now status={}, valid_until={:?})", f.id, f.status, f.valid_until @@ -4396,17 +4431,54 @@ Return ONLY the summary, nothing else."#, iterations_used ); messages.push(ChatMessage::user(format!( - "Based on the context gathered, please write the final photo insight. Start with \"Title: \" on the first line (max 8 words), then a blank line, then the detailed personal summary. Write in first person as {}.", + "You may call more tools if you still need information. Otherwise, write the final photo insight now. Start with \"Title: \" on the first line (max 8 words), then a blank line, then the detailed personal summary. Write in first person as {}.", user_display_name() ))); let (final_response, prompt_tokens, eval_tokens) = backend .chat() - .chat_with_tools(messages.clone(), vec![]) + .chat_with_tools(messages.clone(), tools.clone()) .await?; last_prompt_eval_count = prompt_tokens; last_eval_count = eval_tokens; - final_content = final_response.content.clone(); - messages.push(final_response); + + // The nudge re-offers tools; if the model takes one more call, + // collect its result, then force a tool-less turn so we never + // store an empty insight. + if let Some(ref tcs) = final_response.tool_calls + && !tcs.is_empty() + { + log::info!( + "Loop-exhaustion nudge took {} more tool call(s); collecting results before the final answer", + tcs.len() + ); + messages.push(final_response.clone()); + for tool_call in tcs { + let result = self + .execute_tool( + &tool_call.function.name, + &tool_call.function.arguments, + &backend, + &image_base64, + &file_path, + user_id, + &persona_id, + &loop_cx, + ) + .await; + messages.push(ChatMessage::tool_result(result)); + } + let (resp2, pt2, et2) = backend + .chat() + .chat_with_tools(messages.clone(), vec![]) + .await?; + last_prompt_eval_count = pt2; + last_eval_count = et2; + final_content = resp2.content.clone(); + messages.push(resp2); + } else { + final_content = final_response.content.clone(); + messages.push(final_response); + } } loop_cx @@ -4604,13 +4676,41 @@ Return ONLY the summary, nothing else."#, max_iter ); messages.push(ChatMessage::user( - "Based on the context gathered, please write the final answer. Return ONLY the JSON object, no prose or code fences.", + "You may call more tools if you still need information. Otherwise, write the final answer now. Return ONLY the JSON object, no prose or code fences.", )); let (final_response, _, _) = backend .chat() - .chat_with_tools(messages.clone(), vec![]) + .chat_with_tools(messages.clone(), tools.clone()) .await?; - final_content = final_response.content; + // Same bounded tool round as the insight loop: execute any last + // tool call, then force a tool-less turn for the actual answer. + if let Some(ref tcs) = final_response.tool_calls + && !tcs.is_empty() + { + messages.push(final_response.clone()); + for tool_call in tcs { + let result = self + .execute_tool( + &tool_call.function.name, + &tool_call.function.arguments, + backend, + &None, + "", + 0, + "", + &opentelemetry::Context::new(), + ) + .await; + messages.push(ChatMessage::tool_result(result)); + } + let (resp2, _, _) = backend + .chat() + .chat_with_tools(messages.clone(), vec![]) + .await?; + final_content = resp2.content; + } else { + final_content = final_response.content; + } } Ok(final_content) diff --git a/src/database/knowledge_dao.rs b/src/database/knowledge_dao.rs index 9eddb26..0951b50 100644 --- a/src/database/knowledge_dao.rs +++ b/src/database/knowledge_dao.rs @@ -96,6 +96,7 @@ pub struct EntityPatch { pub confidence: Option, } +#[derive(Clone)] pub struct FactPatch { pub predicate: Option, pub object_value: Option, diff --git a/src/database/models.rs b/src/database/models.rs index 300def1..2102511 100644 --- a/src/database/models.rs +++ b/src/database/models.rs @@ -294,7 +294,7 @@ pub struct InsertLibrary<'a> { // --- Knowledge memory models --- -#[derive(Insertable)] +#[derive(Insertable, Clone)] #[diesel(table_name = entities)] pub struct InsertEntity { pub name: String, @@ -320,7 +320,7 @@ pub struct Entity { pub updated_at: i64, } -#[derive(Insertable)] +#[derive(Insertable, Clone)] #[diesel(table_name = entity_facts)] pub struct InsertEntityFact { pub subject_entity_id: i32, @@ -396,7 +396,7 @@ pub struct EntityFact { pub last_modified_at: Option, } -#[derive(Insertable)] +#[derive(Insertable, Clone)] #[diesel(table_name = entity_photo_links)] pub struct InsertEntityPhotoLink { pub entity_id: i32, diff --git a/src/utils.rs b/src/utils.rs index d76ad56..4e8ba3f 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -54,6 +54,66 @@ where Err(last_err) } +/// Retry a fallible operation with exponential backoff + jitter, but only +/// while `retry_if` classifies the error as transient. A deterministic +/// failure (validation, NotFound, ...) short-circuits immediately instead +/// of burning retries on an error that will repeat. Same backoff schedule +/// as [`retry_with_backoff`]. +/// +/// ``` +/// use image_api::utils::retry_with_backoff_if; +/// +/// fn is_transient(e: &DbError) -> bool { /* ... */ } +/// +/// let result = retry_with_backoff_if("some-write", 4, is_transient, || { +/// Ok::<_, anyhow::Error>(42) +/// }); +/// ``` +pub fn retry_with_backoff_if bool>( + label: &str, + max_retries: u32, + retry_if: U, + mut op: F, +) -> Result +where + F: FnMut() -> Result, + E: std::fmt::Debug, +{ + let mut last_err = match op() { + Ok(v) => return Ok(v), + Err(e) => e, + }; + for attempt in 1..=max_retries { + if !retry_if(&last_err) { + log::error!( + "{}: error is not transient, not retrying: {:?}", + label, + last_err + ); + break; + } + let base = Duration::from_millis(100).saturating_mul(1_u32.pow(attempt - 1)); + let jitter_range = base.as_millis() as f64 * 0.25; + let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range) as u128; + let delay = base.as_millis() as i128 + jitter as i128; + std::thread::sleep(Duration::from_millis(delay.max(0) as u64)); + log::debug!( + "{}: attempt {}/{} failed ({:?}), retrying in {}ms", + label, + attempt, + max_retries, + last_err, + delay.max(0) + ); + match op() { + Ok(v) => return Ok(v), + Err(e) => last_err = e, + } + } + log::error!("{}: retries exhausted or stopped: {:?}", label, last_err); + Err(last_err) +} + /// Normalize a file path to use forward slashes for cross-platform consistency /// This ensures paths stored in the database always use `/` regardless of OS ///