src/monitor.rs
//! Server-monitored self-sync (server-user `nornir`): the per-workspace fetch
//! step, the work gate, and the on-demand embed pass.
//!
//! [`fetch_workspace`] clones/fetches a workspace's git members (pure-Rust HTTPS
//! via [`crate::gitio`]) and records sync state in the [`crate::registry`]; it is
//! shared by the CLI's `nornir workspace fetch` and the server's Register/Fetch
//! RPCs. The v1 60 s poll loop and the whole-workspace populate
//! (`republish_with`) were removed with Mímir v2 (PLAN-MIMIR-V2 §7); the v2
//! consumer loop (git first-parent events per repo) replaces them.
use std::path::Path;
use anyhow::{Context as _, Result, anyhow, bail};
use crate::registry::Registry;
/// One member's clone/fetch outcome — the durable, remotely-readable twin of the
/// per-member `eprintln!("…fetch error…")` that used to be the only signal. Every
/// fetched member produces exactly one of these (ok OR error), so a thin viz/CLI
/// client can see *why* a member didn't populate (auth, vanished repo, network),
/// not just that its data is missing. Persisted to the `clone_events` warehouse
/// table via [`crate::warehouse::clone_events::record_fetch_report`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberOutcome {
/// Workspace member name.
pub member: String,
/// The git remote that was cloned/fetched (empty for local members — but those
/// are skipped, so an outcome always carries a real remote).
pub remote: String,
/// `clone` | `fetch` (today both go through `clone_or_fetch`; recorded as
/// `clone-fetch` so the op string matches the error chain users already saw).
pub op: String,
/// `ok` | `error`.
pub status: String,
/// The error chain (`{e:#}`) on failure; the resolved SHA on success.
pub detail: String,
/// Wall-clock duration of this member's clone/fetch, milliseconds.
pub elapsed_ms: i64,
}
/// Outcome of fetching one workspace's members.
#[derive(Debug, Default)]
pub struct FetchReport {
pub workspace: String,
/// Members with a remote that were clone/fetched.
pub fetched: usize,
/// Members whose HEAD moved since last sync.
pub changed: Vec<String>,
/// `(member, error)` for members that failed.
pub errors: Vec<(String, String)>,
/// One outcome per fetched member (ok or error) — the readable populate-status
/// stream. Same set as `fetched`, but with op/status/detail/elapsed so the
/// failure (or success) is first-class and persistable, not log-only.
pub outcomes: Vec<MemberOutcome>,
}
impl FetchReport {
pub fn changed_any(&self) -> bool {
!self.changed.is_empty()
}
}
/// **The per-workspace WORK GATE — the thing whose absence wrote a red row for a
/// fetch that succeeded.**
///
/// `<root>/<ws>/git/` is a single shared directory with no lock of any kind in
/// it. [`fetch_workspace_with_progress`] `create_dir_all`s into it, and one layer
/// down `gitio::ssh_sync` does a TOCTOU `if dest.join(".git").exists() { open }
/// else { init }`, then `materialize_worktree` **empties the whole worktree**
/// (`remove_dir_all` per entry) before writing every blob back. Two runs over one
/// member therefore fail each other in milliseconds, before any network I/O —
/// which is exactly the shape of the observed `✗ fail workspace_clone skade
/// 13 ms` written while the honest fetch (`skade.git synced 4dce050a…`, ~3 s)
/// was still running and landed fine.
///
/// The register-eager populate is DETACHED (`spawn_blocking`, the `Register` RPC
/// returns at once) and a freshly registered workspace has no `last_synced`, so
/// `is_workspace_due` says *due* on the very next poll tick. Eager and sweep
/// overlapping is not a rare interleaving — it is the default on every add.
///
/// A loser here must **wait, not fail**: the work is idempotent, so serialising
/// it costs a few seconds and a spurious `error` costs an operator a diagnosis
/// they cannot act on. Hence a plain mutual-exclusion gate rather than a
/// try-lock-and-skip.
///
/// It is deliberately OBSERVABLE ([`peak_in_flight`](Self::peak_in_flight),
/// [`waited`](Self::waited)) — a gate that cannot be seen to have fired is a
/// green nobody has seen red (LAW 2). The guard test drives real threads through
/// [`enter`](Self::enter) and asserts the observed peak is 1, and asserts it is
/// >1 when the same closures bypass the gate.
///
/// **Scope, honestly:** this is an in-PROCESS gate on one `WorkspaceCtx`. It
/// closes the server's own eager-vs-sweep race, which is the one that produced
/// the false red. It does NOT gate a `nornir workspace fetch` running in a
/// separate CLI process against the same tree; that needs a filesystem lock and
/// is not fixed here.
/// **Wait, or refuse by name?** [`enter`](WorkspaceGate::enter) waits — a fetch or
/// a republish that loses the race is idempotent and a spurious `error` costs an
/// operator a diagnosis they cannot act on (see the type docs).
#[derive(Debug, Default)]
pub struct WorkspaceGate {
lock: std::sync::Mutex<()>,
in_flight: std::sync::atomic::AtomicUsize,
peak_in_flight: std::sync::atomic::AtomicUsize,
waited: std::sync::atomic::AtomicUsize,
/// Callers blocked in [`enter`](WorkspaceGate::enter) right now (see
/// [`give_way`](WorkspaceGate::give_way)).
queued: std::sync::atomic::AtomicUsize,
/// What is inside the gate right now — the label the current holder passed
/// to [`enter_as`](WorkspaceGate::enter_as) (`"populate"` for the plain
/// [`enter`](WorkspaceGate::enter) callers, a job id for a repopulate).
/// `None` = free. Written under the gate's own mutex, read WITHOUT it, so a
/// refusal can name the holder while the holder still holds the lock.
holder: std::sync::Mutex<Option<String>>,
}
/// Clears the holder label when the guarded work leaves the gate — by return OR
/// by panic, so a panicking holder cannot leave its name standing for ever.
struct HolderLabel<'a>(&'a WorkspaceGate);
impl Drop for HolderLabel<'_> {
fn drop(&mut self) {
*self.0.holder.lock().unwrap_or_else(|p| p.into_inner()) = None;
}
}
/// Decrements the gate's in-flight count even if the guarded work panics.
struct GateEntry<'a>(&'a WorkspaceGate);
impl Drop for GateEntry<'_> {
fn drop(&mut self) {
self.0
.in_flight
.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
}
}
/// The label every plain [`WorkspaceGate::enter`] caller registers — the
/// fetch/republish half. Named once so the refusal message and the callers
/// cannot drift.
pub const POPULATE: &str = "populate";
impl WorkspaceGate {
pub fn new() -> Self {
Self::default()
}
/// Run `f` with this workspace's git tree to itself. Blocks while another
/// caller holds the gate. A poisoned lock is TAKEN anyway (a panicking
/// producer must not wedge every later fetch into an unrecoverable error) —
/// the same `into_inner` recovery `republish_served` already used.
pub fn enter<T>(&self, f: impl FnOnce() -> T) -> T {
self.enter_as(POPULATE, f)
}
/// [`enter`](Self::enter), naming the holder. `label` is what a refusing
/// [`try_enter_as`](Self::try_enter_as) hands back — `"populate"` for the
/// fetch/republish callers, a **jera job id** for a repopulate.
///
/// This is the ONE body: [`enter`](Self::enter) is this with the default
/// label, so there is no second acquisition path to keep in agreement.
pub fn enter_as<T>(&self, label: &str, f: impl FnOnce() -> T) -> T {
use std::sync::atomic::Ordering::SeqCst;
// Contention is measured BEFORE blocking, and the probe guard is dropped
// at this statement's end — taking it into the `lock()` below would
// deadlock on a non-reentrant mutex.
let contended = self.lock.try_lock().is_err();
if contended {
self.waited.fetch_add(1, SeqCst);
}
self.queued.fetch_add(1, SeqCst);
let _held = self.lock.lock().unwrap_or_else(|p| p.into_inner());
self.queued.fetch_sub(1, SeqCst);
let n = self.in_flight.fetch_add(1, SeqCst) + 1;
self.peak_in_flight.fetch_max(n, SeqCst);
let _entry = GateEntry(self);
*self.holder.lock().unwrap_or_else(|p| p.into_inner()) = Some(label.to_string());
let _label = HolderLabel(self);
f()
}
/// **Refuse instead of waiting.** Runs `f` under the gate when it is free;
/// when it is not, runs NOTHING and returns `Err(holder)` — the label of
/// whatever is inside, so the caller can name it (`Err("")` only if the
/// holder is mid-handshake, which is a race the message must not pretend to
/// resolve).
///
/// The only caller is `Workspaces.Repopulate` — see the note above the type.
pub fn try_enter_as<T>(&self, label: &str, f: impl FnOnce() -> T) -> Result<T, String> {
use std::sync::atomic::Ordering::SeqCst;
let Ok(_held) = self.lock.try_lock() else {
self.waited.fetch_add(1, SeqCst);
return Err(self.holder().unwrap_or_default());
};
let n = self.in_flight.fetch_add(1, SeqCst) + 1;
self.peak_in_flight.fetch_max(n, SeqCst);
let _entry = GateEntry(self);
*self.holder.lock().unwrap_or_else(|p| p.into_inner()) = Some(label.to_string());
let _label = HolderLabel(self);
Ok(f())
}
/// **Let whoever is waiting for the gate go first.** Returns once no caller
/// is blocked in [`enter`](Self::enter) — each of them is then inside the
/// gate or done, and this caller's own `enter` queues behind it.
///
/// For work that takes the gate once per small unit, many times in a row (a
/// backfill: once per commit). The gate is a plain mutex and a plain mutex
/// is not fair: the thread that just left it can take it again before a
/// thread that has been waiting is even scheduled, so a tight
/// `leave → enter` loop can keep everyone else out for as long as it runs.
/// Calling this between the units bounds every waiter's wait at ONE unit.
pub fn give_way(&self) {
use std::sync::atomic::Ordering::SeqCst;
while self.queued.load(SeqCst) > 0 {
std::thread::sleep(std::time::Duration::from_millis(5));
}
}
/// Callers blocked in [`enter`](Self::enter) right now.
pub fn queued(&self) -> usize {
self.queued.load(std::sync::atomic::Ordering::SeqCst)
}
/// The label of whatever is inside the gate, or `None` when it is free.
pub fn holder(&self) -> Option<String> {
self.holder
.lock()
.unwrap_or_else(|p| p.into_inner())
.clone()
}
/// The highest number of callers ever inside [`enter`](Self::enter) at once.
/// A working gate reads 1 forever; anything above 1 is the race back.
pub fn peak_in_flight(&self) -> usize {
self.peak_in_flight
.load(std::sync::atomic::Ordering::SeqCst)
}
/// How many callers found the gate held and had to wait — the count of races
/// that would otherwise have been a false `error` row.
pub fn waited(&self) -> usize {
self.waited.load(std::sync::atomic::Ordering::SeqCst)
}
}
/// Clone/fetch every git member of `name` into `<root>/<name>/git/<member>`,
/// write back each member's sync state, and return what changed. Shared by the
/// CLI verb and the poll loop. HTTPS only (SSH refused; see [`crate::gitio`]).
pub fn fetch_workspace(reg: &Registry, root: &Path, name: &str) -> Result<FetchReport> {
fetch_workspace_with_progress(reg, root, name, |_| {})
}
/// Like [`fetch_workspace`], but invokes `on_member` with each member's
/// [`MemberOutcome`] **as that member's clone/fetch completes** — the streaming
/// seam. The server uses it to emit + finish a per-member *child* job (under the
/// populate parent) the instant each clone lands, so the Jobs panel shows members
/// appear one at a time instead of all-at-once at the end. `on_member` is
/// best-effort progress reporting; it must not fail the fetch (it returns `()`).
pub fn fetch_workspace_with_progress(
reg: &Registry,
root: &Path,
name: &str,
on_member: impl FnMut(&MemberOutcome),
) -> Result<FetchReport> {
fetch_members(reg, root, name, None, on_member)
}
/// [`fetch_workspace_with_progress`] for the members in `only` (`None`: all of
/// them) — a workspace registered again clones what is NEW at once, without
/// fetching the other seventeen.
///
/// What it learns is MERGED into the registry row as that row is when the
/// fetch ends ([`Registry::merge_member_states`]), never written over it: the
/// fetch talks to the remotes for a while, and a register that lands meanwhile
/// must still be there afterwards.
pub fn fetch_members(
reg: &Registry,
root: &Path,
name: &str,
only: Option<&[String]>,
mut on_member: impl FnMut(&MemberOutcome),
) -> Result<FetchReport> {
let mut ws = reg
.get(name)?
.ok_or_else(|| anyhow!("no workspace `{name}` in the registry"))?;
let ws_root = root.join(&ws.name);
let git_dir = ws_root.join("git");
std::fs::create_dir_all(&git_dir)?;
std::fs::create_dir_all(ws_root.join("builds"))?;
let now = chrono::Utc::now().to_rfc3339();
let mut report = FetchReport {
workspace: ws.name.clone(),
..Default::default()
};
for m in &mut ws.members {
if m.remote.is_empty() || only.is_some_and(|o| !o.contains(&m.name)) {
continue; // local/external member — source lives outside; or not asked for
}
report.fetched += 1;
let dest = git_dir.join(&m.name);
// SSH remotes authenticate with the server's deploy key (resolved inside
// clone_or_fetch via NORNIR_SSH_DIR / the `nornir` user's home).
let t0 = std::time::Instant::now();
// A member that names its branch is read from THAT branch, not from
// whatever the forge calls its default (git.gunnar.rs answered knut's
// `ny_dok` and znippy's stale `main` after the 2026-10-14 move).
let branch = m.git_ref.trim().to_string();
let outcome = crate::procreap::clone_or_fetch(&m.remote, &dest, None)
.and_then(|sha| if branch.is_empty() { Ok(sha) } else { pin_member_branch(&dest, &branch) });
let elapsed_ms = t0.elapsed().as_millis() as i64;
match outcome {
Ok(sha) => {
let mo = MemberOutcome {
member: m.name.clone(),
remote: m.remote.clone(),
op: "clone-fetch".into(),
status: "ok".into(),
detail: sha.clone(),
elapsed_ms,
};
on_member(&mo); // STREAM: this member's clone landed (ok).
report.outcomes.push(mo);
if sha != m.last_seen_sha {
report.changed.push(m.name.clone());
}
m.last_seen_sha = sha;
m.last_synced = now.clone();
m.sync_state = "ok".into();
// AUT6 freshness: record the clone's working-tree digest so a
// thin client can detect staleness the SHA can't see. A freshly
// fetched server clone is normally clean; best-effort (a digest
// error must not fail the fetch).
match crate::gitio::worktree_freshness(&dest) {
Ok(f) => {
m.worktree_digest = f.digest;
m.worktree_dirty = f.dirty;
}
Err(e) => eprintln!(
"nornir-monitor: {}::{} worktree freshness skipped: {e:#}",
report.workspace, m.name
),
}
}
Err(e) => {
let msg = format!("{e:#}");
m.sync_state = format!("error: {msg}");
let mo = MemberOutcome {
member: m.name.clone(),
remote: m.remote.clone(),
op: "clone-fetch".into(),
status: "error".into(),
detail: msg.clone(),
elapsed_ms,
};
on_member(&mo); // STREAM: this member's clone landed (error).
report.outcomes.push(mo);
report.errors.push((m.name.clone(), msg));
}
}
}
// only what this fetch touched, into the row as it is NOW
let touched: Vec<_> = ws.members.iter().filter(|m| !m.remote.is_empty() && only.is_none_or(|o| o.contains(&m.name))).cloned().collect();
reg.merge_member_states(name, &touched)?;
Ok(report)
}
/// What ONE embed pass over a workspace's present members actually did.
///
/// Exists because the embed step used to report a bare `usize` ("members
/// embedded") — which cannot answer the two questions an operator asks after a
/// run that was killed on temperature: *how much of this is already on disk*, and
/// *did anything fail*. `occurrences - new_vectors` is the RESUMED count: content
/// whose vector was already stored for this model profile and therefore was NOT
/// re-embedded (see [`EmbedReport::reused_vectors`]).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct EmbedReport {
pub workspace: String,
/// Members with a materialized `.git` checkout that the pass considered.
pub members_present: usize,
/// Members whose snapshot + manifest are committed after this pass.
pub members_embedded: usize,
/// Members whose embed failed (isolated — never aborts the other members).
pub members_failed: usize,
/// Total chunk OCCURRENCES across the embedded members (not distinct content).
pub occurrences: usize,
/// Vectors newly written by THIS pass. `0` on a fully-resumed re-run.
pub new_vectors: usize,
/// `Some(why)` when the WHOLE step was skipped (no embedder compiled, or the
/// embedder failed to load). Never `None` on a silent skip — a hollow Mímir
/// must always carry its confession.
pub skipped: Option<String>,
pub elapsed_ms: i64,
}
impl EmbedReport {
/// PURE — content that did NOT need re-embedding this pass, i.e. what a
/// previous (possibly interrupted) run already put on disk and this run
/// RESUMED from. A first run over an empty `embeddings` table yields `0`; a
/// clean re-run yields `occurrences` (every chunk already stored).
#[inline]
pub fn reused_vectors(&self) -> usize {
self.occurrences.saturating_sub(self.new_vectors)
}
/// PURE — the per-member progress line. Rendered as its own function (and
/// unit-tested) because it is the ONLY evidence an operator has while a
/// multi-hour CPU embed is running: member name, chunk count, new vectors,
/// and BOTH the per-member and the cumulative elapsed. A run that prints only
/// a total at the end is indistinguishable from a hung one — that is the
/// Unknown-≠-Idle failure this whole step was reshaped around.
pub fn member_line(
ws: &str,
member: &str,
idx: usize,
total: usize,
occurrences: usize,
new_vectors: usize,
member_secs: f64,
total_secs: f64,
) -> String {
let reused = occurrences.saturating_sub(new_vectors);
format!(
"nornir-embed: {ws}/{member} [{idx}/{total}] {occurrences} chunk(s) · \
{new_vectors} new vector(s) · {reused} reused · {member_secs:.1}s \
(cumulative {total_secs:.1}s) — COMMITTED"
)
}
/// PURE — the report a pass yields when it did **not run at all**: no embedder
/// compiled, or the embedder failed to load. Both loud-skip branches build the
/// report through here so neither can drift into looking like a green pass with
/// zero members — which is exactly how a hollow Mímir stayed invisible before
/// the loud skip existed. `skipped` is ALWAYS `Some`, and every count is zero:
/// the CLI turns that into a non-zero exit rather than a "✓".
pub fn skipped_pass(ws_name: &str, members_present: usize, why: impl Into<String>) -> Self {
Self {
workspace: ws_name.to_string(),
members_present,
members_embedded: 0,
members_failed: 0,
occurrences: 0,
new_vectors: 0,
skipped: Some(why.into()),
elapsed_ms: 0,
}
}
}
/// Populate **step 3b** AND the on-demand `nornir workspace embed` — embed every
/// present member's own sources into the warehouse's vector tables (`embeddings` /
/// `embedding_snapshots` / `embedding_manifest`), so a server built with an
/// embedder fills them and HNSW has vectors to index at query time. This is the
/// same write path `nornir map` uses (`vector::store::index_repos_batched`) — ONE
/// writer, no second embed writer anywhere in the tree. Best-effort per member (a
/// failed embed records an event, never fails the populate).
///
/// ## Why this loops PER MEMBER instead of handing every member to one batch
///
/// It used to build ONE `index_repos_batched` call over all members. Inside that
/// call the *vectors* are durable per 2048-chunk slice, but the per-member
/// `embedding_manifest` + `embedding_snapshots` appends all happen in phase 3, at
/// the very END of the pass. A run killed in the middle therefore left a pile of
/// content-addressed vectors that NO snapshot pointed at — queryable by nothing,
/// creditable to no member. Measured on this box: three consecutive nordisk embed
/// runs were killed mid-flight on thermals and each one left the workspace with
/// zero usable members, having burned the whole elapsed time.
///
/// Calling the SAME writer with a one-member slice moves the manifest + snapshot
/// commit to a per-member boundary, so a kill after member 7 keeps members 1–7 in
/// full. What it costs is honest and small:
///
/// * `existing_content_hashes` is re-read once per member instead of once per
/// pass (a projected single-column scan of `embeddings`) — O(members) scans.
/// Cheap next to an embed that is 96.9 % of populate wall (`oden`,
/// 2026-07-26 measurement).
/// * cross-member forward-batch packing is given up for members smaller than
/// `EMBED_SLICE_CHUNKS` (2048). This is NOT the throughput lever: ingest is
/// bound by the inference device, which serialises on one `ort::Session`
/// regardless of how the caller slices its input, and every embedder does its
/// own `plan_batches` bucketing inside `embed`.
///
/// Durability beats a batching micro-win that the measurements say does not
/// exist. The chunking pool inside `index_repos_batched` still saturates every
/// core for the member being processed, and each member's sources are DROPPED
/// after its commit, so peak resident text no longer grows with member count.
///
/// ## Two halves: the read, and the embed
///
/// `read_member_sources` is the only part that looks at the members' trees
/// (seconds of I/O); `embed_member_sources` turns what was read into vectors
/// and looks at no tree. A caller that must not hold a tree still for as long
/// as an embedder is slow (the code room's consumer) takes the first under its
/// lock and runs the second with nothing held ([`read_embed_inputs`] /
/// [`embed_read_inputs`]).
///
/// The READ half: every present member's Rust sources as they are on disk NOW,
/// with the commit they are of.
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
fn read_member_sources(ws_name: &str, git_dir: &Path, present: &[&String]) -> Vec<crate::vector::store::RepoInput> {
// CPU (parallel): read each member's Rust sources in ONE gatling pool over the
// members (I/O-bound directory walks, independent per member). The walk stays
// fanned out — it is the embed COMMIT, not the source read, that had to become
// per-member. `gatling_forkjoin`, never rayon (workspace law).
let inputs: Vec<crate::vector::store::RepoInput> =
znippy_zoomies::gatling_forkjoin::gatling_map_balanced(
present,
0,
1,
|_t| 1u64,
|_, name| {
let repo_root = git_dir.join(name.as_str());
let (sha, branch) = crate::gitio::head_sha_and_branch(&repo_root)
.unwrap_or_else(|_| ("unknown".into(), "unknown".into()));
let files = crate::vector::store::collect_rust_sources(&repo_root);
crate::vector::store::RepoInput {
workspace: ws_name.to_string(),
repo: (*name).clone(),
git_sha: sha,
branch,
complete: true,
files,
}
},
);
// Members with no Rust sources contribute nothing (kept out so a zero-chunk
// snapshot isn't written), matching the long-standing `files.is_empty()` skip.
inputs
.into_iter()
.filter(|inp| !inp.files.is_empty())
.collect()
}
/// The EMBED half (see `read_member_sources`): sources read earlier
/// ([`read_member_sources`]) → vectors, one commit per member. Touches no
/// member's tree.
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
fn embed_member_sources(
wh: &crate::warehouse::iceberg::IcebergWarehouse,
ws_name: &str,
present: usize,
batch_inputs: Vec<crate::vector::store::RepoInput>,
wsevents: &mut crate::warehouse::workspace_events::WorkspaceEventLog,
) -> EmbedReport {
use crate::warehouse::workspace_events::{status as ws_status, step as ws_step};
use std::time::Instant;
let mut rep = EmbedReport {
workspace: ws_name.to_string(),
members_present: present,
..Default::default()
};
let embedder = match crate::vector::load_embedder() {
Ok(e) => e,
Err(e) => {
eprintln!(
"nornir-monitor: {ws_name} ⚠ EMBED SKIPPED — embedder load failed: {e:#} \
(embeddings/HNSW stay EMPTY — Mímir semantic search is HOLLOW)"
);
let why = format!("embedder load failed: {e:#}");
wsevents.record_workspace(ws_step::EMBED, ws_status::ERROR, why.clone(), 0);
return EmbedReport::skipped_pass(ws_name, present, why);
}
};
eprintln!(
"nornir-monitor: {ws_name} embedding {} member(s) with {} …",
batch_inputs.len(),
crate::vector::embedder_backend()
);
// A CPU fallback on a box that declared a GPU is a DEFECT, not a graceful
// degradation (~67× slower here) — so say WHY, loudly, and record it as an
// event so the warehouse carries the confession too. Silent on a genuine
// CPU server, where tract is the right answer.
if let Some(why) = crate::vector::fallback::live_reason() {
eprintln!("nornir-monitor: {ws_name} {why}");
wsevents.record_workspace(
ws_step::EMBED,
ws_status::SKIPPED,
format!(
"GPU embedding skipped: {}",
why.lines().take(3).collect::<Vec<_>>().join(" ")
),
0,
);
}
let opts = crate::vector::chunk::ChunkOptions::default();
let total = batch_inputs.len();
let t_all = Instant::now();
// `into_iter` on purpose: each member's source text is dropped the moment its
// snapshot is committed, so peak resident text is ONE member's corpus, not the
// whole workspace's (the 7.3 GB RSS shape measured on t14s, 2026-07-27).
for (i, inp) in batch_inputs.into_iter().enumerate() {
let name = inp.repo.clone();
let t_member = Instant::now();
// The SAME writer, called with a one-member slice. `index_repos_batched`
// re-reads `existing_content_hashes` for this model profile on every call,
// which is exactly what makes the re-run RESUMABLE: content already stored
// by an earlier (or interrupted) pass is never re-embedded.
let res = crate::vector::store::index_repos_batched(
wh,
std::slice::from_ref(&inp),
&opts,
&*embedder,
)
.pop()
.unwrap_or_else(|| Err(anyhow!("embed writer returned no result for {name}")));
let member_ms = t_member.elapsed().as_millis() as i64;
match res {
Ok(snap) => {
rep.members_embedded += 1;
rep.occurrences += snap.occurrences;
rep.new_vectors += snap.new_vectors;
// PER-MEMBER, PER-COMMIT progress — printed AFTER the manifest +
// snapshot append, so the line is a statement about what is on
// disk, not about what is in flight.
eprintln!(
"{}",
EmbedReport::member_line(
ws_name,
&name,
i + 1,
total,
snap.occurrences,
snap.new_vectors,
t_member.elapsed().as_secs_f64(),
t_all.elapsed().as_secs_f64(),
)
);
wsevents.record(
name.as_str(),
ws_step::EMBED,
ws_status::OK,
format!("{} occ, {} new vec", snap.occurrences, snap.new_vectors),
member_ms,
);
}
Err(e) => {
rep.members_failed += 1;
eprintln!(
"nornir-embed: {ws_name}/{name} [{}/{total}] embed FAILED after {:.1}s: {e:#} \
(the members already committed above are KEPT — re-run to resume)",
i + 1,
t_member.elapsed().as_secs_f64()
);
wsevents.record(
name.as_str(),
ws_step::EMBED,
ws_status::ERROR,
format!("{e:#}"),
member_ms,
);
}
}
}
rep.elapsed_ms = t_all.elapsed().as_millis() as i64;
rep
}
/// No-embedder build: the embed step SKIPS **loudly** — a warehouse populated by
/// this binary will have EMPTY `embeddings`/`embedding_snapshots`/HNSW. The whole
/// point of the loud skip (eprintln + a `workspace_events` SKIPPED row) is that a
/// hollow Mímir is never silently green. Rebuild `--features embed-ort` (GPU) or
/// `embed-tract` (CPU) to fill the vectors.
#[cfg(not(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
)))]
fn embed_member_sources(
_wh: &crate::warehouse::iceberg::IcebergWarehouse,
ws_name: &str,
present: usize,
_inputs: (),
wsevents: &mut crate::warehouse::workspace_events::WorkspaceEventLog,
) -> EmbedReport {
use crate::warehouse::workspace_events::{status as ws_status, step as ws_step};
const WHY: &str = "built without embedder (embed-tract/embed-ort)";
eprintln!(
"nornir-monitor: {ws_name} ⚠ EMBED SKIPPED — this binary was built WITHOUT an embedder \
(no `embed-tract`/`embed-ort`). embeddings/embedding_snapshots/HNSW stay EMPTY — Mímir \
semantic search is HOLLOW. Rebuild `--features embed-ort` (GPU) or `embed-tract` (CPU)."
);
wsevents.record_workspace(ws_step::EMBED, ws_status::SKIPPED, WHY, 0);
EmbedReport::skipped_pass(ws_name, present, WHY)
}
/// What an embed pass READS of the members' trees before it embeds — taken by
/// [`read_embed_inputs`] while nothing moves them, embedded by
/// [`embed_read_inputs`] with nothing held.
pub struct EmbedInputs {
present: usize,
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
sources: Vec<crate::vector::store::RepoInput>,
#[cfg(not(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
)))]
sources: (),
}
/// The READ half of [`embed_workspace_with`]: the Rust sources of every member
/// of `members` that has a checkout under `git_dir`, as they are on disk now.
/// The same refusal as the whole verb when no member is there.
pub fn read_embed_inputs(git_dir: &Path, members: &[String], ws_name: &str) -> Result<EmbedInputs> {
let present = present_members(members, |name| git_dir.join(name).join(".git").exists());
if present.is_empty() {
bail!(
"no materialized member checkouts under {} — 0 of {} member(s) have a `.git` clone, \
so there is nothing to embed. Run `nornir workspace fetch {ws_name}` first and check \
`nornir workspace events {ws_name}` for the per-member clone errors.",
git_dir.display(),
members.len(),
);
}
Ok(EmbedInputs {
present: present.len(),
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
sources: read_member_sources(ws_name, git_dir, &present),
#[cfg(not(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
)))]
sources: (),
})
}
/// The EMBED half of [`embed_workspace_with`]: what [`read_embed_inputs`] read →
/// vectors, through a caller-owned warehouse handle. Looks at no member's tree,
/// so it needs none of the locks a fetch or a consume takes — it is as long as
/// the embedder is slow, and nothing waits for it.
pub fn embed_read_inputs(
wh: &crate::warehouse::iceberg::IcebergWarehouse,
ws_name: &str,
inputs: EmbedInputs,
) -> EmbedReport {
use std::time::Instant;
let mut wsevents = crate::warehouse::workspace_events::WorkspaceEventLog::new(ws_name);
let t = Instant::now();
let mut rep = embed_member_sources(wh, ws_name, inputs.present, inputs.sources, &mut wsevents);
rep.elapsed_ms = t.elapsed().as_millis() as i64;
// Roll the pass up as its own workspace-level EMBED row so
// `nornir workspace events` shows the on-demand pass, not only populate's.
// Flushed ONCE at the end: `WorkspaceEventLog::flush` appends every buffered
// row, so calling it per member would duplicate the earlier ones. The EVENTS
// are the log; the VECTORS are the durable artefact, and those already commit
// per member.
wsevents.record_workspace(
crate::warehouse::workspace_events::step::EMBED,
if rep.members_failed > 0 {
crate::warehouse::workspace_events::status::ERROR
} else {
crate::warehouse::workspace_events::status::OK
},
format!(
"{}/{} member(s) embedded, {} occ, {} new vec, {} reused",
rep.members_embedded,
rep.members_present,
rep.occurrences,
rep.new_vectors,
rep.reused_vectors(),
),
rep.elapsed_ms,
);
wsevents.flush(wh);
rep
}
/// The reason this BINARY can never embed, or `None` when it can. Compiled from
/// the same `#[cfg]` the two `embed_member_sources` bodies are selected by, so
/// the answer cannot drift from which body is actually linked. Public to the crate
/// so a caller can refuse the verb up front instead of discovering it after a
/// warehouse open.
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
pub(crate) fn no_embedder_reason() -> Option<&'static str> {
None
}
/// See [`no_embedder_reason`] — the no-embedder half.
#[cfg(not(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
)))]
pub(crate) fn no_embedder_reason() -> Option<&'static str> {
Some("built without embedder (embed-tract/embed-ort)")
}
/// PURE — the members an embed pass will actually touch: those with a
/// materialized `.git` checkout, in registry order. Split out of
/// [`embed_workspace_with`] so the "which members" decision is testable without a
/// filesystem, and so it can never silently diverge from [`republish_with`]'s
/// identical `present` filter.
pub(crate) fn present_members<'a>(
members: &'a [String],
has_checkout: impl Fn(&str) -> bool,
) -> Vec<&'a String> {
members
.iter()
.filter(|name| has_checkout(name.as_str()))
.collect()
}
/// **`nornir workspace embed <name>` — the on-demand, idempotent, resumable embed.**
///
/// THE DEFECT THIS FIXES (measured 2026-08-14, `nordisk`): the embed step only ran
/// as part of a full populate/republish pass, and on an UNCHANGED tree the monitor
/// short-circuits long before it. `nornir workspace populate nordisk` returned
/// "19 member(s) cloned; warehouse building in background" and the journal then
/// showed nothing but "serving nordisk / poll loop every 60s" — no scan, no embed,
/// no republish. Consequence: a warehouse whose `embeddings` table is EMPTY (0
/// parquet files, against 74 for `symbol_facts`) could not be filled at all except
/// by a purge + re-register, which re-pays a 200–700 s rust-analyzer resolved
/// ingest FIRST, for nothing.
///
/// This runs ONLY the embed step over the CURRENT snapshot of an already-registered
/// workspace: no fetch, no knowledge scan, no resolved (SCIP) ingest, no deep-scan,
/// no republish, no new index snapshot. Idempotent — a second run over an unchanged
/// tree re-embeds nothing (`index_repos_batched` short-circuits on an identical
/// `(repo, git_sha, model_profile, chunker_hash, fileset_hash)` snapshot, and any
/// content already stored for the model profile is filtered out by
/// `existing_content_hashes`). Resumable — see `embed_member_sources`: each
/// member commits its own manifest + snapshot, so an interrupted run keeps every
/// member it finished.
///
/// Opens the workspace's own warehouse for WRITE. skade-katalog is single-writer,
/// so this cannot run while `nornir-server` holds `catalog.redb` — the lock error
/// is translated into an operator-actionable message rather than a redb string.
pub fn embed_workspace(reg: &Registry, root: &Path, ws_name: &str) -> Result<EmbedReport> {
use crate::warehouse::iceberg::IcebergWarehouse;
let ws_root = root.join(ws_name);
let git_dir = ws_root.join("git");
let builds = ws_root.join("builds");
// Refuse BEFORE touching the warehouse. A binary with no embedder compiled can
// only ever produce the loud skip, and taking the single-writer `catalog.redb`
// lock to do it would evict the running server for nothing.
if let Some(why) = no_embedder_reason() {
bail!(
"this `nornir` binary cannot embed: {why}. embeddings/embedding_snapshots/HNSW would \
stay EMPTY and Mímir semantic search HOLLOW. Rebuild `--features embed-ort` (GPU) or \
`embed-tract` (CPU) and run this again — nothing was opened or written."
);
}
let record = reg
.get(ws_name)?
.ok_or_else(|| anyhow!("no workspace `{ws_name}` in the registry — register it first"))?;
let members: Vec<String> = record.members.iter().map(|m| m.name.clone()).collect();
if !builds.join("catalog.redb").exists() {
bail!(
"`{ws_name}` has no warehouse at {} yet — embed operates on an EXISTING snapshot; \
run `nornir workspace populate {ws_name}` first",
builds.display()
);
}
let wh = IcebergWarehouse::open(&builds).map_err(|e| {
if crate::warehouse::iceberg::is_catalog_lock_error(&e) {
anyhow!(
"the warehouse catalog at {} is LOCKED by another process — almost certainly the \
running `nornir-server`. skade-katalog is single-writer, so an out-of-process \
embed cannot write while the server holds it. Stop the server \
(`systemctl stop nornir-server`), run this, and start it again. \
(underlying: {e:#})",
builds.display()
)
} else {
e.context(format!("open warehouse {}", builds.display()))
}
})?;
embed_workspace_with(&wh, &git_dir, &members, ws_name)
}
/// [`embed_workspace`] through a CALLER-OWNED warehouse handle — the shape the
/// server would use if the verb is ever wired to an RPC (the server must be the
/// sole opener of `catalog.redb`, exactly as [`republish_with`] documents).
pub fn embed_workspace_with(
wh: &crate::warehouse::iceberg::IcebergWarehouse,
git_dir: &Path,
members: &[String],
ws_name: &str,
) -> Result<EmbedReport> {
let inputs = read_embed_inputs(git_dir, members, ws_name)?;
Ok(embed_read_inputs(wh, ws_name, inputs))
}
/// The poll-cadence gate moved to the `nornir-poll-cadence` edda leaf (shrink
/// plan Tier-1): the pure, chrono-only `parse_interval` and `is_workspace_due`
/// predicates are re-exported verbatim, so `monitor::parse_interval` /
/// `monitor::is_workspace_due` call sites (e.g. `server_grpc`'s sweep) resolve
/// unchanged. See `../edda/crates/nornir-poll-cadence`.
pub use nornir_poll_cadence::{is_workspace_due, parse_interval};
/// The most-recent member sync time for a workspace (`max` over members'
/// `last_synced`), or `None` when no member has ever synced (empty/garbage
/// stamps parse away). Feeds [`is_workspace_due`].
///
/// The registry-typed wrapper stays in nornir — a leaf may not touch the
/// `crate::registry::Workspace` domain type — while the pure chrono fold it
/// delegates to lives in the leaf as [`nornir_poll_cadence::latest_synced`].
pub fn workspace_last_synced(
ws: &crate::registry::Workspace,
) -> Option<chrono::DateTime<chrono::Utc>> {
nornir_poll_cadence::latest_synced(ws.members.iter().map(|m| m.last_synced.as_str()))
}
/// Point `refs/heads/<branch>` at the commit the fetch brought for it
/// (`refs/remotes/origin/<branch>`), forced — a rewritten remote is followed,
/// not refused — and answer that commit. The code room's clone reads
/// `refs/heads/*` of this checkout, so this is what makes a descriptor's
/// `branch` the line it reads. A remote without that branch is an error that
/// names it.
pub(crate) fn pin_member_branch(dest: &Path, branch: &str) -> Result<String> {
let mut repo = gix::open(dest).with_context(|| format!("open {}", dest.display()))?;
// The ref write logs a reflog line, which needs a committer; a service
// user has no git identity (seen: the test failed once another test moved
// HOME). Name the fetch itself when nothing else is configured.
if repo.committer().is_none() {
use gix::config::tree::{Committer, Key as _};
let mut config = repo.config_snapshot_mut();
config.set_raw_value(&Committer::NAME, "nornir fetch").map_err(|e| anyhow!("set committer.name: {e:?}"))?;
config.set_raw_value(&Committer::EMAIL, "nornir@localhost").map_err(|e| anyhow!("set committer.email: {e:?}"))?;
}
let remote = format!("refs/remotes/origin/{branch}");
let id = repo
.try_find_reference(remote.as_str())
.with_context(|| format!("look up {remote}"))?
.ok_or_else(|| anyhow!("the remote of {} has no branch `{branch}` (the descriptor names it)", dest.display()))?
.into_fully_peeled_id()
.with_context(|| format!("peel {remote}"))?
.detach();
repo.reference(
format!("refs/heads/{branch}"),
id,
gix::refs::transaction::PreviousValue::Any,
format!("nornir fetch: {branch} follows origin"),
)
.with_context(|| format!("set refs/heads/{branch} in {}", dest.display()))?;
Ok(id.to_string())
}
#[cfg(test)]
mod tests {
/// A fetched checkout whose forge default is another branch: the named
/// branch's local ref follows origin's — forced, across a rewrite — and a
/// branch the remote lacks is an error naming it.
#[test]
fn a_named_branch_follows_origin_even_across_a_rewrite() {
// No git identity anywhere — the server's service user has none.
let dir = std::env::temp_dir().join(format!("nornir-pin-branch-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let repo = gix::init(&dir).unwrap();
let sig = gix::actor::Signature { name: "t".into(), email: "t@x".into(), time: gix::date::Time::new(1, 0) };
let commit = |r: &str, msg: &str| {
let (mut a, mut c) = (gix::date::parse::TimeBuf::default(), gix::date::parse::TimeBuf::default());
repo.commit_as(sig.to_ref(&mut c), sig.to_ref(&mut a), r, msg, repo.empty_tree().id, gix::commit::NO_PARENT_IDS)
.unwrap()
.detach()
};
let old = commit("refs/heads/main", "old main");
let _default = commit("refs/remotes/origin/ny_dok", "the forge default");
let new = commit("refs/remotes/origin/main", "rewritten main");
assert_ne!(old, new);
assert_eq!(super::pin_member_branch(&dir, "main").unwrap(), new.to_string());
let head = repo.find_reference("refs/heads/main").unwrap().into_fully_peeled_id().unwrap().detach();
assert_eq!(head, new, "refs/heads/main follows origin/main, forced over the old id");
let e = super::pin_member_branch(&dir, "master").unwrap_err().to_string();
assert!(e.contains("no branch `master`"), "{e}");
let _ = std::fs::remove_dir_all(&dir);
}
use super::*;
// NOTE: `is_workspace_due` / `parse_interval` are pure predicates that moved
// to the `nornir-poll-cadence` edda leaf (shrink plan Tier-1); their unit
// tests (`is_workspace_due_honors_poll_cadence`, `interval_parsing`) moved
// with them. The registry-typed `workspace_last_synced` wrapper stays here,
// so its test does too.
/// **The gate must actually serialize — measured, not asserted structurally.**
///
/// Eight threads race through `enter`, each recording how many are inside at
/// once. A working gate can only ever observe 1.
///
/// The RED PROBE runs the identical closures WITHOUT the gate and asserts the
/// observed peak is >1 — so this test knows the difference between "the gate
/// held" and "the threads happened not to overlap", which is the failure mode
/// that would make it a green nobody has seen red.
#[test]
fn the_workspace_gate_admits_one_at_a_time_and_the_probe_without_it_does_not() {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
// A shared counter standing in for `<root>/<ws>/git/` — the thing two
// fetches trample. `observed` is the peak simultaneous occupancy.
fn race(body: impl Fn(&dyn Fn()) + Sync + Send + 'static) -> usize {
let inside = Arc::new(AtomicUsize::new(0));
let observed = Arc::new(AtomicUsize::new(0));
let body = Arc::new(body);
std::thread::scope(|sc| {
for _ in 0..8 {
let inside = inside.clone();
let observed = observed.clone();
let body = body.clone();
sc.spawn(move || {
body(&|| {
let n = inside.fetch_add(1, SeqCst) + 1;
observed.fetch_max(n, SeqCst);
// Long enough that unguarded threads WILL overlap.
std::thread::sleep(std::time::Duration::from_millis(25));
inside.fetch_sub(1, SeqCst);
});
});
}
});
observed.load(SeqCst)
}
// RED PROBE FIRST: no gate at all — the state before this fix.
let unguarded = race(|work| work());
assert!(
unguarded > 1,
"the probe never overlapped, so a green below would prove nothing \
(observed peak {unguarded})"
);
// The gate.
let gate = Arc::new(WorkspaceGate::new());
let g = gate.clone();
let guarded = race(move |work| g.enter(|| work()));
assert_eq!(
guarded, 1,
"two callers were inside the gate at once — this is the eager-vs-sweep \
race that wrote `fail workspace_clone skade 13 ms` for a fetch that \
succeeded"
);
assert_eq!(gate.peak_in_flight(), 1, "the gate's own counter must agree");
assert!(
gate.waited() > 0,
"a loser must WAIT, not fail — and the gate must be able to say it did"
);
}
/// **A long job that takes the gate once per unit must not starve a waiter.**
///
/// One thread takes the gate 200 times in a row, each hold a unit of work (a
/// backfill: a commit per hold). A second thread asks for the gate once, in
/// the middle. Counted: the units that began AFTER the waiter was queued and
/// before it got in. With `give_way` between the units that is none — the
/// unit the job was inside when the waiter queued is the one unit it waits
/// for; the gate is a plain mutex, so without it the looping thread takes the
/// gate again before the waiter is even scheduled.
///
/// No clock in it: the TEST runs the job's units. Each unit says it began and
/// waits for the test to end it, so the waiter is queued while the job is
/// parked INSIDE a unit — the only caller that can be in `enter`'s queue then
/// is the waiter (the job's own pass through `enter` would count as queued
/// too; that is what made the old wall-clock version start its count early
/// and fail under load). The test ends units until the waiter says it is in.
///
/// The RED PROBE is the same loop without `give_way`: it must overtake the
/// waiter by more than that, or the green below would prove nothing.
#[test]
fn a_job_that_gives_way_between_its_units_lets_a_waiter_in_after_one_unit_and_the_probe_does_not() {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::mpsc;
enum Ev {
/// The job is inside unit n (and holds it until told).
Began(usize),
/// The waiter is inside; the job had begun this many units.
WaiterIn(usize),
}
const UNITS: usize = 200;
/// Units the looping thread began after the waiter queued and before
/// the waiter was inside.
fn overtaken(give_way: bool) -> usize {
let gate = Arc::new(WorkspaceGate::new());
let begun = Arc::new(AtomicUsize::new(0));
let (ev_tx, ev) = mpsc::channel::<Ev>();
let (end_tx, end_rx) = mpsc::channel::<()>();
let (g, b, tx) = (gate.clone(), begun.clone(), ev_tx.clone());
let job = std::thread::spawn(move || {
for n in 1..=UNITS {
if give_way {
g.give_way();
}
let go_on = g.enter(|| {
b.store(n, SeqCst);
let _ = tx.send(Ev::Began(n));
end_rx.recv().is_ok()
});
if !go_on {
return;
}
}
});
let unit = |ev: &mpsc::Receiver<Ev>| match ev.recv().expect("the job is running") {
Ev::Began(n) => n,
Ev::WaiterIn(_) => unreachable!("no waiter yet"),
};
// a few units go by with nobody waiting
for _ in 0..3 {
unit(&ev);
end_tx.send(()).unwrap();
}
// the job is inside this unit, and stays there until the test says
let at_queue = unit(&ev);
let (g, b) = (gate.clone(), begun.clone());
let waiter = std::thread::spawn(move || {
let at = g.enter(|| b.load(SeqCst));
let _ = ev_tx.send(Ev::WaiterIn(at));
});
// the job holds the gate: whoever is queued now is the waiter
while gate.queued() == 0 {
std::thread::yield_now();
}
end_tx.send(()).unwrap();
// end every unit the job begins until the waiter is in
let at_entry = loop {
match ev.recv().expect("the waiter gets in at the latest when the job is done") {
Ev::Began(_) => {
let _ = end_tx.send(());
}
Ev::WaiterIn(at) => break at,
}
};
// let the job finish its units
drop(end_tx);
waiter.join().unwrap();
job.join().unwrap();
at_entry - at_queue
}
// RED PROBE FIRST: a plain mutex lets the looping thread barge. One
// attempt can get lucky (the waiter's wake-up wins the race), so the
// probe runs until one attempt shows it — bounded, and fast: no sleeps.
let barged = (0..50).map(|_| overtaken(false)).find(|&n| n > 2);
assert!(
barged.is_some(),
"in 50 attempts the probe never starved the waiter by more than 2 units, \
so a green below would prove nothing"
);
for _ in 0..20 {
let n = overtaken(true);
assert_eq!(
n, 0,
"{n} units of the job began after the waiter queued and before it got in — \
a waiter must wait for ONE unit at most (the one the job was inside)"
);
}
}
/// A panic inside the gate must not wedge it: the next caller still gets in,
/// and the in-flight counter is not left leaking.
#[test]
fn a_panicking_caller_does_not_wedge_the_gate() {
let gate = WorkspaceGate::new();
let boom = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
gate.enter(|| panic!("producer blew up"));
}));
assert!(boom.is_err(), "the probe must actually have panicked");
// The mutex is now POISONED. A gate that refused here would turn one bad
// fetch into a permanently dead workspace.
assert_eq!(gate.enter(|| 41 + 1), 42, "the gate must recover from poison");
assert_eq!(gate.peak_in_flight(), 1, "the panicking entry leaked its slot");
}
#[test]
fn workspace_last_synced_picks_latest_member() {
use crate::registry::{MemberState, Mode, Workspace};
let mk = |last: &str| MemberState {
name: "m".into(),
last_synced: last.into(),
..Default::default()
};
let ws = |members: Vec<MemberState>| Workspace {
name: "w".into(),
mode: Mode::Monitored,
descriptor: String::new(),
poll: "60s".into(),
current_snapshot: String::new(),
members,
descriptor_content: String::new(),
created_at: String::new(),
updated_at: String::new(),
};
let got = workspace_last_synced(&ws(vec![
mk("2026-06-01T00:00:00Z"),
mk("2026-06-10T00:00:00Z"),
mk(""), // never-synced member ignored
]))
.unwrap();
assert_eq!(got.to_rfc3339(), "2026-06-10T00:00:00+00:00");
// No member ever synced → None.
assert!(workspace_last_synced(&ws(vec![mk("")])).is_none());
}
// `interval_parsing` moved to the nornir-poll-cadence leaf with `parse_interval`.
// ── on-demand embed (`nornir workspace embed`) ───────────────────────────
//
// The measured defect, 2026-08-14 on `nordisk`: `embeddings` had 0 parquet
// files against 74 for `symbol_facts`, and could not be filled — the embed
// only ran inside a full populate/republish, which short-circuits on an
// unchanged tree. These tests pin the three properties the fix has to carry:
// WHICH members a pass touches, that a re-run RESUMES rather than restarts,
// and that progress is reported PER MEMBER (the last three runs were killed
// mid-flight and, with an end-of-pass commit, kept nothing).
/// Inject-assert: the embed pass touches exactly the members with a
/// materialized checkout, in registry order — the same `present` filter
/// [`republish_with`] applies, so the two can never disagree about what a
/// workspace's member set is. `has_checkout` is INJECTED: no filesystem.
#[test]
fn embed_touches_only_materialized_members_in_registry_order() {
let members: Vec<String> = ["tunnr", "holger", "nornir", "facett"]
.iter()
.map(|s| s.to_string())
.collect();
// `tunnr`'s clone never materialised (the 2026-07 ssh early-eof state);
// `facett` has a broken manifest but a real checkout, so it IS embedded.
let present = present_members(&members, |n| n != "tunnr");
assert_eq!(
present.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
vec!["holger", "nornir", "facett"],
"a member with no checkout must be dropped, and order must be preserved"
);
// The empty case is what `embed_workspace_with` refuses on — a pass over
// zero members must be visible as zero, never as a silent success.
assert!(present_members(&members, |_| false).is_empty());
assert_eq!(present_members(&members, |_| true).len(), 4);
}
/// Inject-assert on the RESUME arithmetic. `index_repos_batched` filters every
/// chunk whose content is already stored for the active model profile
/// (`existing_content_hashes`), so `new_vectors` counts ONLY what this pass
/// wrote. The difference is what an earlier — possibly killed — pass had
/// already put on disk, and it is the number that proves a re-run continued
/// instead of restarting.
#[test]
fn a_rerun_reports_reuse_not_rework() {
// First pass over an EMPTY embeddings table: every chunk is new.
let first = EmbedReport {
occurrences: 1842,
new_vectors: 1842,
..Default::default()
};
assert_eq!(
first.reused_vectors(),
0,
"a cold pass resumes from nothing"
);
// Killed at 60 %, then re-run: the survivors are reused, not re-embedded.
let resumed = EmbedReport {
occurrences: 1842,
new_vectors: 737,
..Default::default()
};
assert_eq!(resumed.reused_vectors(), 1105);
// A clean re-run over an unchanged tree must embed NOTHING. This is the
// idempotence claim, stated as arithmetic.
let idempotent = EmbedReport {
occurrences: 1842,
new_vectors: 0,
..Default::default()
};
assert_eq!(idempotent.reused_vectors(), 1842);
assert_eq!(idempotent.new_vectors, 0);
// Never negative, never a panic: a writer that ever credited more new
// vectors than occurrences would be a bug, but it must not take the
// reporting down with it.
let impossible = EmbedReport {
occurrences: 3,
new_vectors: 9,
..Default::default()
};
assert_eq!(impossible.reused_vectors(), 0);
}
/// The per-member progress line is the ONLY evidence an operator has during a
/// multi-hour CPU embed, so its content is pinned: member name, position in
/// the pass, chunk count, new vectors, and BOTH elapsed clocks. It also has to
/// state that the member is COMMITTED — the whole reshape is that the line is
/// printed after the manifest+snapshot append, not before it.
#[test]
fn per_member_progress_names_the_member_its_chunks_and_its_elapsed() {
let line = EmbedReport::member_line("nordisk", "holger", 3, 19, 1842, 1105, 12.44, 41.9);
assert!(line.contains("nordisk/holger"), "{line}");
assert!(line.contains("[3/19]"), "{line}");
assert!(line.contains("1842 chunk(s)"), "{line}");
assert!(line.contains("1105 new vector(s)"), "{line}");
assert!(line.contains("737 reused"), "{line}");
assert!(line.contains("12.4s"), "per-member elapsed missing: {line}");
assert!(line.contains("41.9s"), "cumulative elapsed missing: {line}");
assert!(
line.contains("COMMITTED"),
"the line must claim durability, not progress: {line}"
);
}
/// A skipped pass is never a green pass. Both loud-skip branches (no embedder
/// compiled, embedder load failed) must leave `skipped = Some(why)` so the CLI
/// can turn it into a non-zero exit instead of printing "✓ embedded 0 members"
/// — the silently-hollow-Mímir failure this module keeps re-litigating.
#[test]
fn a_skipped_embed_carries_its_reason_and_zero_members() {
// The REAL constructor both loud-skip branches go through.
let skipped = EmbedReport::skipped_pass("nordisk", 19, "embedder load failed: no model");
assert_eq!(skipped.workspace, "nordisk");
assert_eq!(
skipped.members_present, 19,
"the members it DIDN'T embed must still be reported — 'nothing to do' and \
'19 members left hollow' are different facts"
);
assert_eq!(skipped.members_embedded, 0);
assert_eq!(skipped.members_failed, 0);
assert_eq!(skipped.occurrences, 0);
assert_eq!(skipped.new_vectors, 0);
assert_eq!(
skipped.skipped.as_deref(),
Some("embedder load failed: no model"),
"the reason must survive verbatim — the CLI turns it into the exit-code message"
);
// A pass that really ran must NOT look skipped.
let ran = EmbedReport {
members_present: 19,
members_embedded: 19,
occurrences: 40_000,
new_vectors: 40_000,
..Default::default()
};
assert!(
ran.skipped.is_none(),
"a completed pass must not carry a skip reason"
);
// The binary-level answer must agree with which `embed_member_sources`
// body is actually linked: under an embedder build there is no reason to
// refuse, and this test runs under exactly that build.
#[cfg(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
))]
assert_eq!(no_embedder_reason(), None);
#[cfg(not(all(
feature = "vector",
any(feature = "embed-tract", feature = "embed-ort")
)))]
assert!(no_embedder_reason().is_some());
}
}