audit: close 8 correctness/robustness gaps across api, registry, and tooling

src/api.rs

@@ -2652,6 +2652,11 @@ mod tests {
             message.contains("group restore exceeded its budget with 2 batch(es) durable"),
             "{message}"
         );
+        // The same machine-readable claim the batch loop's own timeout
+        // makes: an importer resuming on `durable_batches` must not
+        // have to parse it out of the message on this arm alone.
+        assert_eq!(body["integrity"], "durable_prefix", "{body}");
+        assert_eq!(body["durable_batches"], 2, "{body}");
 
         let _ = std::fs::remove_dir_all(&dir);
     }

src/api/import.rs

@@ -413,15 +413,27 @@ pub(super) fn restore_refusal(
     // partial group write of its own (`restore_groups` validates the
     // whole set before applying any of it).
     match &refusal {
-        RestoreGroupsError::Timeout { .. } => error(
-            code,
-            format!(
-                "group restore exceeded its budget with {batches_landed} batch(es) durable \
-                 (TAGURU_REQUEST_TIMEOUT_SECS tunes this); {}",
-                refusal.text()
-            ),
-            started_at,
-        ),
+        RestoreGroupsError::Timeout { .. } => {
+            // Same machine-readable claim `import_budget_refusal` makes
+            // for the batch loop: the fields speak for the BATCHES (all
+            // durable), which is what a resuming importer keys on — the
+            // group phase itself stays a re-POST, as the note says.
+            let (integrity, durable_batches) = stream_integrity(batches_landed, false);
+            validation_error(
+                code,
+                format!(
+                    "group restore exceeded its budget with {batches_landed} batch(es) durable \
+                     (TAGURU_REQUEST_TIMEOUT_SECS tunes this); {}",
+                    refusal.text()
+                ),
+                RefusalDetail {
+                    integrity: Some(integrity),
+                    durable_batches,
+                    ..Default::default()
+                },
+                started_at,
+            )
+        }
         _ => {
             let (integrity, durable_batches) = stream_integrity(batches_landed, false);
             validation_error(

src/api/promote.rs

@@ -32,7 +32,7 @@ use super::import::{
 };
 use super::{
     AppJson, AppPath, AppQuery, ErrorCode, Issue, RefusalDetail, access_error, deadline_exceeded,
-    error, key_name, ok_with_issues_total, validation_error,
+    error, key_name, ok_with_issues_total, overlong, validation_error,
 };
 
 #[derive(Debug, Deserialize)]
@@ -107,6 +107,9 @@ pub async fn promote_sources(
             started_at,
         );
     }
+    if let Some(refusal) = overlong("sources", request.sources.len(), started_at) {
+        return refusal;
+    }
     if request.into == name {
         return error(
             ErrorCode::InvalidArgument,

src/benchmark/compare.rs

@@ -517,24 +517,32 @@ fn load_results(dir: &Path, manifest: &super::BenchManifest) -> Result<LoadedRes
                         Some("start") => {
                             starts.entry(document_id).or_insert(ts);
                         }
+                        // The mirror image: a re-processed document
+                        // logs a fresh `end`, and the LAST one is the
+                        // state the cell finished in — keeping the
+                        // first would freeze a superseded outcome
+                        // while the per-line counters move on.
                         Some("end") => {
-                            ends.entry(document_id).or_insert_with(|| DocumentEndRaw {
-                                ts,
-                                outcome: value
-                                    .get("outcome")
-                                    .and_then(Value::as_str)
-                                    .map(str::to_string),
-                                associations: value.get("associations").and_then(Value::as_u64),
-                                concepts: value.get("concepts").and_then(Value::as_u64),
-                                labels: value.get("labels").and_then(Value::as_u64),
-                                questions: value.get("questions").and_then(Value::as_u64),
-                                duplicates: value.get("duplicates").and_then(Value::as_u64),
-                                dropped: value.get("dropped").and_then(Value::as_u64),
-                                batch_path: value
-                                    .get("batch_path")
-                                    .and_then(Value::as_str)
-                                    .map(str::to_string),
-                            });
+                            ends.insert(
+                                document_id,
+                                DocumentEndRaw {
+                                    ts,
+                                    outcome: value
+                                        .get("outcome")
+                                        .and_then(Value::as_str)
+                                        .map(str::to_string),
+                                    associations: value.get("associations").and_then(Value::as_u64),
+                                    concepts: value.get("concepts").and_then(Value::as_u64),
+                                    labels: value.get("labels").and_then(Value::as_u64),
+                                    questions: value.get("questions").and_then(Value::as_u64),
+                                    duplicates: value.get("duplicates").and_then(Value::as_u64),
+                                    dropped: value.get("dropped").and_then(Value::as_u64),
+                                    batch_path: value
+                                        .get("batch_path")
+                                        .and_then(Value::as_str)
+                                        .map(str::to_string),
+                                },
+                            );
                         }
                         _ => {}
                     }
@@ -852,7 +860,11 @@ fn attempt_distribution_metrics(attempts: &[&AttemptRow]) -> MetricsMap {
 
 fn wall_seconds(doc: &DocRow) -> Option<f64> {
     match (doc.start_ts, doc.end_ts) {
-        (Some(start), Some(end)) => Some(end - start),
+        // An end stamped before its start (a clock stepped mid-run, a
+        // resumed log's mismatched records) is a broken sample, not a
+        // negative duration — dropped like the degenerate inputs the
+        // per-attempt metrics below refuse.
+        (Some(start), Some(end)) if end >= start => Some(end - start),
         _ => None,
     }
 }

src/benchmark/compare/tests.rs

@@ -439,6 +439,40 @@ fn wall_seconds_needs_both_start_and_end() {
     assert_eq!(wall_seconds(&complete), Some(15.0));
 }
 
+#[test]
+fn wall_seconds_drops_an_end_stamped_before_its_start() {
+    let backwards = DocRow {
+        cell_id: "m.run01".into(),
+        model_id: "m".into(),
+        run_index: 1,
+        document_id: "doc".into(),
+        start_ts: Some(10.0),
+        end_ts: Some(9.0),
+        outcome: None,
+        associations: None,
+        concepts: None,
+        labels: None,
+        questions: None,
+        duplicates: None,
+        dropped: None,
+        elapsed_seconds_sum: 0.0,
+        input_tokens_sum: None,
+        batch: None,
+    };
+    assert_eq!(
+        wall_seconds(&backwards),
+        None,
+        "a clock-stepped sample must not sink the distribution's min"
+    );
+    // A zero-length span is a real (if instant) measurement, not a
+    // broken one.
+    let instant = DocRow {
+        end_ts: Some(10.0),
+        ..backwards
+    };
+    assert_eq!(wall_seconds(&instant), Some(0.0));
+}
+
 #[test]
 fn document_outcome_rates_counts_interrupted_in_the_denominator_only() {
     fn doc(outcome: Option<&str>) -> DocRow {
@@ -1218,6 +1252,37 @@ fn compute_measurements_over_a_synthetic_results_directory() {
     let _ = fs::remove_dir_all(&dir);
 }
 
+#[test]
+fn a_reprocessed_documents_second_end_record_supersedes_the_first() {
+    let dir = synthetic_results_dir("duplicate-end");
+    // A resumed cell that re-processed `brewery` logs a fresh `end`;
+    // the cell finished in THAT state, so the later record must win —
+    // mirroring `start`'s keep-the-earliest, not repeating it.
+    let runs_path = dir.join("runs/m.run01.jsonl");
+    let mut runs = fs::read_to_string(&runs_path).unwrap();
+    runs.push_str(
+        &serde_json::json!({
+            "kind": "document", "ts": 120.0, "cell_id": "m.run01",
+            "document_id": "brewery", "source": "corpus/brewery.md",
+            "document_sha256": "sha-brewery", "phase": "end", "outcome": "written",
+            "associations": 5, "concepts": 1, "labels": 0, "questions": 0,
+            "duplicates": 0, "dropped": 0, "batch_path": "cells/m/run01/brewery.jsonl",
+        })
+        .to_string(),
+    );
+    runs.push('\n');
+    fs::write(&runs_path, runs).unwrap();
+
+    let measurements = compute_measurements(&dir).expect("computes");
+    let brewery_run01 = &measurements.documents["m"]["brewery"]["run01"];
+    let MetricValue::Count(associations) = &brewery_run01["extraction.associations"] else {
+        panic!()
+    };
+    assert_eq!(associations.value(), Some(5.0));
+
+    let _ = fs::remove_dir_all(&dir);
+}
+
 #[test]
 fn stability_metrics_with_a_single_run_are_the_defined_zero_shape() {
     // synthetic_results_dir has exactly one run (m.run01) — every

src/embedding.rs

@@ -1439,6 +1439,17 @@ impl PassageAnnIndex {
     /// index only ever costs recall, never correctness.
     fn build(store: &PassageVectorStore, deadline: Deadline) -> Self {
         let n = store.len();
+        // `centroid_count(0)` is still 1, so without this gate the
+        // seed row below would slice an empty `data` and panic —
+        // today's only caller sits behind the ANN threshold and can't
+        // pass 0 rows, but this function's contract shouldn't lean on
+        // that.
+        if n == 0 {
+            return Self {
+                centroids: Vec::new(),
+                lists: Vec::new(),
+            };
+        }
         let dim = store.dim.max(1);
         let target = Self::centroid_count(n);
         let row = |i: usize| -> &[f32] { &store.data[i * dim..i * dim + dim] };
@@ -2438,6 +2449,14 @@ mod tests {
         assert_eq!(PassageAnnIndex::centroid_count(PASSAGE_ANN_THRESHOLD), 100);
     }
 
+    #[test]
+    fn passage_ann_index_build_on_an_empty_store_is_empty_not_a_panic() {
+        let store = synthetic_passage_store(0, 6);
+        let index = PassageAnnIndex::build(&store, Deadline::unbounded());
+        assert!(index.centroids.is_empty());
+        assert!(index.lists.is_empty());
+    }
+
     #[test]
     fn passage_ann_index_build_partitions_every_row_into_exactly_one_list() {
         let rows = 500;

src/evalset.rs

@@ -433,7 +433,7 @@ pub(crate) fn load_eval_file(path: &Path, mode: Extensions) -> Result<LoadedEval
 
         let wire: WireCase = serde_json::from_value(value)
             .map_err(|error| format!("{label}: line {number}: not a valid eval case: {error}"))?;
-        if wire.case_id.is_empty() {
+        if wire.case_id.trim().is_empty() {
             return Err(format!("{label}: line {number}: case_id must not be empty"));
         }
         if wire.query.trim().is_empty() {
@@ -816,6 +816,17 @@ mod tests {
         let _ = fs::remove_file(&path);
     }
 
+    #[test]
+    fn a_whitespace_only_case_id_is_refused_like_an_empty_one() {
+        let path = write_temp(
+            "blank-case-id",
+            &format!("{HEADER}\n{{\"case_id\":\"  \",\"query\":\"q\"}}\n"),
+        );
+        let error = load_eval_file(&path, Extensions::Interpret).unwrap_err();
+        assert!(error.contains("case_id must not be empty"), "{error}");
+        let _ = fs::remove_file(&path);
+    }
+
     #[test]
     fn an_empty_query_is_refused() {
         let path = write_temp(

src/metrics.rs

@@ -495,6 +495,30 @@ mod tests {
         );
     }
 
+    /// A shipped-only report that no longer outruns the applied seq
+    /// clears the age — the same three-way move `note_replica_lane`
+    /// makes, so a lineage whose shipped seqs regress can never leave
+    /// a stale behind-since stamp on a lane that is in fact caught up.
+    #[test]
+    fn a_shipped_report_at_or_below_applied_clears_the_behind_age() {
+        let metrics = Metrics::default();
+        let key = ("sake".to_string(), "graph");
+        metrics.note_replica_lane("sake", "graph", 5, 5);
+        metrics.note_replica_shipped("sake", "graph", 7);
+        let stamped = metrics.replica_lag.lock()[&key].behind_since_epoch;
+        assert_ne!(stamped, 0, "a real gap starts the age");
+        metrics.note_replica_shipped("sake", "graph", 5);
+        let cleared = metrics.replica_lag.lock()[&key].behind_since_epoch;
+        assert_eq!(cleared, 0, "no gap, no age");
+        // Strictly below, not just equal — the caught-up arm is `>=`,
+        // and a regressed shipped seq (a successor lineage's lower
+        // watermark) must clear the age the same way.
+        metrics.note_replica_shipped("sake", "graph", 7);
+        metrics.note_replica_shipped("sake", "graph", 4);
+        let cleared_below = metrics.replica_lag.lock()[&key].behind_since_epoch;
+        assert_eq!(cleared_below, 0, "a lower shipped seq also clears the age");
+    }
+
     /// The in-flight counter: a ceiling refuses at capacity, zero means
     /// count-only, release always returns the slot — and both series
     /// render on /metrics.

src/metrics/record.rs

@@ -464,9 +464,16 @@ impl Metrics {
         let mut lag = self.replica_lag.lock();
         let entry = lag.entry((context.to_string(), lane)).or_default();
         entry.shipped_seq = shipped_seq;
-        if entry.applied_seq < shipped_seq && entry.behind_since_epoch == 0 {
-            entry.behind_since_epoch = Self::unix_now();
-        }
+        // Same three-way move as `note_replica_lane`: today shipped
+        // seqs only grow within a lineage, so the caught-up arm can't
+        // fire here — but the reset must not silently depend on that.
+        entry.behind_since_epoch = if entry.applied_seq >= shipped_seq {
+            0
+        } else if entry.behind_since_epoch == 0 {
+            Self::unix_now()
+        } else {
+            entry.behind_since_epoch
+        };
     }
 
     /// Drops a vanished context's replica lag rows (both lanes).