Compare commits
3
Commits
e104f452f3
...
f62d2ecfb4
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f62d2ecfb4 | ||
|
|
bc27aea81a | ||
|
|
e75e60786d |
No files matched your search
@@ -981,6 +981,20 @@ string. See `src/ai/apollo_client.rs` and `src/ai/insight_generator.rs`:
|
|||||||
variant; auto-enrichment covers the photo-context path and the agentic
|
variant; auto-enrichment covers the photo-context path and the agentic
|
||||||
tool covers ad-hoc lat/lon questions in chat continuation.
|
tool covers ad-hoc lat/lon questions in chat continuation.
|
||||||
|
|
||||||
|
- **Location context** (#177, Apollo #51): Apollo also owns the location
|
||||||
|
track — cleaned of ghost devices and coarse cell fixes, and current (HA
|
||||||
|
siphon, Garmin), unlike our `location_history` table, a one-time Takeout
|
||||||
|
import that ends in 2019. When configured, the per-photo location line
|
||||||
|
comes from `/api/location/at?ts=&max_gap=1800` (no sample within 30 min →
|
||||||
|
no line, instead of the old nearest-within-±3 h at any distance; a Place
|
||||||
|
stay renders as "You were at Home (14:02–18:30)" in the display zone —
|
||||||
|
`render_apollo_location`), and the `get_location_history` tool returns
|
||||||
|
Apollo's `/api/location/day` timeline (stays / stops / travel / gaps) for
|
||||||
|
the date, ±1 day at most. Either falls back to `location_history` when
|
||||||
|
Apollo is unset or the call fails; the local tool path samples evenly
|
||||||
|
across the requested local day(s) rather than listing a window's first
|
||||||
|
rows. The tool is offered whenever Apollo is configured, local table or not.
|
||||||
|
|
||||||
Failure modes degrade silently to the legacy Nominatim path: 5 s timeout,
|
Failure modes degrade silently to the legacy Nominatim path: 5 s timeout,
|
||||||
errors logged at `warn`, empty results returned. Apollo's routes are
|
errors logged at `warn`, empty results returned. Apollo's routes are
|
||||||
unauthenticated (single-user, LAN-trust); add JWT auth here + on Apollo's
|
unauthenticated (single-user, LAN-trust); add JWT auth here + on Apollo's
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
DROP TABLE IF EXISTS clip_failures;
|
||||||
@@ -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)
|
||||||
|
);
|
||||||
+170
-1
@@ -1,4 +1,5 @@
|
|||||||
//! Thin async HTTP client for Apollo's `/api/places/*` endpoints.
|
//! Thin async HTTP client for Apollo's `/api/places/*` and
|
||||||
|
//! `/api/location/*` endpoints.
|
||||||
//!
|
//!
|
||||||
//! Apollo (the personal location-history viewer at the sibling repo) owns
|
//! Apollo (the personal location-history viewer at the sibling repo) owns
|
||||||
//! user-defined Places: `name + lat/lon + radius_m + description (+ optional
|
//! user-defined Places: `name + lat/lon + radius_m + description (+ optional
|
||||||
@@ -10,6 +11,14 @@
|
|||||||
//! 2. The agentic `get_personal_place_at` tool — lets the LLM ask "what
|
//! 2. The agentic `get_personal_place_at` tool — lets the LLM ask "what
|
||||||
//! user-defined place contains this lat/lon" during chat continuation.
|
//! user-defined place contains this lat/lon" during chat continuation.
|
||||||
//!
|
//!
|
||||||
|
//! Apollo also owns the location track itself — cleaned of ghost devices
|
||||||
|
//! and coarse cell fixes, and current (HA siphon, Garmin), unlike the
|
||||||
|
//! frozen 2019 Takeout import in our `location_history` table. The
|
||||||
|
//! insight path's location context (`/api/location/at`) and the
|
||||||
|
//! `get_location_history` tool (`/api/location/day`) read it from Apollo
|
||||||
|
//! when configured (ImageApi #177, Apollo #51), falling back to the local
|
||||||
|
//! table when Apollo is unset or a call fails.
|
||||||
|
//!
|
||||||
//! Apollo does the haversine. This client is plumbing only — no geometry,
|
//! Apollo does the haversine. This client is plumbing only — no geometry,
|
||||||
//! no caching at the moment. If insight throughput ever makes per-photo
|
//! no caching at the moment. If insight throughput ever makes per-photo
|
||||||
//! HTTP latency a problem, swap to a small `Mutex<HashMap>` TTL cache here.
|
//! HTTP latency a problem, swap to a small `Mutex<HashMap>` TTL cache here.
|
||||||
@@ -46,6 +55,73 @@ struct PlacesResponse {
|
|||||||
places: Vec<ApolloPlace>,
|
places: Vec<ApolloPlace>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A kept location sample from Apollo, `offset_s` = sample ts − asked ts.
|
||||||
|
#[allow(dead_code)]
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct LocationSample {
|
||||||
|
pub ts: i64,
|
||||||
|
pub lat: f64,
|
||||||
|
pub lon: f64,
|
||||||
|
#[serde(default)]
|
||||||
|
pub accuracy: Option<i32>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub activity: Option<String>,
|
||||||
|
pub offset_s: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[allow(dead_code)]
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct LocationPlace {
|
||||||
|
pub id: i32,
|
||||||
|
pub name: String,
|
||||||
|
#[serde(default)]
|
||||||
|
pub category: Option<String>,
|
||||||
|
pub radius_m: i32,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The Place stay (run of samples inside one Place) covering an instant.
|
||||||
|
#[allow(dead_code)]
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct LocationStay {
|
||||||
|
pub place_id: i32,
|
||||||
|
#[serde(default)]
|
||||||
|
pub name: Option<String>,
|
||||||
|
pub t0: i64,
|
||||||
|
pub t1: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `/api/location/at`: where the owner was at an instant. `sample` is
|
||||||
|
/// `None` when no kept sample lies within `max_gap` — Apollo says "no
|
||||||
|
/// data" rather than handing back a position from hours away.
|
||||||
|
#[allow(dead_code)]
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct LocationAt {
|
||||||
|
pub ts: i64,
|
||||||
|
#[serde(default)]
|
||||||
|
pub sample: Option<LocationSample>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub nearest_offset_s: Option<i64>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub bracketed: bool,
|
||||||
|
#[serde(default)]
|
||||||
|
pub place: Option<LocationPlace>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub stay: Option<LocationStay>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `/api/location/day`: the owner's timeline for one local day. `text` is
|
||||||
|
/// Apollo's compact rendering (one "HH:MM–HH:MM at Home" line per stay /
|
||||||
|
/// dwell / travel / gap), meant to go into a tool result as-is; the
|
||||||
|
/// structured `items` aren't needed here.
|
||||||
|
#[allow(dead_code)]
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct LocationDay {
|
||||||
|
pub tz: String,
|
||||||
|
pub points: i64,
|
||||||
|
pub km: f64,
|
||||||
|
pub text: String,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct ApolloClient {
|
pub struct ApolloClient {
|
||||||
client: Client,
|
client: Client,
|
||||||
@@ -90,6 +166,52 @@ impl ApolloClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Where the owner was at `ts` (real UTC seconds). `None` when the
|
||||||
|
/// integration is disabled or the call fails — callers fall back to
|
||||||
|
/// the local table; `Some` with `sample: None` means Apollo answered
|
||||||
|
/// "no data within `max_gap`", which callers should respect.
|
||||||
|
pub async fn location_at(&self, ts: i64, max_gap: i64) -> Option<LocationAt> {
|
||||||
|
let base = self.base_url.as_deref()?;
|
||||||
|
let url = format!("{}/api/location/at", base.trim_end_matches('/'));
|
||||||
|
let result: Result<LocationAt> = async {
|
||||||
|
Ok(self
|
||||||
|
.client
|
||||||
|
.get(&url)
|
||||||
|
.query(&[("ts", ts), ("max_gap", max_gap)])
|
||||||
|
.send()
|
||||||
|
.await?
|
||||||
|
.error_for_status()?
|
||||||
|
.json()
|
||||||
|
.await?)
|
||||||
|
}
|
||||||
|
.await;
|
||||||
|
result
|
||||||
|
.map_err(|err| log::warn!("apollo_client: location_at({ts}) failed: {err}"))
|
||||||
|
.ok()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The owner's timeline for `date` (YYYY-MM-DD) as a local day in
|
||||||
|
/// `tz` (IANA). `None` when disabled or the call fails.
|
||||||
|
pub async fn location_day(&self, date: &str, tz: &str) -> Option<LocationDay> {
|
||||||
|
let base = self.base_url.as_deref()?;
|
||||||
|
let url = format!("{}/api/location/day", base.trim_end_matches('/'));
|
||||||
|
let result: Result<LocationDay> = async {
|
||||||
|
Ok(self
|
||||||
|
.client
|
||||||
|
.get(&url)
|
||||||
|
.query(&[("date", date), ("tz", tz)])
|
||||||
|
.send()
|
||||||
|
.await?
|
||||||
|
.error_for_status()?
|
||||||
|
.json()
|
||||||
|
.await?)
|
||||||
|
}
|
||||||
|
.await;
|
||||||
|
result
|
||||||
|
.map_err(|err| log::warn!("apollo_client: location_day({date}, {tz}) failed: {err}"))
|
||||||
|
.ok()
|
||||||
|
}
|
||||||
|
|
||||||
async fn fetch_places_containing(
|
async fn fetch_places_containing(
|
||||||
&self,
|
&self,
|
||||||
base: &str,
|
base: &str,
|
||||||
@@ -108,3 +230,50 @@ impl ApolloClient {
|
|||||||
Ok(body.places)
|
Ok(body.places)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn location_at_parses_hit_and_miss() {
|
||||||
|
let hit: LocationAt = serde_json::from_str(
|
||||||
|
r#"{"ts": 100, "max_gap": 1800,
|
||||||
|
"sample": {"ts": 120, "lat": 41.46, "lon": -81.93, "accuracy": 19, "activity": "STILL", "offset_s": 20},
|
||||||
|
"nearest_offset_s": 20, "bracketed": true,
|
||||||
|
"place": {"id": 7, "name": "Work", "category": null, "radius_m": 250},
|
||||||
|
"stay": {"place_id": 7, "name": "Work", "t0": 50, "t1": 900}}"#,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(hit.sample.unwrap().offset_s, 20);
|
||||||
|
assert_eq!(hit.place.unwrap().name, "Work");
|
||||||
|
assert_eq!(hit.stay.unwrap().t1, 900);
|
||||||
|
|
||||||
|
let miss: LocationAt = serde_json::from_str(
|
||||||
|
r#"{"ts": 100, "max_gap": 1800, "sample": null, "nearest_offset_s": -7200,
|
||||||
|
"bracketed": false, "place": null, "stay": null}"#,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
assert!(miss.sample.is_none());
|
||||||
|
assert_eq!(miss.nearest_offset_s, Some(-7200));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn location_day_ignores_items() {
|
||||||
|
let day: LocationDay = serde_json::from_str(
|
||||||
|
r#"{"start": 0, "end": 86400, "tz": "America/New_York",
|
||||||
|
"items": [{"kind": "gap", "t0": 0, "t1": 1}], "points": 3, "km": 1.5,
|
||||||
|
"text": "00:00–23:59 at Home"}"#,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(day.text, "00:00–23:59 at Home");
|
||||||
|
assert_eq!(day.points, 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn disabled_client_returns_none() {
|
||||||
|
let c = ApolloClient::new(None);
|
||||||
|
assert!(c.location_at(0, 1800).await.is_none());
|
||||||
|
assert!(c.location_day("2016-01-01", "UTC").await.is_none());
|
||||||
|
}
|
||||||
|
}
|
||||||
+19
-9
@@ -27,8 +27,10 @@
|
|||||||
//! L2-normalized so the rerank reduces to a plain dot product.
|
//! L2-normalized so the rerank reduces to a plain dot product.
|
||||||
//!
|
//!
|
||||||
//! Error mapping (reflected in [`ClipError`]):
|
//! Error mapping (reflected in [`ClipError`]):
|
||||||
//! - 422 `decode_failed` / `empty_text` → permanent: ImageApi marks the
|
//! - 422 `decode_failed` / `empty_text` → rejected input: ImageApi marks
|
||||||
//! row failed or surfaces the empty-query error to the search caller.
|
//! 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.
|
//! - 503 `cuda_oom` / `engine_unavailable` → defer-and-retry: no marker.
|
||||||
//! - Any other 5xx / network error → defer.
|
//! - Any other 5xx / network error → defer.
|
||||||
|
|
||||||
@@ -91,8 +93,12 @@ pub struct ClipHealth {
|
|||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum ClipError {
|
pub enum ClipError {
|
||||||
/// Apollo refused for a reason that won't change on retry (decode
|
/// Apollo refused the input itself (422: decode failure on
|
||||||
/// failure on /encode_image, empty text on /encode_text).
|
/// /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),
|
Permanent(anyhow::Error),
|
||||||
/// Apollo couldn't process this turn but might next time (CUDA OOM,
|
/// Apollo couldn't process this turn but might next time (CUDA OOM,
|
||||||
/// engine not loaded, network hiccup).
|
/// engine not loaded, network hiccup).
|
||||||
@@ -105,6 +111,7 @@ pub enum ClipError {
|
|||||||
impl std::fmt::Display for ClipError {
|
impl std::fmt::Display for ClipError {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
match self {
|
match self {
|
||||||
|
ClipError::Rejected(e) => write!(f, "rejected: {e}"),
|
||||||
ClipError::Permanent(e) => write!(f, "permanent: {e}"),
|
ClipError::Permanent(e) => write!(f, "permanent: {e}"),
|
||||||
ClipError::Transient(e) => write!(f, "transient: {e}"),
|
ClipError::Transient(e) => write!(f, "transient: {e}"),
|
||||||
ClipError::Disabled => write!(f, "clip client disabled"),
|
ClipError::Disabled => write!(f, "clip client disabled"),
|
||||||
@@ -278,7 +285,7 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
|
|||||||
.unwrap_or_default();
|
.unwrap_or_default();
|
||||||
|
|
||||||
if status == 422 {
|
if status == 422 {
|
||||||
return ClipError::Permanent(anyhow::anyhow!(
|
return ClipError::Rejected(anyhow::anyhow!(
|
||||||
"clip {} {}: {}",
|
"clip {} {}: {}",
|
||||||
status,
|
status,
|
||||||
detail_code,
|
detail_code,
|
||||||
@@ -323,6 +330,9 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
fn is_rejected(e: &ClipError) -> bool {
|
||||||
|
matches!(e, ClipError::Rejected(_))
|
||||||
|
}
|
||||||
fn is_permanent(e: &ClipError) -> bool {
|
fn is_permanent(e: &ClipError) -> bool {
|
||||||
matches!(e, ClipError::Permanent(_))
|
matches!(e, ClipError::Permanent(_))
|
||||||
}
|
}
|
||||||
@@ -331,16 +341,16 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn classify_422_decode_failed_is_permanent() {
|
fn classify_422_decode_failed_is_rejected() {
|
||||||
assert!(is_permanent(&classify_error_response(
|
assert!(is_rejected(&classify_error_response(
|
||||||
422,
|
422,
|
||||||
r#"{"detail":"decode_failed: bad bytes"}"#
|
r#"{"detail":"decode_failed: bad bytes"}"#
|
||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn classify_422_empty_text_is_permanent() {
|
fn classify_422_empty_text_is_rejected() {
|
||||||
assert!(is_permanent(&classify_error_response(
|
assert!(is_rejected(&classify_error_response(
|
||||||
422,
|
422,
|
||||||
r#"{"detail":"empty_text"}"#
|
r#"{"detail":"empty_text"}"#
|
||||||
)));
|
)));
|
||||||
|
|||||||
+274
-35
@@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex};
|
|||||||
use tracing::Instrument;
|
use tracing::Instrument;
|
||||||
use tracing_opentelemetry::OpenTelemetrySpanExt;
|
use tracing_opentelemetry::OpenTelemetrySpanExt;
|
||||||
|
|
||||||
use crate::ai::apollo_client::{ApolloClient, ApolloPlace};
|
use crate::ai::apollo_client::{ApolloClient, ApolloPlace, LocationAt};
|
||||||
use crate::ai::backend::{BackendKind, ResolvedBackend, SamplingOverrides};
|
use crate::ai::backend::{BackendKind, ResolvedBackend, SamplingOverrides};
|
||||||
use crate::ai::llamacpp::LlamaCppClient;
|
use crate::ai::llamacpp::LlamaCppClient;
|
||||||
use crate::ai::llm_client::LlmClient;
|
use crate::ai::llm_client::LlmClient;
|
||||||
@@ -114,6 +114,105 @@ fn compose_location_string(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// How far from a photo's instant Apollo may look for a location sample
|
||||||
|
/// for the always-on insight context. Past this Apollo answers "no data"
|
||||||
|
/// — the old local lookup took the nearest sample within ±3 h at any
|
||||||
|
/// distance, and 17 % of photo moments got a position > 2 km from where
|
||||||
|
/// the photo was taken.
|
||||||
|
const APOLLO_LOCATION_MAX_GAP_S: i64 = 1800;
|
||||||
|
|
||||||
|
/// Apollo activities that say nothing about what the owner was doing.
|
||||||
|
const UNINFORMATIVE_ACTIVITIES: &[&str] = &["UNKNOWN", "TILTING", "EXITING_VEHICLE"];
|
||||||
|
|
||||||
|
/// Render Apollo's `/api/location/at` answer as the insight prompt's
|
||||||
|
/// location-history line, or `None` when it adds nothing:
|
||||||
|
///
|
||||||
|
/// - no sample within the gap → `None` ("no data" beats a stale position);
|
||||||
|
/// - the photo has EXIF GPS within 500 m of the sample and Apollo knows no
|
||||||
|
/// stay or activity → `None` (the EXIF location line already says it);
|
||||||
|
/// - otherwise "You were at Home (14:02–18:30)", "at Home", or "near
|
||||||
|
/// <nominatim or coords>", plus the activity and, when the fix is more
|
||||||
|
/// than 5 min from the photo, how far off it is.
|
||||||
|
///
|
||||||
|
/// Stay times render in the owner's display zone. Free function so it can
|
||||||
|
/// be tested without an InsightGenerator; `nominatim` is the caller's
|
||||||
|
/// reverse-geocode of the sample (only needed when there's no Place).
|
||||||
|
fn render_apollo_location(
|
||||||
|
at: &LocationAt,
|
||||||
|
exif_gps: Option<(f64, f64)>,
|
||||||
|
zone: chrono_tz::Tz,
|
||||||
|
nominatim: Option<String>,
|
||||||
|
) -> Option<String> {
|
||||||
|
let sample = at.sample.as_ref()?;
|
||||||
|
let activity = sample
|
||||||
|
.activity
|
||||||
|
.as_deref()
|
||||||
|
.filter(|a| !UNINFORMATIVE_ACTIVITIES.contains(a));
|
||||||
|
if let Some((lat, lon)) = exif_gps {
|
||||||
|
let km = InsightGenerator::haversine_distance(lat, lon, sample.lat, sample.lon);
|
||||||
|
if km < 0.5 && at.stay.is_none() && activity.is_none() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let hm = |ts: i64| {
|
||||||
|
DateTime::from_timestamp(ts, 0)
|
||||||
|
.map(|d| d.with_timezone(&zone).format("%H:%M").to_string())
|
||||||
|
.unwrap_or_default()
|
||||||
|
};
|
||||||
|
let place_name = at
|
||||||
|
.stay
|
||||||
|
.as_ref()
|
||||||
|
.and_then(|s| s.name.clone())
|
||||||
|
.or_else(|| at.place.as_ref().map(|p| p.name.clone()));
|
||||||
|
let mut line = match (&at.stay, place_name) {
|
||||||
|
(Some(stay), Some(name)) => {
|
||||||
|
format!("You were at {} ({}–{})", name, hm(stay.t0), hm(stay.t1))
|
||||||
|
}
|
||||||
|
(None, Some(name)) => format!("You were at {}", name),
|
||||||
|
(_, None) => match nominatim {
|
||||||
|
Some(n) => format!("You were near {}", n),
|
||||||
|
None => format!("You were near {:.4}, {:.4}", sample.lat, sample.lon),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
if let Some(a) = activity {
|
||||||
|
line.push_str(&format!(" ({})", a));
|
||||||
|
}
|
||||||
|
if sample.offset_s.abs() > 300 {
|
||||||
|
line.push_str(&format!(
|
||||||
|
" [location fix {} min {} the photo]",
|
||||||
|
sample.offset_s.abs() / 60,
|
||||||
|
if sample.offset_s < 0 {
|
||||||
|
"before"
|
||||||
|
} else {
|
||||||
|
"after"
|
||||||
|
}
|
||||||
|
));
|
||||||
|
}
|
||||||
|
Some(format!("Location history: {}", line))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One day of Apollo's timeline as a tool-result block.
|
||||||
|
fn format_apollo_day(date: &str, day: &crate::ai::apollo_client::LocationDay) -> String {
|
||||||
|
format!(
|
||||||
|
"{} ({} location points, {:.1} km):\n{}",
|
||||||
|
date, day.points, day.km, day.text
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Up to `limit` items spaced evenly through `items`, first and last
|
||||||
|
/// included, in their original order.
|
||||||
|
fn spread_evenly<T>(items: &[T], limit: usize) -> Vec<&T> {
|
||||||
|
if items.len() <= limit {
|
||||||
|
return items.iter().collect();
|
||||||
|
}
|
||||||
|
if limit < 2 {
|
||||||
|
return items.iter().take(limit).collect();
|
||||||
|
}
|
||||||
|
(0..limit)
|
||||||
|
.map(|k| &items[k * (items.len() - 1) / (limit - 1)])
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
#[derive(Deserialize)]
|
||||||
struct NominatimResponse {
|
struct NominatimResponse {
|
||||||
display_name: Option<String>,
|
display_name: Option<String>,
|
||||||
@@ -352,7 +451,8 @@ impl InsightGenerator {
|
|||||||
.expect("Unable to lock CalendarEventDao");
|
.expect("Unable to lock CalendarEventDao");
|
||||||
dao.get_event_count().map(|n| n > 0).unwrap_or(false)
|
dao.get_event_count().map(|n| n > 0).unwrap_or(false)
|
||||||
};
|
};
|
||||||
let location_history_present = {
|
// Apollo serves the tool when configured, local table or not.
|
||||||
|
let location_history_present = self.apollo_enabled() || {
|
||||||
let mut dao = self
|
let mut dao = self
|
||||||
.location_dao
|
.location_dao
|
||||||
.lock()
|
.lock()
|
||||||
@@ -827,8 +927,10 @@ impl InsightGenerator {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Gather location context for photo timestamp
|
/// Gather location context for photo timestamp. Asks Apollo
|
||||||
/// Finds nearest location record (±30 minutes)
|
/// (`/api/location/at`, cleaned + current track, ±30 min) when it's
|
||||||
|
/// configured; falls back to the nearest local `location_history`
|
||||||
|
/// record within ±3 h when it isn't, or when the call fails.
|
||||||
#[tracing::instrument(name = "ai.context.location", skip_all)]
|
#[tracing::instrument(name = "ai.context.location", skip_all)]
|
||||||
async fn gather_location_context(
|
async fn gather_location_context(
|
||||||
&self,
|
&self,
|
||||||
@@ -837,6 +939,27 @@ impl InsightGenerator {
|
|||||||
) -> Result<Option<String>> {
|
) -> Result<Option<String>> {
|
||||||
let span = tracing::Span::current();
|
let span = tracing::Span::current();
|
||||||
|
|
||||||
|
if let Some(at) = self
|
||||||
|
.apollo_client
|
||||||
|
.location_at(timestamp, APOLLO_LOCATION_MAX_GAP_S)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
span.set_attribute("location_source", "apollo");
|
||||||
|
// Reverse-geocode only an unnamed sample on a photo without
|
||||||
|
// EXIF GPS — with EXIF, the location line already has it.
|
||||||
|
let nominatim = match (&at.sample, &at.place, &at.stay, exif_gps) {
|
||||||
|
(Some(s), None, None, None) => self.reverse_geocode(s.lat, s.lon).await,
|
||||||
|
_ => None,
|
||||||
|
};
|
||||||
|
span.set_status(Status::Ok);
|
||||||
|
return Ok(render_apollo_location(
|
||||||
|
&at,
|
||||||
|
exif_gps,
|
||||||
|
self.display_zone(),
|
||||||
|
nominatim,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
let nearest = {
|
let nearest = {
|
||||||
let mut dao = self
|
let mut dao = self
|
||||||
.location_dao
|
.location_dao
|
||||||
@@ -2339,53 +2462,85 @@ Return ONLY the summary, nothing else."#,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Tool: get_location_history — fetch location records near a date
|
/// Tool: get_location_history — where the owner was on a date.
|
||||||
|
///
|
||||||
|
/// With Apollo configured: Apollo's day timeline (`/api/location/day`:
|
||||||
|
/// Place stays, stops, travel with km, gaps) for the date's local day
|
||||||
|
/// in the owner's display zone, or for date±1 at most. Otherwise, or
|
||||||
|
/// if every Apollo call fails: up to `LOCATION_HISTORY_DISPLAY_LIMIT`
|
||||||
|
/// raw local records spaced evenly across the requested local days.
|
||||||
|
/// (The local path used to list the first 20 rows of a ±14-day window
|
||||||
|
/// sorted ascending — always ~14 days before the date asked about.)
|
||||||
async fn tool_get_location_history(&self, args: &serde_json::Value) -> String {
|
async fn tool_get_location_history(&self, args: &serde_json::Value) -> String {
|
||||||
let date_str = match args.get("date").and_then(|v| v.as_str()) {
|
let date_str = match args.get("date").and_then(|v| v.as_str()) {
|
||||||
Some(d) => d,
|
Some(d) => d,
|
||||||
None => return "Error: missing required parameter 'date'".to_string(),
|
None => return "Error: missing required parameter 'date'".to_string(),
|
||||||
};
|
};
|
||||||
let days_radius = args
|
|
||||||
.get("days_radius")
|
|
||||||
.and_then(|v| v.as_i64())
|
|
||||||
.unwrap_or(14)
|
|
||||||
.clamp(1, 30);
|
|
||||||
|
|
||||||
let date = match NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
|
let date = match NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
|
||||||
Ok(d) => d,
|
Ok(d) => d,
|
||||||
Err(e) => return format!("Error: failed to parse date '{}': {}", date_str, e),
|
Err(e) => return format!("Error: failed to parse date '{}': {}", date_str, e),
|
||||||
};
|
};
|
||||||
let timestamp = date.and_hms_opt(12, 0, 0).unwrap().and_utc().timestamp();
|
let days_radius = args
|
||||||
|
.get("days_radius")
|
||||||
|
.and_then(|v| v.as_i64())
|
||||||
|
.unwrap_or(0);
|
||||||
|
let zone = self.display_zone();
|
||||||
|
|
||||||
log::info!(
|
log::info!(
|
||||||
"tool_get_location_history: date={}, days_radius={}",
|
"tool_get_location_history: date={}, days_radius={}, apollo={}",
|
||||||
date,
|
date,
|
||||||
days_radius
|
days_radius,
|
||||||
|
self.apollo_client.is_enabled()
|
||||||
);
|
);
|
||||||
|
|
||||||
let start_ts = timestamp - (days_radius * 86400);
|
if self.apollo_client.is_enabled() {
|
||||||
let end_ts = timestamp + (days_radius * 86400);
|
let r = days_radius.clamp(0, 1);
|
||||||
|
let mut days = Vec::new();
|
||||||
|
for d in -r..=r {
|
||||||
|
let day = date + chrono::Duration::days(d);
|
||||||
|
let day_str = day.format("%Y-%m-%d").to_string();
|
||||||
|
if let Some(t) = self.apollo_client.location_day(&day_str, zone.name()).await {
|
||||||
|
days.push(format_apollo_day(&day_str, &t));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !days.is_empty() {
|
||||||
|
return format!(
|
||||||
|
"Location timeline ({}):\n\n{}",
|
||||||
|
zone.name(),
|
||||||
|
days.join("\n\n")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
// Every call failed — fall through to the local table.
|
||||||
|
}
|
||||||
|
|
||||||
|
let r = days_radius.clamp(0, 30);
|
||||||
|
let midnight = |d: NaiveDate| {
|
||||||
|
zone.from_local_datetime(&d.and_hms_opt(0, 0, 0).unwrap())
|
||||||
|
.earliest()
|
||||||
|
.map(|t| t.timestamp())
|
||||||
|
};
|
||||||
|
let (Some(start_ts), Some(end_ts)) = (
|
||||||
|
midnight(date - chrono::Duration::days(r)),
|
||||||
|
midnight(date + chrono::Duration::days(r + 1)),
|
||||||
|
) else {
|
||||||
|
return format!("Error: could not resolve local midnight for {}", date);
|
||||||
|
};
|
||||||
|
|
||||||
let locations = {
|
let locations = {
|
||||||
let mut dao = self
|
let mut dao = self
|
||||||
.location_dao
|
.location_dao
|
||||||
.lock()
|
.lock()
|
||||||
.expect("Unable to lock LocationHistoryDao");
|
.expect("Unable to lock LocationHistoryDao");
|
||||||
dao.find_locations_in_range(start_ts, end_ts).ok()
|
dao.find_locations_in_range(start_ts, end_ts - 1).ok()
|
||||||
};
|
};
|
||||||
|
|
||||||
match locations {
|
match locations {
|
||||||
Some(locs) if !locs.is_empty() => {
|
Some(locs) if !locs.is_empty() => {
|
||||||
let formatted: Vec<String> = locs
|
let formatted: Vec<String> = spread_evenly(&locs, LOCATION_HISTORY_DISPLAY_LIMIT)
|
||||||
.iter()
|
.into_iter()
|
||||||
.take(LOCATION_HISTORY_DISPLAY_LIMIT)
|
|
||||||
.map(|loc| {
|
.map(|loc| {
|
||||||
let dt = DateTime::from_timestamp(loc.timestamp, 0)
|
let dt = DateTime::from_timestamp(loc.timestamp, 0)
|
||||||
.map(|dt| {
|
.map(|dt| dt.with_timezone(&zone).format("%Y-%m-%d %H:%M").to_string())
|
||||||
dt.with_timezone(&Local)
|
|
||||||
.format("%Y-%m-%d %H:%M")
|
|
||||||
.to_string()
|
|
||||||
})
|
|
||||||
.unwrap_or_else(|| "unknown".to_string());
|
.unwrap_or_else(|| "unknown".to_string());
|
||||||
let activity = loc
|
let activity = loc
|
||||||
.activity
|
.activity
|
||||||
@@ -2403,14 +2558,18 @@ Return ONLY the summary, nothing else."#,
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
|
let header = if formatted.len() < locs.len() {
|
||||||
format!(
|
format!(
|
||||||
"{}\n{}",
|
"Found {} location records (showing {}, evenly spaced across the window):",
|
||||||
Self::found_header(locs.len(), formatted.len(), "location records"),
|
locs.len(),
|
||||||
formatted.join("\n")
|
formatted.len()
|
||||||
)
|
)
|
||||||
|
} else {
|
||||||
|
Self::found_header(locs.len(), formatted.len(), "location records")
|
||||||
|
};
|
||||||
|
format!("{}\n{}", header, formatted.join("\n"))
|
||||||
}
|
}
|
||||||
Some(_) => "No location history found.".to_string(),
|
_ => "No location history found.".to_string(),
|
||||||
None => "No location history found.".to_string(),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -3406,15 +3565,16 @@ Return ONLY the summary, nothing else."#,
|
|||||||
if opts.location_history_present {
|
if opts.location_history_present {
|
||||||
tools.push(Tool::function(
|
tools.push(Tool::function(
|
||||||
"get_location_history",
|
"get_location_history",
|
||||||
"Fetch raw location records (lat/lon/timestamp/activity) near a date. The default 14-day radius is \
|
"Where the user was on a date: a timeline of named places (Home, Work, …) with arrival/departure \
|
||||||
wide because location density varies; tighten to ±1 day for a single-trip query. For a coordinate's \
|
times, stops, travel with distances, and gaps with no data — or, without Apollo, raw location \
|
||||||
named place, use `reverse_geocode` (or `get_personal_place_at` when Apollo is enabled).",
|
records spread across the day. Times are local. For a coordinate's named place, use \
|
||||||
|
`get_personal_place_at` (Apollo) or `reverse_geocode`.",
|
||||||
serde_json::json!({
|
serde_json::json!({
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"required": ["date"],
|
"required": ["date"],
|
||||||
"properties": {
|
"properties": {
|
||||||
"date": { "type": "string", "description": "Center date, YYYY-MM-DD." },
|
"date": { "type": "string", "description": "The day, YYYY-MM-DD (local)." },
|
||||||
"days_radius": { "type": "integer", "description": "Days before and after to include (default 14)." }
|
"days_radius": { "type": "integer", "description": "Also include this many days before and after (default 0; at most 1 with Apollo)." }
|
||||||
}
|
}
|
||||||
}),
|
}),
|
||||||
));
|
));
|
||||||
@@ -4948,6 +5108,85 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn apollo_at(json: &str) -> LocationAt {
|
||||||
|
serde_json::from_str(json).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
const NY: chrono_tz::Tz = chrono_tz::Tz::America__New_York;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn apollo_location_no_sample_says_nothing() {
|
||||||
|
let at = apollo_at(r#"{"ts": 0, "sample": null, "nearest_offset_s": 7200}"#);
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&at, None, NY, Some("Medina, OH".into())),
|
||||||
|
None
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn apollo_location_renders_stay_in_display_zone() {
|
||||||
|
// 2017-01-17 08:07 / 16:19 EST
|
||||||
|
let at = apollo_at(
|
||||||
|
r#"{"ts": 1484669220,
|
||||||
|
"sample": {"ts": 1484669240, "lat": 41.469, "lon": -81.934, "activity": "TILTING", "offset_s": 20},
|
||||||
|
"place": {"id": 7, "name": "Hyland Software", "radius_m": 250},
|
||||||
|
"stay": {"place_id": 7, "name": "Hyland Software", "t0": 1484658450, "t1": 1484687985}}"#,
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&at, None, NY, None).as_deref(),
|
||||||
|
Some("Location history: You were at Hyland Software (08:07–16:19)")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn apollo_location_unnamed_uses_nominatim_activity_and_offset() {
|
||||||
|
let at = apollo_at(
|
||||||
|
r#"{"ts": 1000,
|
||||||
|
"sample": {"ts": 100, "lat": 41.1, "lon": -81.8, "activity": "IN_VEHICLE", "offset_s": -900}}"#,
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&at, None, NY, Some("Medina, OH".into())).as_deref(),
|
||||||
|
Some(
|
||||||
|
"Location history: You were near Medina, OH (IN_VEHICLE) [location fix 15 min before the photo]"
|
||||||
|
)
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&at, None, NY, None).as_deref(),
|
||||||
|
Some(
|
||||||
|
"Location history: You were near 41.1000, -81.8000 (IN_VEHICLE) [location fix 15 min before the photo]"
|
||||||
|
)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn apollo_location_skips_what_exif_already_says() {
|
||||||
|
let plain = apollo_at(
|
||||||
|
r#"{"ts": 0, "sample": {"ts": 0, "lat": 41.1, "lon": -81.8, "activity": "UNKNOWN", "offset_s": 0},
|
||||||
|
"place": {"id": 1, "name": "Park", "radius_m": 300}}"#,
|
||||||
|
);
|
||||||
|
// EXIF 100 m away, no stay, no informative activity → nothing new.
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&plain, Some((41.1009, -81.8)), NY, None),
|
||||||
|
None
|
||||||
|
);
|
||||||
|
// Without EXIF the same answer is worth a line.
|
||||||
|
assert_eq!(
|
||||||
|
render_apollo_location(&plain, None, NY, None).as_deref(),
|
||||||
|
Some("Location history: You were at Park")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn spread_evenly_keeps_ends_and_order() {
|
||||||
|
let v: Vec<i32> = (0..100).collect();
|
||||||
|
let picked: Vec<i32> = spread_evenly(&v, 20).into_iter().copied().collect();
|
||||||
|
assert_eq!(picked.len(), 20);
|
||||||
|
assert_eq!((picked[0], picked[19]), (0, 99));
|
||||||
|
assert!(picked.windows(2).all(|w| w[0] < w[1]));
|
||||||
|
assert_eq!(spread_evenly(&v[..5], 20).len(), 5);
|
||||||
|
assert_eq!(spread_evenly(&v, 1), vec![&0]);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn found_header_labels_truncation_honestly() {
|
fn found_header_labels_truncation_honestly() {
|
||||||
// Truncated: total exceeds what's listed below the header.
|
// Truncated: total exceeds what's listed below the header.
|
||||||
|
|||||||
@@ -227,7 +227,7 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
}
|
}
|
||||||
let _ = call_start;
|
let _ = call_start;
|
||||||
}
|
}
|
||||||
Err(ClipError::Permanent(e)) => {
|
Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
|
||||||
perm_fail += 1;
|
perm_fail += 1;
|
||||||
warn!("permanent encode failure for {rel_path}: {e}");
|
warn!("permanent encode failure for {rel_path}: {e}");
|
||||||
}
|
}
|
||||||
|
|||||||
+3
-1
@@ -172,7 +172,9 @@ pub async fn score_photos(
|
|||||||
// 1. Encode the query text. Fast — Apollo's text encoder is ~50ms on CPU.
|
// 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 {
|
let query_resp = match state.clip_client.encode_text(q_text).await {
|
||||||
Ok(r) => r,
|
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::Transient(e)) => return Err(ScoreError::Unavailable(e.to_string())),
|
||||||
Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
|
Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
|
||||||
};
|
};
|
||||||
|
|||||||
+28
-14
@@ -6,12 +6,12 @@
|
|||||||
//! parallel encode calls to Apollo, and persist the resulting embeddings
|
//! parallel encode calls to Apollo, and persist the resulting embeddings
|
||||||
//! into `image_exif.clip_embedding` / `clip_model_version`.
|
//! into `image_exif.clip_embedding` / `clip_model_version`.
|
||||||
//!
|
//!
|
||||||
//! Unlike the face pipeline, CLIP has no marker rows — a permanent
|
//! A permanent failure (un-decodable bytes) records a `clip_failures`
|
||||||
//! failure (un-decodable bytes) leaves the row's `clip_embedding` NULL
|
//! marker scoped to the row's `content_hash`, so the drain stops
|
||||||
//! and the drain will retry on the next tick. For personal-library
|
//! re-sending the file (and re-logging the same 422) every pass until
|
||||||
//! scale this is fine; the per-tick cap bounds the wasted work, and
|
//! its bytes change. `file_types::is_image_file` filters out videos /
|
||||||
//! `file_types::is_image_file` filters out videos / non-media client-
|
//! non-media client-side, so most permanent failures are corrupt files
|
||||||
//! side so most permanent failures are decoded-but-corrupt files (rare).
|
//! or non-images with an image extension.
|
||||||
//!
|
//!
|
||||||
//! The watcher thread isn't in any pre-existing async context, so we
|
//! 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
|
//! 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
|
cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
Err(ClipError::Permanent(e)) => {
|
Err(ClipError::Rejected(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.
|
|
||||||
warn!(
|
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
|
cand.rel_path, e
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -551,6 +551,17 @@ pub trait ExifDao: Sync + Send {
|
|||||||
model_version: &str,
|
model_version: &str,
|
||||||
) -> Result<(), DbError>;
|
) -> 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
|
/// Load every `(content_hash, clip_embedding)` pair from the live
|
||||||
/// image_exif rows for the given libraries, optionally filtered to a
|
/// image_exif rows for the given libraries, optionally filtered to a
|
||||||
/// single `model_version` (cosine sim across mixed geometries is
|
/// 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)
|
// (clip_embedding IS NULL AND content_hash IS NOT NULL)
|
||||||
// filter; the planner hits it directly. ORDER BY id ASC
|
// filter; the planner hits it directly. ORDER BY id ASC
|
||||||
// keeps drain progress monotone across ticks.
|
// 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
|
image_exif
|
||||||
.filter(library_id.eq(library_id_val))
|
.filter(library_id.eq(library_id_val))
|
||||||
.filter(clip_embedding.is_null())
|
.filter(clip_embedding.is_null())
|
||||||
.filter(content_hash.is_not_null())
|
.filter(content_hash.is_not_null())
|
||||||
|
.filter(diesel::dsl::not(diesel::dsl::exists(failed)))
|
||||||
.select((rel_path, content_hash.assume_not_null()))
|
.select((rel_path, content_hash.assume_not_null()))
|
||||||
.order(id.asc())
|
.order(id.asc())
|
||||||
.limit(limit)
|
.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(
|
fn list_clip_index(
|
||||||
&mut self,
|
&mut self,
|
||||||
library_ids_val: &[i32],
|
library_ids_val: &[i32],
|
||||||
@@ -2559,6 +2616,37 @@ mod exif_dao_tests {
|
|||||||
SqliteExifDao::from_connection(conn)
|
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]
|
#[test]
|
||||||
fn get_all_with_date_taken_union_returns_all_libraries() {
|
fn get_all_with_date_taken_union_returns_all_libraries() {
|
||||||
let mut dao = setup_two_libraries();
|
let mut dao = setup_two_libraries();
|
||||||
|
|||||||
@@ -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! {
|
diesel::table! {
|
||||||
daily_conversation_summaries (id) {
|
daily_conversation_summaries (id) {
|
||||||
id -> Integer,
|
id -> Integer,
|
||||||
@@ -348,6 +358,7 @@ diesel::table! {
|
|||||||
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
|
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
|
||||||
diesel::joinable!(entity_photo_links -> entities (entity_id));
|
diesel::joinable!(entity_photo_links -> entities (entity_id));
|
||||||
diesel::joinable!(entity_photo_links -> libraries (library_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 -> libraries (library_id));
|
||||||
diesel::joinable!(face_detections -> persons (person_id));
|
diesel::joinable!(face_detections -> persons (person_id));
|
||||||
diesel::joinable!(image_exif -> libraries (library_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!(
|
diesel::allow_tables_to_appear_in_same_query!(
|
||||||
calendar_events,
|
calendar_events,
|
||||||
|
clip_failures,
|
||||||
daily_conversation_summaries,
|
daily_conversation_summaries,
|
||||||
entities,
|
entities,
|
||||||
entity_facts,
|
entity_facts,
|
||||||
|
|||||||
@@ -1910,6 +1910,16 @@ mod tests {
|
|||||||
Ok(())
|
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(
|
fn list_clip_index(
|
||||||
&mut self,
|
&mut self,
|
||||||
_library_ids: &[i32],
|
_library_ids: &[i32],
|
||||||
|
|||||||
Reference in new issue
Block a user