Feature/agentic loop fixes #111

Merged
cameron merged 3 commits from feature/agentic-loop-fixes into master 2026-09-24 15:59:59 +00:00
10 changed files with 255 additions and 42 deletions

No files matched your search

+4
View File
@@ -52,6 +52,10 @@ AGENTIC_CHAT_MAX_ITERATIONS=6
# OPENROUTER_EMBEDDING_MODEL=openai/text-embedding-3-small # OPENROUTER_EMBEDDING_MODEL=openai/text-embedding-3-small
# OPENROUTER_HTTP_REFERER=https://your-site.example # OPENROUTER_HTTP_REFERER=https://your-site.example
# OPENROUTER_APP_TITLE=ImageApi # 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 ────────────────────────────────── # ── AI Insights — local backend switch ──────────────────────────────────
# Picks which local LLM stack the server uses for chat, vision describe, # Picks which local LLM stack the server uses for chat, vision describe,
+3
View File
@@ -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_EMBEDDING_MODEL=openai/text-embedding-3-small # Optional, embeddings stay local today
OPENROUTER_HTTP_REFERER=https://your-site.example # Optional attribution header OPENROUTER_HTTP_REFERER=https://your-site.example # Optional attribution header
OPENROUTER_APP_TITLE=ImageApi # 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 # Local LLM backend switch. `ollama` (default) keeps the OLLAMA_* settings
# above; `llamacpp` swaps the entire local stack (chat + vision describe + # above; `llamacpp` swaps the entire local stack (chat + vision describe +
+138 -38
View File
@@ -19,20 +19,32 @@ use crate::ai::sms_client::{SmsApiClient, SmsSearchHit, SmsSearchParams};
use crate::ai::user_display_name; use crate::ai::user_display_name;
use crate::database::models::InsertPhotoInsight; use crate::database::models::InsertPhotoInsight;
use crate::database::{ use crate::database::{
CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, KnowledgeDao, LocationHistoryDao, CalendarEventDao, DailySummaryDao, DbError, ExifDao, InsightDao, KnowledgeDao,
SearchHistoryDao, LocationHistoryDao, SearchHistoryDao,
}; };
use crate::libraries::Library; use crate::libraries::Library;
use crate::memories::extract_date_from_filename; use crate::memories::extract_date_from_filename;
use crate::otel::global_tracer; use crate::otel::global_tracer;
use crate::tags::TagDao; 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 /// Max location records rendered by `tool_get_location_history`. The DAO
/// query is range-bounded, not limited, so the tool caps the rendered list /// query is range-bounded, not limited, so the tool caps the rendered list
/// and labels the truncation via `found_header`. /// and labels the truncation via `found_header`.
const LOCATION_HISTORY_DISPLAY_LIMIT: usize = 20; 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, /// Strip common markdown decoration (bold/italic markers, heading hashes,
/// backticks, quotes) from both ends of a model-emitted title. Models wrap /// backticks, quotes) from both ends of a model-emitted title. Models wrap
/// the line despite the prompt: `**Title: A Day in the Woods**`, /// the line despite the prompt: `**Title: A Day in the Woods**`,
@@ -3004,11 +3016,13 @@ Return ONLY the summary, nothing else."#,
updated_at: now, updated_at: now,
}; };
let mut kdao = self match retry_with_backoff_if("tool_store_entity/upsert_entity", 4, is_sqlite_busy, || {
.knowledge_dao let mut kdao = self
.lock() .knowledge_dao
.expect("Unable to lock KnowledgeDao"); .lock()
match kdao.upsert_entity(cx, insert) { .expect("Unable to lock KnowledgeDao");
kdao.upsert_entity(cx, insert.clone())
}) {
Ok(entity) => { Ok(entity) => {
let mut response = format!( let mut response = format!(
"Entity stored: ID:{} | {} | {} | confidence:{:.2}", "Entity stored: ID:{} | {} | {} | confidence:{:.2}",
@@ -3116,16 +3130,17 @@ Return ONLY the summary, nothing else."#,
last_modified_at: None, last_modified_at: None,
}; };
let mut kdao = self let (stored_fact, is_new) =
.knowledge_dao match retry_with_backoff_if("tool_store_fact/upsert_fact", 4, is_sqlite_busy, || {
.lock() let mut kdao = self
.expect("Unable to lock KnowledgeDao"); .knowledge_dao
.lock()
// Upsert the fact (corroboration bumps confidence if duplicate) .expect("Unable to lock KnowledgeDao");
let (stored_fact, is_new) = match kdao.upsert_fact(cx, fact) { kdao.upsert_fact(cx, fact.clone())
Ok(r) => r, }) {
Err(e) => return format!("Error storing fact: {:?}", e), 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 // Upsert a photo link so this entity is associated with this photo
let link = InsertEntityPhotoLink { let link = InsertEntityPhotoLink {
@@ -3134,7 +3149,20 @@ Return ONLY the summary, nothing else."#,
file_path: file_path.to_string(), file_path: file_path.to_string(),
role: photo_role, 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!( log::warn!(
"Failed to upsert photo link for entity {}: {:?}", "Failed to upsert photo link for entity {}: {:?}",
subject_entity_id, subject_entity_id,
@@ -3220,12 +3248,13 @@ Return ONLY the summary, nothing else."#,
log::info!("tool_update_fact: fact_id={}", fact_id); log::info!("tool_update_fact: fact_id={}", fact_id);
let mut kdao = self match retry_with_backoff_if("tool_update_fact/update_fact", 4, is_sqlite_busy, || {
.knowledge_dao let mut kdao = self
.lock() .knowledge_dao
.expect("Unable to lock KnowledgeDao"); .lock()
.expect("Unable to lock KnowledgeDao");
match kdao.update_fact(cx, fact_id, patch, Some((model, backend))) { kdao.update_fact(cx, fact_id, patch.clone(), Some((model, backend)))
}) {
Ok(Some(f)) => format!( Ok(Some(f)) => format!(
"Updated fact ID:{} (status={}, confidence={:.2})", "Updated fact ID:{} (status={}, confidence={:.2})",
f.id, f.status, f.confidence f.id, f.status, f.confidence
@@ -3276,12 +3305,18 @@ Return ONLY the summary, nothing else."#,
new_fact_id new_fact_id
); );
let mut kdao = self match retry_with_backoff_if(
.knowledge_dao "tool_supersede_fact/supersede_fact",
.lock() 4,
.expect("Unable to lock KnowledgeDao"); is_sqlite_busy,
|| {
match kdao.supersede_fact(cx, old_fact_id, new_fact_id, Some((model, backend))) { 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!( Ok(Some(f)) => format!(
"Superseded fact ID:{} (now status={}, valid_until={:?})", "Superseded fact ID:{} (now status={}, valid_until={:?})",
f.id, f.status, f.valid_until f.id, f.status, f.valid_until
@@ -4396,17 +4431,54 @@ Return ONLY the summary, nothing else."#,
iterations_used iterations_used
); );
messages.push(ChatMessage::user(format!( messages.push(ChatMessage::user(format!(
"Based on the context gathered, please write the final photo insight. Start with \"Title: <short 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: <short 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() user_display_name()
))); )));
let (final_response, prompt_tokens, eval_tokens) = backend let (final_response, prompt_tokens, eval_tokens) = backend
.chat() .chat()
.chat_with_tools(messages.clone(), vec![]) .chat_with_tools(messages.clone(), tools.clone())
.await?; .await?;
last_prompt_eval_count = prompt_tokens; last_prompt_eval_count = prompt_tokens;
last_eval_count = eval_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 loop_cx
@@ -4604,13 +4676,41 @@ Return ONLY the summary, nothing else."#,
max_iter max_iter
); );
messages.push(ChatMessage::user( 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 let (final_response, _, _) = backend
.chat() .chat()
.chat_with_tools(messages.clone(), vec![]) .chat_with_tools(messages.clone(), tools.clone())
.await?; .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) Ok(final_content)
+16
View File
@@ -19,6 +19,11 @@ use futures::stream::{BoxStream, StreamExt};
const DEFAULT_BASE_URL: &str = "https://openrouter.ai/api/v1"; const DEFAULT_BASE_URL: &str = "https://openrouter.ai/api/v1";
const DEFAULT_EMBEDDING_MODEL: &str = "openai/text-embedding-3-small"; const DEFAULT_EMBEDDING_MODEL: &str = "openai/text-embedding-3-small";
const CACHE_DURATION_SECS: u64 = 15 * 60; 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)] #[derive(Clone)]
struct CachedEntry<T> { struct CachedEntry<T> {
@@ -64,6 +69,9 @@ pub struct OpenRouterClient {
top_p: Option<f32>, top_p: Option<f32>,
top_k: Option<i32>, top_k: Option<i32>,
min_p: Option<f32>, min_p: Option<f32>,
/// Upper bound on generated tokens per call, sent as `max_tokens`.
/// `None` omits the field (provider default).
max_tokens: Option<u32>,
/// Optional `HTTP-Referer` header OpenRouter uses for attribution. /// Optional `HTTP-Referer` header OpenRouter uses for attribution.
pub referer: Option<String>, pub referer: Option<String>,
/// Optional `X-Title` header OpenRouter uses for attribution. /// Optional `X-Title` header OpenRouter uses for attribution.
@@ -87,6 +95,7 @@ impl OpenRouterClient {
top_p: None, top_p: None,
top_k: None, top_k: None,
min_p: None, min_p: None,
max_tokens: Some(DEFAULT_MAX_TOKENS),
referer: None, referer: None,
app_title: None, app_title: None,
} }
@@ -115,6 +124,10 @@ impl OpenRouterClient {
self.min_p = min_p; self.min_p = min_p;
} }
pub fn set_max_tokens(&mut self, max_tokens: Option<u32>) {
self.max_tokens = max_tokens;
}
pub fn set_attribution(&mut self, referer: Option<String>, app_title: Option<String>) { pub fn set_attribution(&mut self, referer: Option<String>, app_title: Option<String>) {
self.referer = referer; self.referer = referer;
self.app_title = app_title; self.app_title = app_title;
@@ -288,6 +301,9 @@ impl OpenRouterClient {
// directly transferable. Skip rather than silently mis-map. // directly transferable. Skip rather than silently mis-map.
let _ = c; let _ = c;
} }
if let Some(m) = self.max_tokens {
v.push(("max_tokens", json!(m)));
}
v v
} }
} }
+1
View File
@@ -96,6 +96,7 @@ pub struct EntityPatch {
pub confidence: Option<f32>, pub confidence: Option<f32>,
} }
#[derive(Clone)]
pub struct FactPatch { pub struct FactPatch {
pub predicate: Option<String>, pub predicate: Option<String>,
pub object_value: Option<String>, pub object_value: Option<String>,
+3 -3
View File
@@ -294,7 +294,7 @@ pub struct InsertLibrary<'a> {
// --- Knowledge memory models --- // --- Knowledge memory models ---
#[derive(Insertable)] #[derive(Insertable, Clone)]
#[diesel(table_name = entities)] #[diesel(table_name = entities)]
pub struct InsertEntity { pub struct InsertEntity {
pub name: String, pub name: String,
@@ -320,7 +320,7 @@ pub struct Entity {
pub updated_at: i64, pub updated_at: i64,
} }
#[derive(Insertable)] #[derive(Insertable, Clone)]
#[diesel(table_name = entity_facts)] #[diesel(table_name = entity_facts)]
pub struct InsertEntityFact { pub struct InsertEntityFact {
pub subject_entity_id: i32, pub subject_entity_id: i32,
@@ -396,7 +396,7 @@ pub struct EntityFact {
pub last_modified_at: Option<i64>, pub last_modified_at: Option<i64>,
} }
#[derive(Insertable)] #[derive(Insertable, Clone)]
#[diesel(table_name = entity_photo_links)] #[diesel(table_name = entity_photo_links)]
pub struct InsertEntityPhotoLink { pub struct InsertEntityPhotoLink {
pub entity_id: i32, pub entity_id: i32,
+19
View File
@@ -156,6 +156,25 @@ pub fn detect_missing_files_for_library(
(deleted, next_offset) (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 /// Refresh the `(library_id, rel_path)` back-refs on hash-keyed
/// tables. A back-ref is stale when: /// tables. A back-ref is stale when:
/// - its `content_hash` is non-null, /// - its `content_hash` is non-null,
+7
View File
@@ -402,6 +402,13 @@ fn build_openrouter_from_env() -> Option<Arc<OpenRouterClient>> {
if let Ok(model) = env::var("OPENROUTER_EMBEDDING_MODEL") { if let Ok(model) = env::var("OPENROUTER_EMBEDDING_MODEL") {
client.set_embedding_model(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::<u32>().ok())
{
client.set_max_tokens((n > 0).then_some(n));
}
Some(Arc::new(client)) Some(Arc::new(client))
} }
+61 -1
View File
@@ -27,7 +27,7 @@ where
Err(e) => e, Err(e) => e,
}; };
for attempt in 1..=max_retries { 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_range = base.as_millis() as f64 * 0.25;
let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range) as u128; let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range) as u128;
let delay = base.as_millis() as i128 + jitter as i128; let delay = base.as_millis() as i128 + jitter as i128;
@@ -54,6 +54,66 @@ where
Err(last_err) 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<F, T, E, U: Fn(&E) -> bool>(
label: &str,
max_retries: u32,
retry_if: U,
mut op: F,
) -> Result<T, E>
where
F: FnMut() -> Result<T, E>,
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 /// Normalize a file path to use forward slashes for cross-platform consistency
/// This ensures paths stored in the database always use `/` regardless of OS /// This ensures paths stored in the database always use `/` regardless of OS
/// ///
+3
View File
@@ -577,6 +577,9 @@ pub fn watch_files(
// model" for the rules. // model" for the rules.
{ {
let mut conn = image_api::database::connect(); 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); let _ = image_api::database::reconcile::run(&mut conn);
// Back-ref refresh: hash-keyed rows whose // Back-ref refresh: hash-keyed rows whose