diff --git a/.env.example b/.env.example index 64c31d3..d9704d9 100644 --- a/.env.example +++ b/.env.example @@ -52,6 +52,10 @@ AGENTIC_CHAT_MAX_ITERATIONS=6 # OPENROUTER_EMBEDDING_MODEL=openai/text-embedding-3-small # OPENROUTER_HTTP_REFERER=https://your-site.example # OPENROUTER_APP_TITLE=ImageApi +# Cap on generated tokens per call (sent as max_tokens). OpenRouter reserves +# credit against the requested max, so leaving it uncapped makes pricey +# models fail with credit-limit errors. Default 16384; 0 = no cap. +# OPENROUTER_MAX_TOKENS=16384 # ── AI Insights — local backend switch ────────────────────────────────── # Picks which local LLM stack the server uses for chat, vision describe, diff --git a/CLAUDE.md b/CLAUDE.md index 4faec1c..bd06509 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -636,6 +636,9 @@ OPENROUTER_BASE_URL=https://openrouter.ai/api/v1 # Override base URL (option OPENROUTER_EMBEDDING_MODEL=openai/text-embedding-3-small # Optional, embeddings stay local today OPENROUTER_HTTP_REFERER=https://your-site.example # Optional attribution header OPENROUTER_APP_TITLE=ImageApi # Optional attribution header +OPENROUTER_MAX_TOKENS=16384 # Per-call max_tokens cap (default 16384; 0 = uncapped). + # OpenRouter reserves credit against this, so an uncapped + # request to a pricey model can fail on credit limits. # Local LLM backend switch. `ollama` (default) keeps the OLLAMA_* settings # above; `llamacpp` swaps the entire local stack (chat + vision describe + 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/ai/openrouter.rs b/src/ai/openrouter.rs index 62a3cca..5ec420f 100644 --- a/src/ai/openrouter.rs +++ b/src/ai/openrouter.rs @@ -19,6 +19,11 @@ use futures::stream::{BoxStream, StreamExt}; const DEFAULT_BASE_URL: &str = "https://openrouter.ai/api/v1"; const DEFAULT_EMBEDDING_MODEL: &str = "openai/text-embedding-3-small"; const CACHE_DURATION_SECS: u64 = 15 * 60; +/// Default completion cap sent as `max_tokens`. OpenRouter pre-authorizes +/// credit against the requested max (defaulting to the model's full output +/// window when omitted), so expensive models fail with credit-limit errors +/// even when the actual reply is short. Override via `OPENROUTER_MAX_TOKENS`. +pub const DEFAULT_MAX_TOKENS: u32 = 16384; #[derive(Clone)] struct CachedEntry { @@ -64,6 +69,9 @@ pub struct OpenRouterClient { top_p: Option, top_k: Option, min_p: Option, + /// Upper bound on generated tokens per call, sent as `max_tokens`. + /// `None` omits the field (provider default). + max_tokens: Option, /// Optional `HTTP-Referer` header OpenRouter uses for attribution. pub referer: Option, /// Optional `X-Title` header OpenRouter uses for attribution. @@ -87,6 +95,7 @@ impl OpenRouterClient { top_p: None, top_k: None, min_p: None, + max_tokens: Some(DEFAULT_MAX_TOKENS), referer: None, app_title: None, } @@ -115,6 +124,10 @@ impl OpenRouterClient { self.min_p = min_p; } + pub fn set_max_tokens(&mut self, max_tokens: Option) { + self.max_tokens = max_tokens; + } + pub fn set_attribution(&mut self, referer: Option, app_title: Option) { self.referer = referer; self.app_title = app_title; @@ -288,6 +301,9 @@ impl OpenRouterClient { // directly transferable. Skip rather than silently mis-map. let _ = c; } + if let Some(m) = self.max_tokens { + v.push(("max_tokens", json!(m))); + } v } } 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/library_maintenance.rs b/src/library_maintenance.rs index ffa87bb..a4f44b9 100644 --- a/src/library_maintenance.rs +++ b/src/library_maintenance.rs @@ -156,6 +156,25 @@ pub fn detect_missing_files_for_library( (deleted, next_offset) } +/// Re-gather query-planner statistics (`sqlite_stat1`). +/// +/// The migrations run `ANALYZE` once, but the stats never refresh after +/// that. A snapshot taken while `image_exif.content_hash` was all-NULL +/// told the planner every hash matches every row, so the correlated +/// `content_hash = ...` subqueries in [`refresh_back_refs`] scanned +/// the index instead of seeking it: ~20s per tick holding the write +/// lock, starving chat persistence past `busy_timeout`. With fresh +/// stats the same pass takes ~0.1s. `PRAGMA optimize` doesn't catch +/// this on SQLite 3.45 (verified), hence a plain `ANALYZE`. It costs +/// well under a second at current scale, so run it on full-scan ticks. +pub fn refresh_planner_stats(conn: &mut SqliteConnection) { + let started = std::time::Instant::now(); + match sql_query("ANALYZE").execute(conn) { + Ok(_) => debug!("ANALYZE completed in {:?}", started.elapsed()), + Err(e) => warn!("ANALYZE failed: {:?}", e), + } +} + /// Refresh the `(library_id, rel_path)` back-refs on hash-keyed /// tables. A back-ref is stale when: /// - its `content_hash` is non-null, diff --git a/src/state.rs b/src/state.rs index 3f456e7..0d09db2 100644 --- a/src/state.rs +++ b/src/state.rs @@ -402,6 +402,13 @@ fn build_openrouter_from_env() -> Option> { if let Ok(model) = env::var("OPENROUTER_EMBEDDING_MODEL") { client.set_embedding_model(model); } + // OPENROUTER_MAX_TOKENS=0 omits the cap (provider default). + if let Some(n) = env::var("OPENROUTER_MAX_TOKENS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + { + client.set_max_tokens((n > 0).then_some(n)); + } Some(Arc::new(client)) } diff --git a/src/utils.rs b/src/utils.rs index d76ad56..6439d2a 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -27,7 +27,7 @@ where Err(e) => e, }; for attempt in 1..=max_retries { - let base = Duration::from_millis(100).saturating_mul(1_u32.pow(attempt - 1)); + let base = Duration::from_millis(100).saturating_mul(2_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; @@ -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(2_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 /// diff --git a/src/watcher.rs b/src/watcher.rs index 39127ca..6d2a727 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -577,6 +577,9 @@ pub fn watch_files( // model" for the rules. { let mut conn = image_api::database::connect(); + if is_full_scan { + library_maintenance::refresh_planner_stats(&mut conn); + } let _ = image_api::database::reconcile::run(&mut conn); // Back-ref refresh: hash-keyed rows whose