AI Flow

@@ -4,7 +4,7 @@ version = 4  [[package]] name = "adom-aiflow"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-analyze",  "aiflow-board",@@ -25,7 +25,7 @@ dependencies = [  [[package]] name = "aiflow-analyze"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "serde",@@ -35,7 +35,7 @@ dependencies = [  [[package]] name = "aiflow-board"-version = "0.2.67"+version = "0.2.68" dependencies = [  "roxmltree",  "serde",@@ -44,7 +44,7 @@ dependencies = [  [[package]] name = "aiflow-bridge"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "serde",@@ -53,7 +53,7 @@ dependencies = [  [[package]] name = "aiflow-copper"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "aiflow-grid",@@ -63,7 +63,7 @@ dependencies = [  [[package]] name = "aiflow-grid"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "serde",@@ -72,7 +72,7 @@ dependencies = [  [[package]] name = "aiflow-place"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "serde",@@ -81,7 +81,7 @@ dependencies = [  [[package]] name = "aiflow-pours"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "aiflow-copper",@@ -91,7 +91,7 @@ dependencies = [  [[package]] name = "aiflow-router"-version = "0.2.67"+version = "0.2.68" dependencies = [  "aiflow-board",  "aiflow-grid",@@ -101,7 +101,7 @@ dependencies = [  [[package]] name = "aiflow-run"-version = "0.2.67"+version = "0.2.68" dependencies = [  "serde",  "serde_json",

@@ -306,6 +306,10 @@ enum Cmd {     #[arg(long)] about: Option<String> },     /// `priorities`: the board's two rankings (issues #129, #130), spec.signalPriority (signal classes, each with nets and why; routed in that order) and spec.placementPriorities (thermal, EMC, routability with rank and why; `place check` scores in that order). Shows what the spec has, drafts what it lacks from the spec and the net names (a start to reorder and explain, never the answer), and scores the current board's placement. `--write <file>` saves the drafts as JSON to edit into the spec.     Priorities { #[arg(long)] write: Option<String> },+    /// The background media worker (issue #170): checks and cuts the clips queued in <run>/media/queue and writes+    /// the results to <run>/media/done for the next command to fold in. Started by AI Flow itself, one at a time.+    #[command(hide = true)]+    MediaWorker,     /// The end of the run as the human sees it: the moment you say "done, here is your video". Needs finish first and the video file.     Deliver { #[arg(long)] video: String, #[arg(long)] message: String, /// deliver even though a clip is flagged as having filmed nothing (say why in --message)     #[arg(long)] accept_suspect: bool },@@ -699,7 +703,9 @@ fn run_dir(cli: &Cli) -> PathBuf {  fn load_run(cli: &Cli) -> Run {     let d = run_dir(cli);-    if let Some(r) = Run::open(&d) {+    if let Some(mut r) = Run::open(&d) {+        // issue #170: the media worker's results come into the run here (every command but the stop hook)+        if !matches!(cli.cmd, Cmd::FinishGuard { .. } | Cmd::MediaWorker) && media_dir(&d).join("done").is_dir() { media_reconcile(&mut r); }         if let Some(b) = r.data["remoteBoard"].as_str() { aiflow_bridge::eda::set_run_board(b); }         // the run's desktop is where a remote KiCad runs, unless the human named another         if let (Some(t), Err(_)) = (r.data["target"].as_str(), std::env::var("KICAD_REMOTE_TARGET")) { std::env::set_var("KICAD_REMOTE_TARGET", t); }@@ -1064,6 +1070,9 @@ fn step_table(r: &Run) -> String {             names.push(k.clone());         }     }+    // issue #171: media processing (inside tool time) and owner hold (a pause) get their own columns when a run has any+    let extra = rep.values().any(|e| e["mediaSeconds"].as_f64().unwrap_or(0.0) > 0.0 || e["holdSeconds"].as_f64().unwrap_or(0.0) > 0.0);+    let ext = |e: &Value| if extra { format!("  media {:.1} min, hold {:.1} min", e["mediaSeconds"].as_f64().unwrap_or(0.0) / 60.0, e["holdSeconds"].as_f64().unwrap_or(0.0) / 60.0) } else { String::new() };     let mut lines = vec![format!("{:<11} {:>9} {:>9} {:>9} {:>6} {:>7} {:>8}  {}", "step", "wall min", "think min", "tool min", "turns", "visits", "returns", "longest think")];     let (mut tw, mut tt, mut tb, mut tn) = (0u64, 0u64, 0u64, 0u64);     for n in names {@@ -1072,7 +1081,7 @@ fn step_table(r: &Run) -> String {         tw += w; tt += t; tb += b; tn += k;         let a = attention::step_attention(&r.data, &n);         let look = if a.is_null() { String::new() } else if a["kind"] == "error" { format!("  error ({})", a["count"]) } else { format!("  needs a look ({} finding(s))", a["count"]) };-        lines.push(format!("{:<11} {:>9.1} {:>9.1} {:>9.1} {:>6} {:>7} {:>8}  {:.1} min{look}", n, w as f64 / 60.0, t as f64 / 60.0, b as f64 / 60.0, k, e["visits"].as_u64().unwrap_or(0), e["returns"].as_u64().unwrap_or(0), e["longestThinkSeconds"].as_u64().unwrap_or(0) as f64 / 60.0));+        lines.push(format!("{:<11} {:>9.1} {:>9.1} {:>9.1} {:>6} {:>7} {:>8}  {:.1} min{look}{}", n, w as f64 / 60.0, t as f64 / 60.0, b as f64 / 60.0, k, e["visits"].as_u64().unwrap_or(0), e["returns"].as_u64().unwrap_or(0), e["longestThinkSeconds"].as_u64().unwrap_or(0) as f64 / 60.0, ext(e)));     }     lines.push(format!("{:<11} {:>9.1} {:>9.1} {:>9.1} {:>6}", "total", tw as f64 / 60.0, tt as f64 / 60.0, tb as f64 / 60.0, tn));     lines.join("\n")@@ -1809,7 +1818,197 @@ fn media_seconds(path: &str) -> f64 {  /// Cut a sped-up copy of a pulled clip next to it (`<name>-10x.mp4`) and record both as artifacts /// of the step. The window recorder only emits frames on change, so the cut is by timestamp.-fn clip_artifacts(r: &mut Run, step: &Value, file: &str, speed: u32) { clip_artifacts_with(r, step, file, speed, MEDIA_TOOLS) }+fn clip_artifacts(r: &mut Run, step: &Value, file: &str, speed: u32) {+    // issue #170: a clip of the background size goes to the media worker, and the command goes on+    if !MEDIA_NOW.load(std::sync::atomic::Ordering::Relaxed) && !IN_WORKER.load(std::sync::atomic::Ordering::Relaxed) {+        if let Some(lim) = media_background_mb() {+            let mb = std::fs::metadata(file).map(|m| m.len() / 1_048_576).unwrap_or(0);+            if mb >= lim && media_enqueue(r, step, file, speed, mb) { return; }+        }+    }+    let t0 = std::time::Instant::now();+    clip_artifacts_with(r, step, file, speed, MEDIA_TOOLS);+    // issue #171: media processing is its own line in the step report (it is part of the command's tool time)+    let secs = t0.elapsed().as_secs_f64();+    if secs >= 0.5 { r.log("media-time", json!({"clip": step, "file": file, "seconds": (secs * 10.0).round() / 10.0, "chargedTo": r.data["currentStep"], "background": IN_WORKER.load(std::sync::atomic::Ordering::Relaxed)})); }+}++/// Issue #170: set inside `media-worker`, which runs the clip work away from the engineering commands.+static IN_WORKER: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);++// Issue #170 part B (Josh Aberdeen's proposal): a durable background worker for the clip media. A step that+// stops a clip of AIFLOW_MEDIA_BACKGROUND_MB or more (default 64; 0 = every clip; "off" = none) queues it and+// goes on; `adom-aiflow media-worker` (started detached, one at a time by a lease) runs the same check, sheet+// and cuts against a scratch copy of the run and writes what it found to an outbox. The run's own commands are+// the only writers of run.json and run.jsonl: each one folds the outbox in when it loads the run (and says what+// was found: a blank clip, its contact sheet). A worker that died leaves its job to be taken again; `compose`,+// `recut` and `deliver` wait for the worker and do any job still queued themselves. Media problems stay notes.++fn media_background_mb() -> Option<u64> {+    match std::env::var("AIFLOW_MEDIA_BACKGROUND_MB") {+        Ok(v) if v.trim().eq_ignore_ascii_case("off") => None,+        Ok(v) => Some(v.trim().parse().unwrap_or(64)),+        Err(_) => Some(64),+    }+}++fn media_dir(dir: &Path) -> PathBuf { dir.join("media") }++fn media_files(dir: &Path, sub: &str) -> Vec<PathBuf> {+    let mut v: Vec<PathBuf> = std::fs::read_dir(media_dir(dir).join(sub)).map(|d| d.filter_map(|e| e.ok().map(|e| e.path())).filter(|p| p.extension().is_some_and(|x| x == "json")).collect()).unwrap_or_default();+    v.sort();+    v+}++/// A worker holds the lease: its pid is alive and the lease was renewed in the last 6 hours.+fn media_worker_alive(dir: &Path) -> bool {+    let Some(l) = std::fs::read_to_string(media_dir(dir).join("worker.lock")).ok().and_then(|t| serde_json::from_str::<Value>(&t).ok()) else { return false };+    let pid_alive = l["pid"].as_u64().is_some_and(|p| Path::new(&format!("/proc/{p}")).exists());+    let fresh = l["at"].as_str().and_then(aiflow_run::parse).zip(aiflow_run::parse(&now())).is_some_and(|(a, n)| n.saturating_sub(a) < 6 * 3600);+    pid_alive && fresh+}++fn media_worker_spawn(dir: &Path) -> bool {+    if media_worker_alive(dir) { return true; }+    let Ok(exe) = std::env::current_exe() else { return false };+    let log = std::fs::OpenOptions::new().create(true).append(true).open(media_dir(dir).join("worker.log")).ok();+    let mut c = std::process::Command::new(exe);+    c.arg("--run").arg(dir).arg("media-worker").env("AIFLOW_DASHBOARD", "off").stdin(std::process::Stdio::null()).stdout(std::process::Stdio::null());+    match log { Some(f) => { c.stderr(f); } None => { c.stderr(std::process::Stdio::null()); } }+    c.spawn().is_ok()+}++/// Queue a clip for the worker; false (do it here) when the queue or the worker cannot be had.+fn media_enqueue(r: &mut Run, step: &Value, file: &str, speed: u32, mb: u64) -> bool {+    let q = media_dir(&r.dir).join("queue");+    if std::fs::create_dir_all(&q).is_err() { return false; }+    let stem: String = Path::new(file).file_stem().and_then(|n| n.to_str()).unwrap_or("clip").chars().map(|c| if c.is_ascii_alphanumeric() || c == '-' { c } else { '_' }).collect();+    let id = format!("{}-{stem}", now().replace([':', '-'], "").replace('T', "-").trim_end_matches('Z'));+    let job = q.join(format!("{id}.json"));+    if std::fs::write(&job, json!({"id": id, "step": step, "file": file, "speed": speed, "mb": mb, "at": now()}).to_string()).is_err() { return false; }+    if !media_worker_spawn(&r.dir) { let _ = std::fs::remove_file(&job); return false; }+    if let Some(c) = r.data["captures"].as_array_mut().and_then(|a| a.iter_mut().find(|c| c["file"].as_str() == Some(file))) { c["mediaQueued"] = json!({"id": id, "at": now()}); }+    r.log("clip-media-queued", json!({"step": step, "file": file, "id": id, "mb": mb}));+    eprintln!("clip {} ({mb} MB) is checked and cut in the background (media worker); the next adom-aiflow command reports its blank check, contact sheet and cuts", Path::new(file).file_name().and_then(|n| n.to_str()).unwrap_or(""));+    true+}++/// The worker: take the lease, put back any job a dead worker left half done, then work the queue.+fn media_worker_run(dir: &Path) {+    let md = media_dir(dir);+    for sub in ["queue", "work", "done"] { let _ = std::fs::create_dir_all(md.join(sub)); }+    let lock = md.join("worker.lock");+    let take = || std::fs::OpenOptions::new().write(true).create_new(true).open(&lock).map(|mut f| { use std::io::Write; let _ = f.write_all(json!({"pid": std::process::id(), "at": now()}).to_string().as_bytes()); });+    if take().is_err() {+        if media_worker_alive(dir) { return; }+        let _ = std::fs::remove_file(&lock);+        if take().is_err() { return; }+    }+    for w in media_files(dir, "work") { if let Some(n) = w.file_name() { let _ = std::fs::rename(&w, md.join("queue").join(n)); } }+    IN_WORKER.store(true, std::sync::atomic::Ordering::Relaxed);+    MEDIA_NOW.store(true, std::sync::atomic::Ordering::Relaxed);+    loop {+        let jobs = media_files(dir, "queue");+        if jobs.is_empty() { break; }+        for j in jobs {+            let Some(name) = j.file_name().map(|n| n.to_owned()) else { continue };+            let work = md.join("work").join(&name);+            if std::fs::rename(&j, &work).is_err() { continue; }+            let _ = std::fs::write(&lock, json!({"pid": std::process::id(), "at": now(), "job": name.to_string_lossy()}).to_string());+            let out = media_job(dir, &work);+            let done = md.join("done").join(&name);+            let tmp = done.with_extension("tmp");+            if std::fs::write(&tmp, out.to_string()).is_ok() { let _ = std::fs::rename(&tmp, &done); }+            let _ = std::fs::remove_file(&work);+        }+    }+    let _ = std::fs::remove_file(&lock);+}++/// One job against a scratch copy of the run: the same `clip_artifacts`, its ledger lines and what it changed+/// on the capture come back as the outbox record.+fn media_job(dir: &Path, work: &Path) -> Value {+    let job: Value = std::fs::read_to_string(work).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or(Value::Null);+    let (file, speed) = (job["file"].as_str().unwrap_or("").to_string(), job["speed"].as_u64().unwrap_or(10) as u32);+    let real = Run::open(dir).map(|r| r.data).unwrap_or(Value::Null);+    let scratch = media_dir(dir).join("scratch").join(job["id"].as_str().unwrap_or("job"));+    let _ = std::fs::remove_dir_all(&scratch);+    let _ = std::fs::create_dir_all(&scratch);+    let mut t = Run { dir: scratch.clone(), data: json!({"clock": real["clock"], "captures": real["captures"], "currentStep": real["currentStep"]}) };+    let before = t.data["captures"].as_array().and_then(|a| a.iter().find(|c| c["file"].as_str() == Some(&file)).cloned()).unwrap_or(json!({}));+    if Path::new(&file).is_file() { clip_artifacts(&mut t, &job["step"], &file, speed); }+    let after = t.data["captures"].as_array().and_then(|a| a.iter().find(|c| c["file"].as_str() == Some(&file)).cloned()).unwrap_or(json!({}));+    let (set, removed) = capture_patch(&before, &after);+    let events = t.ledger();+    let _ = std::fs::remove_dir_all(&scratch);+    json!({"job": job, "events": events, "set": set, "removed": removed, "missing": !Path::new(&file).is_file(), "finishedAt": now()})+}++/// What the worker changed on a capture: the keys set (new or different) and the keys removed.+fn capture_patch(before: &Value, after: &Value) -> (Value, Vec<String>) {+    let (b, a) = (before.as_object().cloned().unwrap_or_default(), after.as_object().cloned().unwrap_or_default());+    let set: serde_json::Map<String, Value> = a.iter().filter(|(k, v)| b.get(*k) != Some(*v)).map(|(k, v)| (k.clone(), v.clone())).collect();+    let removed: Vec<String> = b.keys().filter(|k| !a.contains_key(*k)).cloned().collect();+    (Value::Object(set), removed)+}++/// Fold the worker's outbox into the run (the single writer), say what it found, and start a worker again+/// when jobs are queued and none holds the lease. Returns how many jobs were folded in.+fn media_reconcile(r: &mut Run) -> usize {+    let done = media_files(&r.dir, "done");+    let mut n = 0;+    for d in &done {+        let Some(o) = std::fs::read_to_string(d).ok().and_then(|t| serde_json::from_str::<Value>(&t).ok()) else { continue };+        let file = o["job"]["file"].as_str().unwrap_or("").to_string();+        let name = Path::new(&file).file_name().and_then(|n| n.to_str()).unwrap_or("").to_string();+        for e in o["events"].as_array().into_iter().flatten() { r.log_raw(e); }+        if let Some(c) = r.data["captures"].as_array_mut().and_then(|a| a.iter_mut().find(|c| c["file"].as_str() == Some(&file))) {+            for (k, v) in o["set"].as_object().into_iter().flatten() { c[k.as_str()] = v.clone(); }+            if let Some(m) = c.as_object_mut() { for k in o["removed"].as_array().into_iter().flatten().filter_map(Value::as_str) { m.remove(k); } m.remove("mediaQueued"); }+        }+        let evs: Vec<&Value> = o["events"].as_array().map(|a| a.iter().collect()).unwrap_or_default();+        for e in &evs {+            match e["event"].as_str() {+                Some("clip-suspect") => eprintln!("WARNING clip {name} looks blank (background check): {}", e["why"].as_str().unwrap_or("")),+                Some("clip-truncated") => eprintln!("WARNING clip {name} is {} (background check). Re-record the step (`step <name> --back --why \"clip truncated\"`) while the work is fresh.", e["why"].as_str().unwrap_or("")),+                Some("artifact") if e["kind"] == "contact-sheet" => eprintln!("contact sheet of {name} (from the background worker; look at it): {}", e["file"].as_str().unwrap_or("")),+                _ => {}+            }+        }+        if o["missing"] == true { eprintln!("NOTE (a suggestion, not a stop): the background worker found no file for clip {name}; `adom-aiflow recut` once it is back"); }+        else { eprintln!("clip {name}: checked and cut in the background ({})", if evs.iter().any(|e| e["kind"] == "clip10x") { "10x and action cuts made" } else { "no cuts: see above" }); }+        r.log("clip-media-reconciled", json!({"file": file, "id": o["job"]["id"], "finishedAt": o["finishedAt"]}));+        let _ = std::fs::remove_file(d);+        n += 1;+    }+    if n > 0 { let _ = r.save(); }+    if !media_files(&r.dir, "queue").is_empty() && !media_worker_alive(&r.dir) { media_worker_spawn(&r.dir); }+    n+}++/// `compose`, `recut` and `deliver` need every clip judged and cut: wait for the worker (AIFLOW_MEDIA_WAIT_MIN,+/// default 30), fold its outbox in, and do any job still queued here.+fn media_drain(r: &mut Run) {+    let limit = std::env::var("AIFLOW_MEDIA_WAIT_MIN").ok().and_then(|v| v.parse::<u64>().ok()).unwrap_or(30) * 60;+    let t0 = std::time::Instant::now();+    let mut said = false;+    while media_worker_alive(&r.dir) && t0.elapsed().as_secs() < limit {+        if !said { eprintln!("waiting for the background media worker to finish its clips (up to {} min)", limit / 60); said = true; }+        std::thread::sleep(std::time::Duration::from_secs(2));+    }+    media_reconcile(r);+    let left = media_files(&r.dir, "queue");+    if left.is_empty() || media_worker_alive(&r.dir) { return; }+    MEDIA_NOW.store(true, std::sync::atomic::Ordering::Relaxed);+    for j in left {+        let Some(job) = std::fs::read_to_string(&j).ok().and_then(|t| serde_json::from_str::<Value>(&t).ok()) else { continue };+        let _ = std::fs::remove_file(&j);+        let file = job["file"].as_str().unwrap_or("").to_string();+        if Path::new(&file).is_file() { clip_artifacts(r, &job["step"], &file, job["speed"].as_u64().unwrap_or(10) as u32); }+        if let Some(c) = r.data["captures"].as_array_mut().and_then(|a| a.iter_mut().find(|c| c["file"].as_str() == Some(file.as_str()))).and_then(|c| c.as_object_mut()) { c.remove("mediaQueued"); }+    }+    let _ = r.save();+}  /// Issue #170: set by `recut` and `compose`, the commands whose job is the media; every other command /// (a step, a tour, finish) leaves a clip over the deferral size for them.@@ -3223,6 +3422,15 @@ fn continuation_report(r: &Run, evidence: Option<&Path>) -> Value {     let stop = evidence.as_ref().map(|p| std::fs::read_to_string(p).map_err(|e| e.to_string()).and_then(|s| serde_json::from_str::<Value>(&s).map_err(|e| e.to_string())));     let mut v = continuation::assess(&r.data, &pending, stop);     if let (Some(p), Some(o)) = (&evidence, v.as_object_mut()) { o.insert("stopEvidencePath".into(), json!(p.display().to_string())); }+    // issue #171: the blocker as one fingerprint (what is still pending, finish's open items, the decision,+    // the board and the spec), so a host sees "same blocker, nothing changed" between calls without its own hash+    // (the decision is hashed too; the basis printed beside it leaves it out, the report already says it)+    let basis = json!({"pending": pending, "openItems": r.data["lastRefusalItems"], "suggestedReturn": r.data["suggestedReturn"]["step"], "boardSha256": bs, "specSha256": ss});+    let hashed = json!({"decision": v["decision"], "basis": basis});+    if let Some(o) = v.as_object_mut() {+        o.insert("blockerFingerprint".into(), json!(provenance::sha256_hex(hashed.to_string().as_bytes())[..16].to_string()));+        o.insert("blockerBasis".into(), basis);+    }     v } @@ -3278,6 +3486,8 @@ fn main() {         }     };     if let Some(h) = input_hint { let _ = INPUT_HINT.set(h); }+    // issue #170: the media worker writes nothing to the run's ledger or run.json itself (no turn either)+    if let Cmd::MediaWorker = &cli.cmd { media_worker_run(&run_dir(&cli)); return; }     // A host stop hook must not open windows, start captures or append timing events.     if let Cmd::FinishGuard { json: as_json, stop_evidence } = &cli.cmd {         let r = load_run(&cli);@@ -5322,6 +5532,9 @@ fn main() {         Cmd::Finish => {             let t = thread(&cli);             let mut r = load_run(&cli);+            // issue #170: finish does not wait on the media; it says what is still being processed+            let queued = media_files(&r.dir, "queue").len() + media_files(&r.dir, "work").len();+            if queued > 0 { note(&format!("{queued} clip(s) are still being checked and cut by the background media worker; compose and deliver wait for it"), &[]); }             if let Some(target) = r.data["target"].as_str().map(str::to_string) {                 let (pulled, missing) = repull_failed_clips(&mut r, &Bridge { ai_thread: t.clone(), target }, &dir);                 if pulled + missing > 0 { eprintln!("pulled {pulled} clip(s) that had not reached the run; {missing} still missing"); r.save().unwrap(); }@@ -5604,6 +5817,8 @@ fn main() {             let parts_v = o.get("parts").map(|p| p["parts"].clone()).unwrap_or(Value::Null);             let lib_parts = project_lib::report(&parts_v);             for g in project_lib::recheck(&parts_v) { missing.push(format!("parts: project library: {g}")); }+            // issue #199: a board page (or a molecule's) must show this board's design review, not an earlier one's+            if let Some(item) = page_results::page_item(&r.data, &sha256_file(&r.current_board()), molecule_pins::wants_molecule(&spec_soft(&r), &requirements_of(&r))) { missing.push(item); }             // issue #193: a part with no page is reported with its requests, its assessment and the next command             for m in o.get("parts").and_then(|p| p["missing"].as_array()).into_iter().flatten().filter_map(Value::as_str) {                 let (tries, a, ch) = (supply::attempts(&r.data, m), supply_assessed(&r, m), r.data["supplyChoices"].get(m).cloned());@@ -5624,6 +5839,17 @@ fn main() {                 eprintln!("Open items (suggestions, the board is not blocked):\n  {}", open_items.join("\n  "));             }             // issue #171: name the earliest step the open items go back to (a suggestion; the AI declares the return)+            // issue #171: a check that fails a step the run already accepted is an `invalidated` event (once per+            // step, item, board and spec), and the step carries it, so the timeline says why it went back+            for (step, item) in sequence::invalidated(&seq, r.data["currentStep"].as_str().unwrap_or(""), &r.data, &open_items) {+                let key = format!("{step}|{}|{bs}|{ss}", deviations::key(&item));+                if r.data["invalidations"].as_array().is_some_and(|a| a.iter().any(|k| k == key.as_str())) { continue; }+                let rec = json!({"step": step, "checker": deviations::key(&item), "result": item, "boardSha256": bs, "specSha256": ss, "at": now(), "by": "finish"});+                r.log("invalidated", rec.clone());+                r.data["steps"][step.as_str()]["invalidated"] = rec;+                if !r.data["invalidations"].is_array() { r.data["invalidations"] = json!([]); }+                r.data["invalidations"].as_array_mut().unwrap().push(json!(key));+            }             let back_to = sequence::earliest_return(&seq, r.data["currentStep"].as_str().unwrap_or(""), &open_items);             if let Some((s, i)) = &back_to { eprintln!("Suggested: {}", sequence::return_suggestion(s, i)); }             r.data["suggestedReturn"] = back_to.map(|(s, i)| json!({"step": s, "item": i, "at": now()})).unwrap_or(Value::Null);@@ -5817,6 +6043,7 @@ fn main() {             if let Err(e) = media_tools_missing(MEDIA_TOOLS) { err(&e, &["Install ffmpeg (it brings ffprobe) on this machine, then recut again. No clip was judged.".into()]); }             MEDIA_NOW.store(true, std::sync::atomic::Ordering::Relaxed);             let mut r = load_run(&cli);+            media_drain(&mut r);             // a clip whose pull failed at its stop comes first: the desktop cleans its recordings folder             if let Some(target) = r.data["target"].as_str().map(str::to_string) {                 let (pulled, missing) = repull_failed_clips(&mut r, &Bridge { ai_thread: t, target }, &dir);@@ -5844,6 +6071,7 @@ fn main() {             if recover_capture_files(&mut r, &dir) > 0 { r.save().unwrap(); }             // issue #170: clips a step left for later (too big to check inside it) are checked and cut now             MEDIA_NOW.store(true, std::sync::atomic::Ordering::Relaxed);+            media_drain(&mut r);             let deferred: Vec<(String, String)> = r.data["captures"].as_array().into_iter().flatten().filter(|c| c.get("mediaDeferred").is_some()).filter_map(|c| Some((c["file"].as_str()?.to_string(), format!("{}-{}", c["step"].as_str()?, c["visit"].as_u64().unwrap_or(1))))).collect();             for (f, tag) in &deferred { if Path::new(f).is_file() { clip_artifacts(&mut r, &json!(tag), f, 10); } }             if !deferred.is_empty() { r.save().unwrap(); }@@ -5945,6 +6173,7 @@ fn main() {             print_next(&cli);         }         Cmd::FinishGuard { .. } => unreachable!("handled before side effects"),+        Cmd::MediaWorker => unreachable!("handled before side effects"),         Cmd::Done { message, stop_evidence } => {             thread(&cli);             let mut r = load_run(&cli);@@ -7338,6 +7567,10 @@ fn main() {         Cmd::Deliver { video, message, accept_suspect } => {             thread(&cli);             let mut r = load_run(&cli);+            // issue #170: every clip judged before the delivery reads the suspect flags+            media_drain(&mut r);+            // issue #199: say it when the page would go out with another board's design review+            if let Some(item) = page_results::page_item(&r.data, &sha256_file(&r.current_board()), molecule_pins::wants_molecule(&spec_soft(&r), &requirements_of(&r))) { note(&item, &["Delivering anyway is your call; tell the human the page does not show this board's results yet.".into()]); }             if let Some(target) = r.data["target"].as_str().map(str::to_string) {                 let (pulled, missing) = repull_failed_clips(&mut r, &Bridge { ai_thread: thread(&cli), target }, &dir);                 if pulled + missing > 0 { eprintln!("pulled {pulled} clip(s) that had not reached the run; {missing} still missing"); r.save().unwrap(); }@@ -7600,6 +7833,46 @@ mod media_tools_tests {         let _ = std::fs::remove_dir_all(dir);     } +    /// Issue #170 part B: a queued clip is checked and cut by the worker against a scratch copy, and the run's+    /// own command folds the result in (ledger lines with their own times, the capture's flags), once.+    #[test]+    fn the_media_worker_cuts_in_the_background_and_the_next_command_folds_it_in() {+        let Some((dir, source)) = media_fixture("worker", "routing-motion.mp4", &["-f", "lavfi", "-i", "testsrc2=size=128x128:rate=10", "-t", "6", "-pix_fmt", "yuv420p"]) else { return };+        let Some((blank_dir, blank)) = media_fixture("worker-blank", "placement-blank.mp4", &["-f", "lavfi", "-i", "color=c=black:s=96x96:r=2", "-t", "4", "-pix_fmt", "yuv420p"]) else { return };+        let mut r = Run { dir: dir.clone(), data: json!({"clock": {"promptTime": now()}, "currentStep": "routing", "captures": [{"file": source, "step": "routing", "visit": 1}, {"file": blank, "step": "placement", "visit": 1}]}) };+        r.save().unwrap();+        // queue both by hand (the spawn is the binary's), then run the worker loop in this process+        let q = media_dir(&dir).join("queue");+        std::fs::create_dir_all(&q).unwrap();+        for (i, (f, st)) in [(&source, "routing-1"), (&blank, "placement-1")].iter().enumerate() {+            std::fs::write(q.join(format!("{i}.json")), json!({"id": format!("j{i}"), "step": st, "file": f, "speed": 10}).to_string()).unwrap();+        }+        media_worker_run(&dir);+        IN_WORKER.store(false, std::sync::atomic::Ordering::Relaxed);+        MEDIA_NOW.store(false, std::sync::atomic::Ordering::Relaxed);+        assert!(dir.join("routing-motion-10x.mp4").is_file(), "the worker made the cut");+        assert!(r.ledger().iter().all(|e| e["event"] != "artifact"), "the worker never writes the run's ledger");+        assert_eq!(media_files(&dir, "done").len(), 2);+        let mut r = Run::open(&dir).unwrap();+        assert_eq!(media_reconcile(&mut r), 2);+        let led = r.ledger();+        assert!(led.iter().any(|e| e["event"] == "artifact" && e["kind"] == "clip10x"));+        assert!(led.iter().any(|e| e["event"] == "media-time" && e["background"] == true));+        assert!(r.data["captures"][1]["suspect"].as_str().unwrap_or("").contains("flat colour"), "{}", r.data["captures"][1]);+        assert!(r.data["captures"][0].get("suspect").is_none());+        assert_eq!(media_reconcile(&mut r), 0, "folded in once");+        assert!(!media_dir(&dir).join("worker.lock").exists());+        let _ = std::fs::remove_dir_all(dir);+        let _ = std::fs::remove_dir_all(blank_dir);+    }++    #[test]+    fn the_capture_patch_names_what_changed() {+        let (set, removed) = capture_patch(&json!({"file": "a", "mediaDeferred": {"mb": 2}, "x": 1}), &json!({"file": "a", "x": 2, "suspect": "blank"}));+        assert_eq!(set, json!({"x": 2, "suspect": "blank"}));+        assert_eq!(removed, vec!["mediaDeferred".to_string()]);+    }+     /// Changing footage still gets its cuts.     #[test]     fn healthy_motion_clip_still_generates_video_derivatives() {