fix(images): pick the decoder from the bytes, not the extension

`image::open` chooses the decoder by extension, so a PNG saved as .jpg
(Android screenshots, chat exports) failed thumbnails, previews,
perceptual hashes, insights and face crops with "Illegal start bytes:
8950". New `file_types::open_image` sniffs the header and falls back to
the extension when it's unknown; every decode site uses it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

migrations/2026-10-18-000000_add_clip_failures/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS clip_failures;

migrations/2026-10-18-000000_add_clip_failures/up.sql

@@ -0,0 +1,17 @@
+-- "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)
+);

src/ai/clip_client.rs

@@ -27,8 +27,10 @@
 //! L2-normalized so the rerank reduces to a plain dot product.
 //!
 //! Error mapping (reflected in [`ClipError`]):
-//! - 422 `decode_failed` / `empty_text` → permanent: ImageApi marks the
-//!   row failed or surfaces the empty-query error to the search caller.
+//! - 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.
 //! - 503 `cuda_oom` / `engine_unavailable` → defer-and-retry: no marker.
 //! - Any other 5xx / network error → defer.
 
@@ -91,8 +93,12 @@ pub struct ClipHealth {
 
 #[derive(Debug)]
 pub enum ClipError {
-    /// Apollo refused for a reason that won't change on retry (decode
-    /// failure on /encode_image, empty text on /encode_text).
+    /// 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.
     Permanent(anyhow::Error),
     /// Apollo couldn't process this turn but might next time (CUDA OOM,
     /// engine not loaded, network hiccup).
@@ -105,6 +111,7 @@ 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"),
@@ -278,7 +285,7 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
         .unwrap_or_default();
 
     if status == 422 {
-        return ClipError::Permanent(anyhow::anyhow!(
+        return ClipError::Rejected(anyhow::anyhow!(
             "clip {} {}: {}",
             status,
             detail_code,
@@ -323,6 +330,9 @@ 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(_))
     }
@@ -331,16 +341,16 @@ mod tests {
     }
 
     #[test]
-    fn classify_422_decode_failed_is_permanent() {
-        assert!(is_permanent(&classify_error_response(
+    fn classify_422_decode_failed_is_rejected() {
+        assert!(is_rejected(&classify_error_response(
             422,
             r#"{"detail":"decode_failed: bad bytes"}"#
         )));
     }
 
     #[test]
-    fn classify_422_empty_text_is_permanent() {
-        assert!(is_permanent(&classify_error_response(
+    fn classify_422_empty_text_is_rejected() {
+        assert!(is_rejected(&classify_error_response(
             422,
             r#"{"detail":"empty_text"}"#
         )));

src/bin/probe_clip_search.rs

@@ -227,7 +227,7 @@ async fn main() -> anyhow::Result<()> {
                     }
                     let _ = call_start;
                 }
-                Err(ClipError::Permanent(e)) => {
+                Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
                     perm_fail += 1;
                     warn!("permanent encode failure for {rel_path}: {e}");
                 }

src/clip_search.rs

@@ -172,7 +172,9 @@ 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::Permanent(e)) => return Err(ScoreError::Rejected(e.to_string())),
+        Err(ClipError::Rejected(e) | 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),
     };

src/clip_watch.rs

@@ -6,12 +6,12 @@
 //! parallel encode calls to Apollo, and persist the resulting embeddings
 //! into `image_exif.clip_embedding` / `clip_model_version`.
 //!
-//! 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).
+//! 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.
 //!
 //! 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,15 +172,29 @@ async fn process_one(
                 cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version
             );
         }
-        Err(ClipError::Permanent(e)) => {
-            // 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.
+        Err(ClipError::Rejected(e)) => {
             warn!(
-                "clip_watch: permanent failure on {} (will retry next pass): {}",
+                "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.
+            warn!(
+                "clip_watch: Apollo refused {} (will retry next pass): {}",
                 cand.rel_path, e
             );
         }

src/database/mod.rs

@@ -551,6 +551,17 @@ 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
@@ -1519,10 +1530,21 @@ 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)
@@ -1585,6 +1607,41 @@ 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],
@@ -2559,6 +2616,37 @@ 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();

src/database/schema.rs

@@ -18,6 +18,16 @@ 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,
@@ -348,6 +358,7 @@ 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));
@@ -360,6 +371,7 @@ 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,

src/files.rs

@@ -1910,6 +1910,16 @@ 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],