1 Commits
Author SHA1 Message Date
Cameron CordesandClaude Opus 5.5 e104f452f3 feat(insights): read location context from Apollo (#177)
The local location_history table is a one-time Takeout import: it ends
in 2019, still has the ghost-device and coarse cell-fix samples Apollo
has since excluded, and was read loosely. The per-photo line took the
nearest sample within +/-3 h at any distance (17% of photo moments got a
position >2 km away), and get_location_history listed the first 20 rows
of a +/-14-day window sorted ascending (always ~14 days before the date).

With APOLLO_API_BASE_URL set:
- per-photo context comes from /api/location/at (30 min max gap; no
  sample -> no line). A Place stay renders as "at Home (14:02-18:30)" in
  the display zone; an unnamed sample is reverse-geocoded.
- get_location_history returns Apollo's /api/location/day timeline for
  the date (+/-1 day at most), and is offered even with an empty local
  table.

Both fall back to the local table when Apollo is unset or a call fails;
the local tool path now covers the requested local day(s) and samples
evenly instead of listing a window's first rows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 04:11:37 +00:00
9 changed files with 24 additions and 178 deletions

No files matched your search

@@ -1 +0,0 @@
DROP TABLE IF EXISTS clip_failures;
@@ -1,17 +0,0 @@
-- "Don't retry" markers for the CLIP drain. A permanent encode failure
-- (bytes Apollo can't decode — corrupt or non-image files with an image
-- extension) used to leave `clip_embedding` NULL, so the drain re-sent
-- the file and logged the same 422 every pass.
--
-- Keyed on the path, scoped to the bytes that failed: the drain skips a
-- row only while its `content_hash` still matches, so replacing the file
-- makes it eligible again with no cleanup. To retry everything (e.g.
-- after an Apollo decoder upgrade): `DELETE FROM clip_failures;`
CREATE TABLE clip_failures (
library_id INTEGER NOT NULL REFERENCES libraries(id) ON DELETE CASCADE,
rel_path TEXT NOT NULL,
content_hash TEXT NOT NULL,
error TEXT NOT NULL,
failed_at BIGINT NOT NULL,
PRIMARY KEY (library_id, rel_path)
);
+9 -19
View File
@@ -27,10 +27,8 @@
//! L2-normalized so the rerank reduces to a plain dot product.
//!
//! Error mapping (reflected in [`ClipError`]):
//! - 422 `decode_failed` / `empty_text` → rejected input: ImageApi marks
//! the row failed or surfaces the empty-query error to the search caller.
//! - Any other 4xx → permanent, but not the input's fault (auth, routing),
//! so nothing is marked.
//! - 422 `decode_failed` / `empty_text` → permanent: ImageApi marks the
//! row failed or surfaces the empty-query error to the search caller.
//! - 503 `cuda_oom` / `engine_unavailable` → defer-and-retry: no marker.
//! - Any other 5xx / network error → defer.
@@ -93,12 +91,8 @@ pub struct ClipHealth {
#[derive(Debug)]
pub enum ClipError {
/// Apollo refused the input itself (422: decode failure on
/// /encode_image, empty text on /encode_text). Retrying the same
/// bytes will fail the same way.
Rejected(anyhow::Error),
/// Any other 4xx: won't fix itself on retry, but it's a deployment
/// problem (auth, routing), not this input's.
/// Apollo refused for a reason that won't change on retry (decode
/// failure on /encode_image, empty text on /encode_text).
Permanent(anyhow::Error),
/// Apollo couldn't process this turn but might next time (CUDA OOM,
/// engine not loaded, network hiccup).
@@ -111,7 +105,6 @@ pub enum ClipError {
impl std::fmt::Display for ClipError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ClipError::Rejected(e) => write!(f, "rejected: {e}"),
ClipError::Permanent(e) => write!(f, "permanent: {e}"),
ClipError::Transient(e) => write!(f, "transient: {e}"),
ClipError::Disabled => write!(f, "clip client disabled"),
@@ -285,7 +278,7 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
.unwrap_or_default();
if status == 422 {
return ClipError::Rejected(anyhow::anyhow!(
return ClipError::Permanent(anyhow::anyhow!(
"clip {} {}: {}",
status,
detail_code,
@@ -330,9 +323,6 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
mod tests {
use super::*;
fn is_rejected(e: &ClipError) -> bool {
matches!(e, ClipError::Rejected(_))
}
fn is_permanent(e: &ClipError) -> bool {
matches!(e, ClipError::Permanent(_))
}
@@ -341,16 +331,16 @@ mod tests {
}
#[test]
fn classify_422_decode_failed_is_rejected() {
assert!(is_rejected(&classify_error_response(
fn classify_422_decode_failed_is_permanent() {
assert!(is_permanent(&classify_error_response(
422,
r#"{"detail":"decode_failed: bad bytes"}"#
)));
}
#[test]
fn classify_422_empty_text_is_rejected() {
assert!(is_rejected(&classify_error_response(
fn classify_422_empty_text_is_permanent() {
assert!(is_permanent(&classify_error_response(
422,
r#"{"detail":"empty_text"}"#
)));
+1 -1
View File
@@ -227,7 +227,7 @@ async fn main() -> anyhow::Result<()> {
}
let _ = call_start;
}
Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
Err(ClipError::Permanent(e)) => {
perm_fail += 1;
warn!("permanent encode failure for {rel_path}: {e}");
}
+1 -3
View File
@@ -172,9 +172,7 @@ pub async fn score_photos(
// 1. Encode the query text. Fast — Apollo's text encoder is ~50ms on CPU.
let query_resp = match state.clip_client.encode_text(q_text).await {
Ok(r) => r,
Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
return Err(ScoreError::Rejected(e.to_string()));
}
Err(ClipError::Permanent(e)) => return Err(ScoreError::Rejected(e.to_string())),
Err(ClipError::Transient(e)) => return Err(ScoreError::Unavailable(e.to_string())),
Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
};
+13 -27
View File
@@ -6,12 +6,12 @@
//! parallel encode calls to Apollo, and persist the resulting embeddings
//! into `image_exif.clip_embedding` / `clip_model_version`.
//!
//! A permanent failure (un-decodable bytes) records a `clip_failures`
//! marker scoped to the row's `content_hash`, so the drain stops
//! re-sending the file (and re-logging the same 422) every pass until
//! its bytes change. `file_types::is_image_file` filters out videos /
//! non-media client-side, so most permanent failures are corrupt files
//! or non-images with an image extension.
//! Unlike the face pipeline, CLIP has no marker rows — a permanent
//! failure (un-decodable bytes) leaves the row's `clip_embedding` NULL
//! and the drain will retry on the next tick. For personal-library
//! scale this is fine; the per-tick cap bounds the wasted work, and
//! `file_types::is_image_file` filters out videos / non-media client-
//! side so most permanent failures are decoded-but-corrupt files (rare).
//!
//! The watcher thread isn't in any pre-existing async context, so we
//! build a short-lived tokio runtime per pass and `block_on` the join
@@ -172,29 +172,15 @@ async fn process_one(
cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version
);
}
Err(ClipError::Rejected(e)) => {
warn!(
"clip_watch: permanent failure on {}: {} — marking failed",
cand.rel_path, e
);
let mut dao = exif_dao.lock().expect("exif dao");
if let Err(me) = dao.mark_clip_failed(
library_id,
&cand.rel_path,
&cand.content_hash,
&e.to_string(),
) {
warn!(
"clip_watch: mark failed errored for {}: {:?}",
cand.rel_path, me
);
}
}
Err(ClipError::Permanent(e)) => {
// Not this file's fault (auth / routing) — don't mark it, or
// a misconfigured Apollo would blacklist the whole library.
// No marker — the row sits with NULL embedding and the drain
// retries next pass. For personal-library scale the cost of
// re-attempting permanently-broken files is bounded by the
// per-tick cap. If this becomes a recurring noise source,
// add a `clip_status` column with `failed` semantics like
// face_detections has.
warn!(
"clip_watch: Apollo refused {} (will retry next pass): {}",
"clip_watch: permanent failure on {} (will retry next pass): {}",
cand.rel_path, e
);
}
-88
View File
@@ -551,17 +551,6 @@ pub trait ExifDao: Sync + Send {
model_version: &str,
) -> Result<(), DbError>;
/// Record that Apollo permanently rejected these bytes, so
/// `list_clip_unencoded_candidates` stops offering the row until its
/// `content_hash` changes. Upserts — one marker per path.
fn mark_clip_failed(
&mut self,
library_id: i32,
rel_path: &str,
content_hash: &str,
error: &str,
) -> Result<(), DbError>;
/// Load every `(content_hash, clip_embedding)` pair from the live
/// image_exif rows for the given libraries, optionally filtered to a
/// single `model_version` (cosine sim across mixed geometries is
@@ -1530,21 +1519,10 @@ impl ExifDao for SqliteExifDao {
// (clip_embedding IS NULL AND content_hash IS NOT NULL)
// filter; the planner hits it directly. ORDER BY id ASC
// keeps drain progress monotone across ticks.
// Skip rows whose current bytes already failed permanently
// (`mark_clip_failed`); a changed hash makes them eligible.
let failed = schema::clip_failures::table
.filter(schema::clip_failures::library_id.eq(library_id))
.filter(schema::clip_failures::rel_path.eq(rel_path))
.filter(
schema::clip_failures::content_hash
.nullable()
.eq(content_hash),
);
image_exif
.filter(library_id.eq(library_id_val))
.filter(clip_embedding.is_null())
.filter(content_hash.is_not_null())
.filter(diesel::dsl::not(diesel::dsl::exists(failed)))
.select((rel_path, content_hash.assume_not_null()))
.order(id.asc())
.limit(limit)
@@ -1607,41 +1585,6 @@ impl ExifDao for SqliteExifDao {
})
}
fn mark_clip_failed(
&mut self,
library_id_val: i32,
rel_path_val: &str,
content_hash_val: &str,
error_val: &str,
) -> Result<(), DbError> {
trace_db_call("insert", "mark_clip_failed", |_span| {
use schema::clip_failures::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get ExifDao");
let now = chrono::Utc::now().timestamp();
diesel::insert_into(clip_failures)
.values((
library_id.eq(library_id_val),
rel_path.eq(rel_path_val),
content_hash.eq(content_hash_val),
error.eq(error_val),
failed_at.eq(now),
))
.on_conflict((library_id, rel_path))
.do_update()
.set((
content_hash.eq(content_hash_val),
error.eq(error_val),
failed_at.eq(now),
))
.execute(connection.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))
})
.map_err(|e| DbError::log(DbErrorKind::InsertError, e))
}
fn list_clip_index(
&mut self,
library_ids_val: &[i32],
@@ -2616,37 +2559,6 @@ mod exif_dao_tests {
SqliteExifDao::from_connection(conn)
}
#[test]
fn clip_failure_marker_skips_row_until_its_bytes_change() {
let mut dao = setup_two_libraries();
insert_row(&mut dao, 1, "bad.jpg", None);
insert_row(&mut dao, 1, "good.jpg", None);
insert_row(&mut dao, 2, "bad.jpg", None);
for (lib, rel) in [(1, "bad.jpg"), (1, "good.jpg"), (2, "bad.jpg")] {
dao.backfill_content_hash(lib, rel, "h1", 1).unwrap();
}
let rels = |dao: &mut SqliteExifDao, lib| -> Vec<String> {
dao.list_clip_unencoded_candidates(lib, 10)
.unwrap()
.into_iter()
.map(|(rel, _)| rel)
.collect()
};
dao.mark_clip_failed(1, "bad.jpg", "h1", "decode_failed")
.unwrap();
assert_eq!(rels(&mut dao, 1), vec!["good.jpg"]);
// Same path in another library is a different file.
assert_eq!(rels(&mut dao, 2), vec!["bad.jpg"]);
// Re-marking upserts rather than erroring on the primary key.
dao.mark_clip_failed(1, "bad.jpg", "h1", "decode_failed")
.unwrap();
// The file was replaced: new bytes get another try.
dao.backfill_content_hash(1, "bad.jpg", "h2", 1).unwrap();
assert_eq!(rels(&mut dao, 1), vec!["bad.jpg", "good.jpg"]);
}
#[test]
fn get_all_with_date_taken_union_returns_all_libraries() {
let mut dao = setup_two_libraries();
-12
View File
@@ -18,16 +18,6 @@ diesel::table! {
}
}
diesel::table! {
clip_failures (library_id, rel_path) {
library_id -> Integer,
rel_path -> Text,
content_hash -> Text,
error -> Text,
failed_at -> BigInt,
}
}
diesel::table! {
daily_conversation_summaries (id) {
id -> Integer,
@@ -358,7 +348,6 @@ diesel::table! {
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
diesel::joinable!(entity_photo_links -> entities (entity_id));
diesel::joinable!(entity_photo_links -> libraries (library_id));
diesel::joinable!(clip_failures -> libraries (library_id));
diesel::joinable!(face_detections -> libraries (library_id));
diesel::joinable!(face_detections -> persons (person_id));
diesel::joinable!(image_exif -> libraries (library_id));
@@ -371,7 +360,6 @@ diesel::joinable!(video_preview_clips -> libraries (library_id));
diesel::allow_tables_to_appear_in_same_query!(
calendar_events,
clip_failures,
daily_conversation_summaries,
entities,
entity_facts,
-10
View File
@@ -1910,16 +1910,6 @@ mod tests {
Ok(())
}
fn mark_clip_failed(
&mut self,
_library_id: i32,
_rel_path: &str,
_content_hash: &str,
_error: &str,
) -> Result<(), DbError> {
Ok(())
}
fn list_clip_index(
&mut self,
_library_ids: &[i32],