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/// A measured duration for one stage's outstanding work, or `None` when there
202/// is no measurement to base one on. `secs` is `Some` only when a real prior
203/// run supplied a throughput; otherwise callers show the item count and an
204/// "intensive" marker rather than a number.
205#[derive(Debug, Clone, PartialEq, serde::Serialize)]
206pub struct CostEstimate {
207    pub secs: Option<u64>,
208    pub approximate: bool,
209}
210
211/// A duration for the outstanding work, but only when it can be measured. We
212/// deliberately do NOT fabricate one from a per-item constant: throughput
213/// swings by orders of magnitude across hardware (GPU vs CPU), batch size and
214/// media type, so a guessed duration is confidently wrong more often than it is
215/// useful, and a wrong number erodes trust in everything else the tool reports.
216/// `secs` is `Some` only when the last successful run recorded both how long it
217/// took and how many items it processed; until `pipeline_runs` stores item
218/// counts (see the item-count work) that is never, so today this returns `None`
219/// and callers render the count plus an "intensive" marker. `None` also when
220/// nothing is outstanding.
221pub fn estimate_cost(
222    outstanding: i64,
223    last_run_ms: Option<i64>,
224    last_run_items: Option<i64>,
225) -> CostEstimate {
226    let secs = match (last_run_ms, last_run_items) {
227        (Some(ms), Some(items)) if outstanding > 0 && ms > 0 && items > 0 => {
228            let per_item = ms as f64 / items as f64 / 1000.0;
229            Some((outstanding as f64 * per_item).ceil() as u64)
230        }
231        _ => None,
232    };
233    CostEstimate {
234        secs,
235        approximate: true,
236    }
237}
238
239/// The whole operational picture for one library, from one call. Rendered
240/// by `videre status` (text and --json); `stats` and the MCP tools read the
241/// same model rather than re-deriving it.
242#[derive(Debug, Clone, PartialEq, serde::Serialize)]
243pub struct StatusReport {
244    pub coverage: Vec<StageCoverage>,
245    pub pipelines: Vec<crate::pipeline_runs::PipelineRunStatus>,
246    pub watch: WatchLiveness,
247    /// Per-stage cost estimate for the outstanding work, in the same stage
248    /// order as `coverage` entries that have outstanding work.
249    pub costs: Vec<(&'static str, CostEstimate)>,
250    /// The embedding model the coverage numbers were measured against.
251    pub embed_model: String,
252}
253
254impl StatusReport {
255    /// True only when a pipeline run actually failed or crashed. Staleness
256    /// is informational and never a failure: a library mid-setup is healthy
257    /// (spec decision D5).
258    pub fn has_problem(&self) -> bool {
259        self.pipelines
260            .iter()
261            .any(|p| matches!(p.status.as_deref(), Some("failed") | Some("crashed")))
262    }
263}
264
265/// Assemble the report: coverage, pipeline health, watch liveness, and the
266/// cost of closing each gap. One call, one connection, read-only.
267pub fn compute_status_in(
268    conn: &Connection,
269    ctx: &crate::library::LibraryContext,
270) -> Result<StatusReport> {
271    let embed_model = ctx.settings.default_model.clone();
272    let coverage = coverage_in(conn, &embed_model, &embed_model)?;
273    let pipelines = crate::pipeline_runs::read_all_in(conn, ctx)?;
274    let watch = watch_liveness_in(conn, ctx)?;
275    let costs = coverage
276        .iter()
277        .filter(|c| c.outstanding > 0)
278        .map(|c| {
279            let prior = pipelines.iter().find(|p| p.command == c.stage);
280            (
281                c.stage,
282                // pipeline_runs records a duration but not an item count yet, so
283                // the second input is None and the estimate stays None until
284                // that lands. No fabricated fallback.
285                estimate_cost(c.outstanding, prior.and_then(|p| p.duration_ms), None),
286            )
287        })
288        .collect();
289    Ok(StatusReport {
290        coverage,
291        pipelines,
292        watch,
293        costs,
294        embed_model,
295    })
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301
302    const TEST_MODEL: &str = "test/model";
303
304    /// A main database with `file_hashes`, plus a real attached model
305    /// database (mirroring production's split), and the faces tables.
306    fn seed_db(tag: &str) -> Connection {
307        let ctx = crate::embeddings_db::test_context(tag);
308        let conn = Connection::open_in_memory().unwrap();
309        conn.execute_batch(
310            "CREATE TABLE file_hashes (
311                path        TEXT PRIMARY KEY,
312                hash        TEXT NOT NULL,
313                mime        TEXT,
314                size_bytes  INTEGER,
315                created_at  TEXT,
316                modified_at TEXT,
317                ext         TEXT,
318                phash       INTEGER,
319                exif_date   TEXT,
320                gps_lat     REAL,
321                gps_lon     REAL,
322                width       INTEGER,
323                height      INTEGER,
324                location_name TEXT
325            );",
326        )
327        .unwrap();
328        conn.execute_batch(
329            "CREATE TABLE IF NOT EXISTS classifications (
330                model_id      TEXT NOT NULL,
331                hash          TEXT NOT NULL,
332                category      TEXT NOT NULL,
333                confidence    REAL NOT NULL,
334                classified_at TEXT NOT NULL,
335                PRIMARY KEY (model_id, hash)
336            );",
337        )
338        .unwrap();
339        crate::embeddings_db::attach_in(&conn, &ctx, TEST_MODEL, true).unwrap();
340        crate::face_db::create_faces_table(&conn).unwrap();
341        conn
342    }
343
344    fn insert_file(conn: &Connection, path: &str, hash: &str, ext: &str) {
345        conn.execute(
346            "INSERT INTO file_hashes (path, hash, ext) VALUES (?1, ?2, ?3)",
347            [path, hash, ext],
348        )
349        .unwrap();
350    }
351
352    #[test]
353    fn coverage_counts_outstanding_per_stage() {
354        let conn = seed_db("status_cov_main");
355        // 3 embeddable images: 1 embedded+classified, 1 embedded, 1 neither.
356        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
357        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
358        insert_file(&conn, "/a/3.jpg", "h3", "jpg");
359        crate::embeddings::insert_embeddings(
360            &conn,
361            TEST_MODEL,
362            &[("h1".into(), vec![0u8; 4]), ("h2".into(), vec![0u8; 4])],
363        )
364        .unwrap();
365        conn.execute(
366            "INSERT INTO classifications (model_id, hash, category, confidence, classified_at)
367             VALUES ('test/model', 'h1', 'cat', 0.9, 'now')",
368            [],
369        )
370        .unwrap();
371
372        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
373        let embed = cov.iter().find(|c| c.stage == "embed").unwrap();
374        assert_eq!(embed.outstanding, 1, "h3 is the only un-embedded image");
375        assert_eq!(embed.total, 3);
376        assert!(embed.heavy);
377        assert_eq!(embed.next_command, Some("videre embed"));
378
379        let classify = cov.iter().find(|c| c.stage == "classify").unwrap();
380        assert_eq!(classify.outstanding, 1, "h2 is embedded but unclassified");
381        assert_eq!(classify.total, 2);
382    }
383
384    #[test]
385    fn faces_coverage_counts_unscanned_not_faceless() {
386        let conn = seed_db("status_cov_faces");
387        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
388        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
389        insert_file(&conn, "/a/3.png", "h3", "png");
390        insert_file(&conn, "/a/v.mp4", "h4", "mp4"); // not faces-eligible
391                                                     // h1 was scanned and found faceless: still done. h2 has faces: done.
392        conn.execute("INSERT INTO faces_scanned (hash) VALUES ('h1')", [])
393            .unwrap();
394        conn.execute(
395            "INSERT INTO faces (hash, bbox, embedding) VALUES ('h2', '0,0,10,10', X'00')",
396            [],
397        )
398        .unwrap();
399
400        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
401        let faces = cov.iter().find(|c| c.stage == "faces").unwrap();
402        assert_eq!(faces.total, 3, "videos are not faces-eligible");
403        assert_eq!(faces.outstanding, 1, "only h3 was never tried");
404        assert!(!faces.heavy);
405    }
406
407    #[test]
408    fn locations_and_fix_dates_count_only_true_gaps() {
409        let conn = seed_db("status_cov_locfix");
410        // "In sync" is judged against THIS machine's target: the transform
411        // resolves the camera-local time through the local timezone, so a
412        // hardcoded offset passes on one continent and fails on another (it
413        // did, on CI). Rows in sync here carry exactly what
414        // `target_modified_at` would write.
415        let exif = "2021-07-04T15:30:00";
416        let in_sync = crate::fix_dates_target::target_modified_at(exif).unwrap();
417        // geotagged, named: done. geotagged, unnamed: outstanding.
418        conn.execute(
419            "INSERT INTO file_hashes (path, hash, ext, gps_lat, gps_lon, location_name,
420                                      exif_date, modified_at)
421             VALUES ('/a/1.jpg', 'h1', 'jpg', 52.5, 13.4, 'Berlin, Germany', ?1, ?2)",
422            [exif.to_string(), in_sync.clone()],
423        )
424        .unwrap();
425        conn.execute(
426            "INSERT INTO file_hashes (path, hash, ext, gps_lat, gps_lon, exif_date, modified_at)
427             VALUES ('/a/2.jpg', 'h2', 'jpg', 48.8, 2.3, '2020-01-02T03:04:05', '1970-01-01T00:00:00+00:00')",
428            [],
429        )
430        .unwrap();
431        conn.execute(
432            "INSERT INTO file_hashes (path, hash, ext, exif_date, modified_at)
433             VALUES ('/a/3.jpg', 'h3', 'jpg', ?1, ?2)",
434            [exif.to_string(), in_sync],
435        )
436        .unwrap();
437
438        let cov = coverage_in(&conn, TEST_MODEL, TEST_MODEL).unwrap();
439        let locations = cov.iter().find(|c| c.stage == "locations").unwrap();
440        assert_eq!(locations.total, 2);
441        assert_eq!(locations.outstanding, 1, "h2 has GPS and no place name");
442
443        let fix = cov.iter().find(|c| c.stage == "fix-dates").unwrap();
444        assert_eq!(fix.total, 3);
445        assert_eq!(
446            fix.outstanding, 1,
447            "only h2's mtime disagrees with its exif_date"
448        );
449    }
450
451    #[test]
452    fn cost_is_shown_only_when_it_can_be_measured() {
453        // 100 items took 50_000ms -> 500ms/item; 10 outstanding -> ~5s.
454        let c = estimate_cost(10, Some(50_000), Some(100));
455        assert_eq!(c.secs, Some(5));
456        assert!(c.approximate);
457        // No prior run with an item count: no fabricated estimate.
458        assert_eq!(estimate_cost(10, None, None).secs, None);
459        // A duration but no item count (today's pipeline_runs): still None.
460        assert_eq!(estimate_cost(10, Some(50_000), None).secs, None);
461        // Nothing outstanding: no estimate.
462        assert_eq!(estimate_cost(0, Some(50_000), Some(100)).secs, None);
463        // A nonsense prior run (zero items) must not divide by zero -> None.
464        assert_eq!(estimate_cost(10, Some(50_000), Some(0)).secs, None);
465    }
466
467    #[test]
468    fn compute_status_assembles_the_whole_report() {
469        let ctx = crate::embeddings_db::test_context("status_compute");
470        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
471        let conn = seed_db("status_compute");
472        insert_file(&conn, "/a/1.jpg", "h1", "jpg");
473        insert_file(&conn, "/a/2.jpg", "h2", "jpg");
474
475        let report = compute_status_in(&conn, &ctx).unwrap();
476        assert_eq!(report.embed_model, ctx.settings.default_model);
477        assert!(!report.coverage.is_empty());
478        let embed = report.coverage.iter().find(|c| c.stage == "embed").unwrap();
479        assert_eq!(embed.outstanding, 2);
480        // embed has an entry, but with no measured prior run there is no
481        // fabricated duration.
482        let embed_cost = report
483            .costs
484            .iter()
485            .find(|(stage, _)| *stage == "embed")
486            .expect("embed cost entry");
487        assert_eq!(embed_cost.1.secs, None);
488        assert!(report.pipelines.iter().all(|p| p.status.is_none()));
489        assert!(!report.watch.running);
490        assert_eq!(report.watch.last_cycle_at, None);
491        assert!(
492            !report.has_problem(),
493            "a fresh library is healthy, not failing"
494        );
495
496        // A failed run is what flips --check, and staleness never does.
497        crate::pipeline_runs::start_run(&conn, "faces").unwrap();
498        crate::pipeline_runs::finish_run(&conn, "faces", "failed", 5, Some("boom")).unwrap();
499        let report = compute_status_in(&conn, &ctx).unwrap();
500        assert!(report.has_problem());
501    }
502}
videre_core/status_report.rs

videre_core/
status_report.rs
