videre_core/
status_report.rs

1//! The shared status model: one producer for everything that reports
2//! pipeline state, so the CLI, `stats`, and the MCP tools cannot drift
3//! into telling different stories about the same library (the duplication
4//! this closes was flagged as DEBT:8).
5//!
6//! `status` owns operational health: per-stage coverage, pipeline run
7//! health, watch liveness, and the next action + cost per gap. Inventory
8//! (what is IN the library) stays with `videre stats` and
9//! `library_stats::compute_full_in`; this module composes those outputs
10//! rather than re-deriving them.
11
12use anyhow::Result;
13use rusqlite::Connection;
14use rusqlite::OptionalExtension;
15
16/// One stage's outstanding-vs-done shape. Stages without a true
17/// outstanding-vs-done count (scan, dedupe) deliberately get no line here:
18/// `status` stays glanceable rather than becoming an everything-dashboard.
19#[derive(Debug, Clone, PartialEq, serde::Serialize)]
20pub struct StageCoverage {
21    pub stage: &'static str,
22    pub outstanding: i64,
23    pub total: i64,
24    /// The command that closes this gap, when there is one.
25    pub next_command: Option<&'static str>,
26    /// Embed and classify are optional, hours-long stages; a newcomer should
27    /// not read "behind" on a stage they may never want (spec decision D4).
28    pub heavy: bool,
29}
30
31/// Faces-eligible hashes: the same population `videre faces` walks (image
32/// extensions), deduplicated by hash with a deterministic representative.
33fn faces_eligible_hashes(conn: &Connection) -> Result<Vec<String>> {
34    let mut stmt = conn.prepare(
35        "SELECT hash FROM file_hashes
36         WHERE lower(COALESCE(ext, '')) IN
37               ('jpg','jpeg','png','gif','webp','bmp','tiff','heic')
38         GROUP BY hash",
39    )?;
40    let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
41    rows.collect::<std::result::Result<Vec<String>, _>>()
42        .map_err(Into::into)
43}
44
45/// Embed stage: outstanding under the active model of everything eligible.
46fn embed_coverage(conn: &Connection, embed_model: &str) -> Result<StageCoverage> {
47    let total = crate::embeddings::embeddable_images(conn, embed_model)?.len() as i64;
48    let outstanding = match crate::embeddings::pending_images(conn, embed_model)? {
49        v => v.len() as i64,
50    };
51    Ok(StageCoverage {
52        stage: "embed",
53        outstanding,
54        total,
55        next_command: Some("videre embed"),
56        heavy: true,
57    })
58}
59
60/// Classify stage: embedded hashes under the model missing a classification.
61fn classify_coverage(conn: &Connection, classify_model: &str) -> Result<StageCoverage> {
62    let total: i64 = conn.query_row(
63        "SELECT COUNT(*) FROM emb.embeddings WHERE model_id = ?1",
64        [classify_model],
65        |r| r.get(0),
66    )?;
67    let outstanding = crate::classify::pending_hashes(conn, classify_model)?.len() as i64;
68    Ok(StageCoverage {
69        stage: "classify",
70        outstanding,
71        total,
72        next_command: Some("videre classify"),
73        heavy: true,
74    })
75}
76
77/// Faces stage: eligible hashes with no detection record at all. The skip
78/// set is "already tried" (including images where zero faces were found),
79/// which is why unscanned - not faceless - is the outstanding shape.
80fn faces_coverage(conn: &Connection) -> Result<StageCoverage> {
81    let eligible = faces_eligible_hashes(conn)?;
82    let scanned: std::collections::HashSet<String> =
83        crate::face_db::scanned_hashes(conn)?.into_iter().collect();
84    let with_faces: std::collections::HashSet<String> = crate::face_db::hashes_with_faces(conn)?
85        .into_iter()
86        .collect();
87    let outstanding = eligible
88        .iter()
89        .filter(|h| !scanned.contains(*h) && !with_faces.contains(*h))
90        .count() as i64;
91    Ok(StageCoverage {
92        stage: "faces",
93        outstanding,
94        total: eligible.len() as i64,
95        next_command: Some("videre faces"),
96        heavy: false,
97    })
98}
99
100/// Locations stage: geotagged photos with no place name yet. Grouping is a
101/// global recompute, so the count is informational; the command line names
102/// what closes it.
103fn locations_coverage(conn: &Connection) -> Result<StageCoverage> {
104    let total: i64 = conn.query_row(
105        "SELECT COUNT(*) FROM file_hashes WHERE gps_lat IS NOT NULL AND gps_lon IS NOT NULL",
106        [],
107        |r| r.get(0),
108    )?;
109    let outstanding: i64 = conn.query_row(
110        "SELECT COUNT(*) FROM file_hashes
111         WHERE gps_lat IS NOT NULL AND gps_lon IS NOT NULL AND location_name IS NULL",
112        [],
113        |r| r.get(0),
114    )?;
115    Ok(StageCoverage {
116        stage: "locations",
117        outstanding,
118        total,
119        next_command: Some("videre locations"),
120        heavy: false,
121    })
122}
123
124/// Fix-dates stage: rows with a camera date whose stored modified time does
125/// not match what fix-dates would write. Judged in Rust because the target
126/// depends on the local timezone, which SQL cannot compute.
127fn fix_dates_coverage(conn: &Connection) -> Result<StageCoverage> {
128    let mut stmt =
129        conn.prepare("SELECT exif_date, modified_at FROM file_hashes WHERE exif_date IS NOT NULL")?;
130    let rows = stmt.query_map([], |r| {
131        Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?))
132    })?;
133    let mut total = 0i64;
134    let mut outstanding = 0i64;
135    for row in rows {
136        let (exif_date, modified_at) = row?;
137        total += 1;
138        if let Some(target) = crate::fix_dates_target::target_modified_at(&exif_date) {
139            if modified_at.as_deref() != Some(target.as_str()) {
140                outstanding += 1;
141            }
142        }
143    }
144    Ok(StageCoverage {
145        stage: "fix-dates",
146        outstanding,
147        total,
148        next_command: Some("videre fix-dates"),
149        heavy: false,
150    })
151}
152
153/// Coverage for every stage that has a true outstanding-vs-done count.
154/// One aggregate query per stage; no per-file work outside the fix-dates
155/// transform, which is in-memory arithmetic over already-stored strings.
156pub fn coverage_in(
157    conn: &Connection,
158    embed_model: &str,
159    classify_model: &str,
160) -> Result<Vec<StageCoverage>> {
161    Ok(vec![
162        embed_coverage(conn, embed_model)?,
163        classify_coverage(conn, classify_model)?,
164        faces_coverage(conn)?,
165        locations_coverage(conn)?,
166        fix_dates_coverage(conn)?,
167    ])
168}
169
170/// Whether a watcher is alive, and when its last cycle completed. Running-
171/// ness comes from the watch lock; the last-cycle time from the heartbeat
172/// row `videre watch` writes at the end of each successful cycle. A watcher
173/// that died mid-cycle leaves the previous heartbeat, so it reads as a stale
174/// last-cycle time rather than pretending nothing is wrong.
175#[derive(Debug, Clone, PartialEq, serde::Serialize)]
176pub struct WatchLiveness {
177    pub running: bool,
178    pub last_cycle_at: Option<String>,
179}
180
181/// Watch liveness for one library. Never errors on a library that never
182/// watched: that is the "never run" case, not a failure.
183pub fn watch_liveness_in(
184    conn: &Connection,
185    ctx: &crate::library::LibraryContext,
186) -> Result<WatchLiveness> {
187    crate::pipeline_runs::ensure_pipeline_runs_table(conn)?;
188    let last_cycle_at: Option<String> = conn
189        .query_row(
190            "SELECT started_at FROM pipeline_runs WHERE command = 'watch'",
191            [],
192            |r| r.get(0),
193        )
194        .optional()?;
195    Ok(WatchLiveness {
196        running: crate::library_locks::command_locked(ctx, "watch")?,
197        last_cycle_at,
198    })
199}
200
201/// An approximate duration for one stage's outstanding work, clearly a
202/// guess and rendered as one ("~1.6h"). `secs` is `None` when there is
203/// nothing outstanding: no work, no estimate.
204#[derive(Debug, Clone, PartialEq, serde::Serialize)]
205pub struct CostEstimate {
206    pub secs: Option<u64>,
207    pub approximate: bool,
208}
209
210/// Estimate cost from the last successful run's measured throughput when the
211/// run recorded an item count, else from a coarse per-item constant. The
212/// estimator is deliberately isolated: pipeline_runs does not yet store item
213/// counts, so today the measured path is exercised by tests and the fallback
214/// is what users see; when runs start recording items, this function
215/// improves without touching any caller.
216pub fn estimate_cost(
217    outstanding: i64,
218    last_run_ms: Option<i64>,
219    last_run_items: Option<i64>,
220    fallback_secs_per_item: f64,
221) -> CostEstimate {
222    let secs = if outstanding <= 0 {
223        None
224    } else {
225        let per_item = match (last_run_ms, last_run_items) {
226            (Some(ms), Some(items)) if ms > 0 && items > 0 => ms as f64 / items as f64 / 1000.0,
227            _ => fallback_secs_per_item,
228        };
229        Some((outstanding as f64 * per_item).ceil() as u64)
230    };
231    CostEstimate {
232        secs,
233        approximate: true,
234    }
235}
236
237/// The whole operational picture for one library, from one call. Rendered
238/// by `videre status` (text and --json); `stats` and the MCP tools read the
239/// same model rather than re-deriving it.
240#[derive(Debug, Clone, PartialEq, serde::Serialize)]
241pub struct StatusReport {
242    pub coverage: Vec<StageCoverage>,
243    pub pipelines: Vec<crate::pipeline_runs::PipelineRunStatus>,
244    pub watch: WatchLiveness,
245    /// Per-stage cost estimate for the outstanding work, in the same stage
246    /// order as `coverage` entries that have outstanding work.
247    pub costs: Vec<(&'static str, CostEstimate)>,
248    /// The embedding model the coverage numbers were measured against.
249    pub embed_model: String,
250}
251
252impl StatusReport {
253    /// True only when a pipeline run actually failed or crashed. Staleness
254    /// is informational and never a failure: a library mid-setup is healthy
255    /// (spec decision D5).
256    pub fn has_problem(&self) -> bool {
257        self.pipelines
258            .iter()
259            .any(|p| matches!(p.status.as_deref(), Some("failed") | Some("crashed")))
260    }
261}
262
263/// Fallback per-item seconds for stages with no measured history. Coarse by
264/// design and always rendered as approximate; embed/classify dominate real
265/// runs, so their constants err on the slow side of honest.
266fn fallback_secs_per_item(stage: &str) -> f64 {
267    match stage {
268        "embed" => 2.0,
269        "classify" => 0.05,
270        "faces" => 1.0,
271        "locations" => 0.001,
272        "fix-dates" => 0.001,
273        _ => 1.0,
274    }
275}
276
277/// Assemble the report: coverage, pipeline health, watch liveness, and the
278/// cost of closing each gap. One call, one connection, read-only.
279pub fn compute_status_in(
280    conn: &Connection,
281    ctx: &crate::library::LibraryContext,
282) -> Result<StatusReport> {
283    let embed_model = ctx.settings.default_model.clone();
284    let coverage = coverage_in(conn, &embed_model, &embed_model)?;
285    let pipelines = crate::pipeline_runs::read_all_in(conn, ctx)?;
286    let watch = watch_liveness_in(conn, ctx)?;
287    let costs = coverage
288        .iter()
289        .filter(|c| c.outstanding > 0)
290        .map(|c| {
291            let prior = pipelines.iter().find(|p| p.command == c.stage);
292            (
293                c.stage,
294                estimate_cost(
295                    c.outstanding,
296                    prior.and_then(|p| p.duration_ms),
297                    None,
298                    fallback_secs_per_item(c.stage),
299                ),
300            )
301        })
302        .collect();
303    Ok(StatusReport {
304        coverage,
305        pipelines,
306        watch,
307        costs,
308        embed_model,
309    })
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315
316    const TEST_MODEL: &str = "test/model";
317
318    /// A main database with `file_hashes`, plus a real attached model
319    /// database (mirroring production's split), and the faces tables.
320    fn seed_db(tag: &str) -> Connection {
321        let ctx = crate::embeddings_db::test_context(tag);
322        let conn = Connection::open_in_memory().unwrap();
323        conn.execute_batch(
324            "CREATE TABLE file_hashes (
325                path        TEXT PRIMARY KEY,
326                hash        TEXT NOT NULL,
327                mime        TEXT,
328                size_bytes  INTEGER,
329                created_at  TEXT,
330                modified_at TEXT,
331                ext         TEXT,
332                phash       INTEGER,
333                exif_date   TEXT,
334                gps_lat     REAL,
335                gps_lon     REAL,
336                width       INTEGER,
337                height      INTEGER,
338                location_name TEXT
339            );",
340        )
341        .unwrap();
342        conn.execute_batch(
343            "CREATE TABLE IF NOT EXISTS classifications (
344                model_id      TEXT NOT NULL,
345                hash          TEXT NOT NULL,
346                category      TEXT NOT NULL,
347                confidence    REAL NOT NULL,
348                classified_at TEXT NOT NULL,
349                PRIMARY KEY (model_id, hash)
350            );",
351        )
352        .unwrap();
353        crate::embeddings_db::attach_in(&conn, &ctx, TEST_MODEL, true).unwrap();
354        crate::face_db::create_faces_table(&conn).unwrap();
355        conn
356    }
357
358    fn insert_file(conn: &Connection, path: &str, hash: &str, ext: &str) {
359        conn.execute(
360            "INSERT INTO file_hashes (path, hash, ext) VALUES (?1, ?2, ?3)",
361            [path, hash, ext],
362        )
363        .unwrap();
364    }
365
366    #[test]
367    fn coverage_counts_outstanding_per_stage() {
368        let conn = seed_db("status_cov_main");
369        // 3 embeddable images: 1 embedded+classified, 1 embedded, 1 neither.
370        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
371        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
372        insert_file(&conn, "/a/3.jpg", "h3", "jpg");
373        crate::embeddings::insert_embeddings(
374            &conn,
375            TEST_MODEL,
376            &[("h1".into(), vec![0u8; 4]), ("h2".into(), vec![0u8; 4])],
377        )
378        .unwrap();
379        conn.execute(
380            "INSERT INTO classifications (model_id, hash, category, confidence, classified_at)
381             VALUES ('test/model', 'h1', 'cat', 0.9, 'now')",
382            [],
383        )
384        .unwrap();
385
386        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
387        let embed = cov.iter().find(|c| c.stage == "embed").unwrap();
388        assert_eq!(embed.outstanding, 1, "h3 is the only un-embedded image");
389        assert_eq!(embed.total, 3);
390        assert!(embed.heavy);
391        assert_eq!(embed.next_command, Some("videre embed"));
392
393        let classify = cov.iter().find(|c| c.stage == "classify").unwrap();
394        assert_eq!(classify.outstanding, 1, "h2 is embedded but unclassified");
395        assert_eq!(classify.total, 2);
396    }
397
398    #[test]
399    fn faces_coverage_counts_unscanned_not_faceless() {
400        let conn = seed_db("status_cov_faces");
401        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
402        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
403        insert_file(&conn, "/a/3.png", "h3", "png");
404        insert_file(&conn, "/a/v.mp4", "h4", "mp4"); // not faces-eligible
405                                                     // h1 was scanned and found faceless: still done. h2 has faces: done.
406        conn.execute("INSERT INTO faces_scanned (hash) VALUES ('h1')", [])
407            .unwrap();
408        conn.execute(
409            "INSERT INTO faces (hash, bbox, embedding) VALUES ('h2', '0,0,10,10', X'00')",
410            [],
411        )
412        .unwrap();
413
414        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
415        let faces = cov.iter().find(|c| c.stage == "faces").unwrap();
416        assert_eq!(faces.total, 3, "videos are not faces-eligible");
417        assert_eq!(faces.outstanding, 1, "only h3 was never tried");
418        assert!(!faces.heavy);
419    }
420
421    #[test]
422    fn locations_and_fix_dates_count_only_true_gaps() {
423        let conn = seed_db("status_cov_locfix");
424        // "In sync" is judged against THIS machine's target: the transform
425        // resolves the camera-local time through the local timezone, so a
426        // hardcoded offset passes on one continent and fails on another (it
427        // did, on CI). Rows in sync here carry exactly what
428        // `target_modified_at` would write.
429        let exif = "2021-07-04T15:30:00";
430        let in_sync = crate::fix_dates_target::target_modified_at(exif).unwrap();
431        // geotagged, named: done. geotagged, unnamed: outstanding.
432        conn.execute(
433            "INSERT INTO file_hashes (path, hash, ext, gps_lat, gps_lon, location_name,
434                                      exif_date, modified_at)
435             VALUES ('/a/1.jpg', 'h1', 'jpg', 52.5, 13.4, 'Berlin, Germany', ?1, ?2)",
436            [exif.to_string(), in_sync.clone()],
437        )
438        .unwrap();
439        conn.execute(
440            "INSERT INTO file_hashes (path, hash, ext, gps_lat, gps_lon, exif_date, modified_at)
441             VALUES ('/a/2.jpg', 'h2', 'jpg', 48.8, 2.3, '2020-01-02T03:04:05', '1970-01-01T00:00:00+00:00')",
442            [],
443        )
444        .unwrap();
445        conn.execute(
446            "INSERT INTO file_hashes (path, hash, ext, exif_date, modified_at)
447             VALUES ('/a/3.jpg', 'h3', 'jpg', ?1, ?2)",
448            [exif.to_string(), in_sync],
449        )
450        .unwrap();
451
452        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
453        let locations = cov.iter().find(|c| c.stage == "locations").unwrap();
454        assert_eq!(locations.total, 2);
455        assert_eq!(locations.outstanding, 1, "h2 has GPS and no place name");
456
457        let fix = cov.iter().find(|c| c.stage == "fix-dates").unwrap();
458        assert_eq!(fix.total, 3);
459        assert_eq!(
460            fix.outstanding, 1,
461            "only h2's mtime disagrees with its exif_date"
462        );
463    }
464
465    #[test]
466    fn cost_uses_measured_rate_then_falls_back() {
467        // 100 items took 50_000ms -> 500ms/item; 10 outstanding -> ~5s.
468        let c = estimate_cost(10, Some(50_000), Some(100), 2.0);
469        assert_eq!(c.secs, Some(5));
470        // no prior run -> fallback 2s/item * 10 = 20s.
471        let f = estimate_cost(10, None, None, 2.0);
472        assert_eq!(f.secs, Some(20));
473        assert!(f.approximate);
474        // nothing outstanding: no estimate, not zero seconds of work.
475        assert_eq!(estimate_cost(0, None, None, 2.0).secs, None);
476        // a nonsense prior run (zero items) must not divide by zero.
477        assert_eq!(estimate_cost(10, Some(50_000), Some(0), 2.0).secs, Some(20));
478    }
479
480    #[test]
481    fn compute_status_assembles_the_whole_report() {
482        let ctx = crate::embeddings_db::test_context("status_compute");
483        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
484        let conn = seed_db("status_compute");
485        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
486        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
487
488        let report = compute_status_in(&conn, &ctx).unwrap();
489        assert_eq!(report.embed_model, ctx.settings.default_model);
490        assert!(!report.coverage.is_empty());
491        let embed = report.coverage.iter().find(|c| c.stage == "embed").unwrap();
492        assert_eq!(embed.outstanding, 2);
493        assert!(report
494            .costs
495            .iter()
496            .any(|(stage, cost)| *stage == "embed" && cost.secs.is_some()));
497        assert!(report.pipelines.iter().all(|p| p.status.is_none()));
498        assert!(!report.watch.running);
499        assert_eq!(report.watch.last_cycle_at, None);
500        assert!(
501            !report.has_problem(),
502            "a fresh library is healthy, not failing"
503        );
504
505        // A failed run is what flips --check, and staleness never does.
506        crate::pipeline_runs::start_run(&conn, "faces").unwrap();
507        crate::pipeline_runs::finish_run(&conn, "faces", "failed", 5, Some("boom")).unwrap();
508        let report = compute_status_in(&conn, &ctx).unwrap();
509        assert!(report.has_problem());
510    }
511}
videre_core/status_report.rs

videre_core/
status_report.rs
