feat(agentic-loop): retry transient knowledge-write locks + re-open tools on loop exhaustion
Transcript analysis (359 insights / ~3.3k tool results) showed 31
"database is locked" errors from knowledge-memory tools and 50/359
runs (14%) exhausting the 6-iteration budget with no final answer.
The exhaustion nudge re-invoked chat with an EMPTY tool list, so a
model that still needed information could not call anything.
- utils: add retry_with_backoff_if — same exponential backoff + jitter
as retry_with_backoff, but retries only while the predicate classifies
the error as transient; deterministic failures short-circuit.
- insight_generator: wrap the four knowledge write tools
(store_fact/upsert_fact, store_entity/upsert_entity, update_fact,
supersede_fact) plus the store_fact photo-link upsert in
retry_with_backoff_if gated on a SQLITE_BUSY detector ("database is
locked"/"database is busy" massage in DbError.source). WAL +
busy_timeout absorb most contention; a parallel insight run or the
face-watch write storm still surfaces it, and a short backoff retry
clears it instead of surfacing a hard error to the model.
- insight_generator: exhaustion nudge now re-offers the full tool set
instead of vec![] and explicitly says the model may call more tools;
if it does take one, results are collected and a bounded tool-less
turn forces the actual answer so we never store an empty insight.
Same treatment for the read-only tool loop.
- models/knowledge_dao: derive Clone on InsertEntity, InsertEntityFact,
InsertEntityPhotoLink and FactPatch so retried FnMut closures can
re-issue the same write value on each attempt.
This commit is contained in:
1 parent
3a93e5cae6
commit
4e3ee1973c
4 files changed
+202
-41
No files matched your search
+138
-38
@@ -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)
|
||||||
|
|||||||
@@ -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>,
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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(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
|
/// 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
|
||||||
///
|
///
|
||||||
|
|||||||
Reference in new issue
Block a user