Skip to main content

rivet/pipeline/
validate_cmd.rs

1//! **Layer: Coordinator** (config → destination → verification → render)
2//!
3//! `rivet validate` — re-run the manifest-aware `--validate` pass against
4//! an existing destination, without performing an extraction.
5//!
6//! ADR-0013 amendment: this is **not** a new trust noun.  It is a standalone
7//! driver for the same M5/M6 verification surface that `rivet run --validate`
8//! already performs at end-of-run (see [`crate::pipeline::validate_manifest`]).
9//! The verdict shape is identical; the only difference is no source query,
10//! no extraction, no state writes.
11//!
12//! Use cases:
13//! - "Is the output at this prefix still complete?" — Airflow / CI poller
14//!   between runs.
15//! - "Did someone delete a part by mistake?" — operator triage on a
16//!   suspected-broken dataset.
17//! - "Does this legacy prefix have a manifest yet?" — fast check for M6.
18//! - "Was yesterday's run complete?" — `--date YYYY-MM-DD` or `--run-id`
19//!   re-targets a prior day's prefix without re-running the export
20//!   (v0.7.2 historical-validation flags).
21//!
22//! Out of scope:
23//! - Source-side reconciliation (`COUNT(*)`).  That's `--reconcile` /
24//!   `rivet reconcile`, which already exists.
25//! - Per-byte re-fingerprint of every part (`--validate --deep`, future).
26//!
27//! Exit code: `0` if `passed` (or the legacy-run case where the verifier
28//! cannot certify but no failures were seen); non-zero on any explicit
29//! failure (`PartMissing`, `PartSizeMismatch`, `SuccessMarkerStale`, …).
30
31use std::path::Path;
32
33use chrono::NaiveDate;
34
35use crate::config::Config;
36use crate::destination::placeholder::PlaceholderContext;
37use crate::error::Result;
38use crate::pipeline::ManifestVerification;
39use crate::pipeline::validate_manifest::{ValidateDepth, verify_at_destination};
40
41/// Output format mirroring the `rivet reconcile` / `rivet repair` pattern.
42pub enum ValidateOutputFormat {
43    /// Human-readable summary printed to stdout.
44    Pretty,
45    /// JSON to the given path or stdout if `None`.
46    Json(Option<String>),
47}
48
49/// Re-targeting overrides for `rivet validate`.
50///
51/// Default (`ValidateTarget::default()`) reproduces the v0.7.1 behaviour:
52/// resolve `{date}` against today's UTC date, with no `{run_id}`
53/// substitution, and use the config's destination prefix/path unchanged.
54#[derive(Debug, Default, Clone)]
55pub struct ValidateTarget {
56    /// `--date YYYY-MM-DD` — override the date used for `{date}`.
57    pub date: Option<NaiveDate>,
58    /// `--run-id RID` — substitute `{run_id}` in the destination template.
59    pub run_id: Option<String>,
60    /// `--prefix STRING` — bypass placeholder resolution entirely and
61    /// verify exactly this prefix.  Replaces both `prefix` and `path`.
62    pub prefix_override: Option<String>,
63    /// `--depth light|sample|full` — the graded verify layer (see
64    /// [`ValidateDepth`]).  Defaults (`ValidateDepth::default()` →
65    /// [`ValidateDepth::Full`]) to the pre-graded behaviour: all five
66    /// sections **plus** the Form B value re-read, so existing callers
67    /// constructing `ValidateTarget::default()` are unchanged.
68    pub depth: ValidateDepth,
69}
70
71impl ValidateTarget {
72    fn placeholder_context(&self, export_name: &str) -> PlaceholderContext {
73        let mut ctx = match self.date {
74            Some(d) => PlaceholderContext::for_date(d, export_name),
75            None => PlaceholderContext::for_today(export_name),
76        };
77        if let Some(rid) = &self.run_id {
78            ctx = ctx.with_run_id(rid.clone());
79        }
80        ctx
81    }
82}
83
84/// Driver for `rivet validate <export>` (or every export when
85/// `export_name` is `None`).
86///
87/// Returns `Err` on the first explicit verification failure across the
88/// requested exports so an Airflow / CI step can branch on the exit code.
89/// Per-export verdicts are still printed to stdout / written to JSON for
90/// every export, including subsequent ones — the bail at the end is the
91/// last action.
92pub fn run_validate_command(
93    config_path: &str,
94    export_name: Option<&str>,
95    format: ValidateOutputFormat,
96    target: ValidateTarget,
97) -> Result<()> {
98    let config = Config::load_with_params(config_path, None)?;
99
100    let exports: Vec<&crate::config::ExportConfig> = match export_name {
101        Some(name) => match config.exports.iter().find(|e| e.name == name) {
102            Some(e) => vec![e],
103            None => anyhow::bail!("export '{}' not found in config", name),
104        },
105        None => config.exports.iter().collect(),
106    };
107
108    if exports.is_empty() {
109        anyhow::bail!("no exports defined in config — nothing to validate");
110    }
111
112    // `--prefix` only makes sense for a single export; with multiple
113    // exports it would silently re-point all of them at the same physical
114    // bytes.  Catch this at the boundary so we never head-check the wrong
115    // dataset under operator triage pressure.
116    if target.prefix_override.is_some() && exports.len() > 1 {
117        anyhow::bail!(
118            "--prefix requires --export <name>: cannot apply one override to {} exports",
119            exports.len()
120        );
121    }
122
123    let mut all_results: Vec<ExportVerdict> = Vec::with_capacity(exports.len());
124    let mut hard_failures: Vec<String> = Vec::new();
125
126    for export in &exports {
127        // Apply the operator-supplied re-targeting if any, else fall back
128        // to today's UTC date (same shape `rivet run` resolves at write
129        // time).
130        let ctx = target.placeholder_context(&export.name);
131        let mut expanded_dest =
132            crate::destination::placeholder::expand_destination(export.destination.clone(), &ctx);
133        if let Some(p) = &target.prefix_override {
134            // Bypass placeholder resolution: trust the operator's literal
135            // prefix.  Replace both `path` (local) and `prefix` (cloud)
136            // so whichever the backend reads picks up the override.
137            expanded_dest.path = Some(p.clone());
138            expanded_dest.prefix = Some(p.clone());
139        }
140        // A CDC export's output is one table prefix per captured table: a
141        // `tables:` stream fans each out under `<base>/<table>/` (via
142        // `cdc_job::dest_for_table`), a single-table export uses its prefix
143        // directly. Either way the change parts live at the table prefix with an
144        // optional nested `snapshot/` dataset, and a fanned-out base holds no
145        // manifest of its own — so descend per table (using the writer's builder)
146        // rather than verifying the base, which would read back "legacy_run" and
147        // fail the `__pos` check on a missing part.
148        let multiplex = if export.mode == crate::config::ExportMode::Cdc {
149            export.tables.as_deref().filter(|t| !t.is_empty())
150        } else {
151            None
152        };
153        let has_snapshot = export.mode == crate::config::ExportMode::Cdc
154            && export.cdc.as_ref().and_then(|c| c.initial)
155                == Some(crate::config::CdcInitialMode::Snapshot);
156        match multiplex {
157            Some(tables) => {
158                for table in tables {
159                    verify_cdc_table(
160                        crate::pipeline::cdc_job::dest_for_table(&expanded_dest, table),
161                        format!("{}/{}", export.name, table),
162                        export,
163                        &target,
164                        has_snapshot,
165                        &mut all_results,
166                        &mut hard_failures,
167                    );
168                }
169            }
170            None if export.mode == crate::config::ExportMode::Cdc => {
171                // Single-table CDC: the export prefix IS the one table prefix.
172                verify_cdc_table(
173                    expanded_dest,
174                    export.name.clone(),
175                    export,
176                    &target,
177                    has_snapshot,
178                    &mut all_results,
179                    &mut hard_failures,
180                );
181            }
182            None => {
183                // Batch export: verify the prefix directly — no CDC `__pos`
184                // continuity, no nested snapshot dataset.
185                verify_one_prefix(
186                    expanded_dest,
187                    export.name.clone(),
188                    export,
189                    &target,
190                    false,
191                    false,
192                    &mut all_results,
193                    &mut hard_failures,
194                );
195            }
196        }
197    }
198
199    match format {
200        ValidateOutputFormat::Pretty => render_pretty(&all_results, &hard_failures),
201        ValidateOutputFormat::Json(out_path) => {
202            render_json(&all_results, &hard_failures, out_path)?
203        }
204    }
205
206    // Exit-code policy: the standalone driver fails when an export's
207    // verdict surfaced an explicit failure it could not pass over
208    // (`verdict_fails_exit`) — an M5 verification failure on a found
209    // manifest (missing part, size mismatch, stale _SUCCESS,
210    // self-inconsistent manifest) or a manifest that could not even be
211    // read (`ManifestReadError`: `manifest_found` is false, but the
212    // verifier has a concrete reason to refuse, not a legacy prefix).
213    // Surplus untracked objects (`UntrackedObject`) are surfaced in
214    // `failures` for operator audit but do NOT flip `passed`, because
215    // their cleanup is M9's job (resume), not validate's.  An operator
216    // who wants strict "no surplus allowed" can grep the JSON report for
217    // `kind: untracked_object` themselves; a future
218    // `rivet validate --strict` flag may surface that exit-code mode if
219    // demand appears (out of scope for this PR).
220    //
221    // Legacy runs (M6) keep exit 0: `passed: false` with no failures
222    // means "verifier cannot certify", not "verifier found a problem".
223    let failed_verdicts = all_results
224        .iter()
225        .filter(|r| verdict_fails_exit(&r.verification))
226        .count();
227    if failed_verdicts > 0 {
228        // A verified-and-wrong verdict (missing part, size mismatch, stale
229        // _SUCCESS, self-inconsistent manifest) is the data-integrity class
230        // (exit 3) — typed so a scheduler stops rather than blindly retries.
231        // `hard_failures` (couldn't open / read the destination) are operational
232        // "could not verify", not "verified wrong", so they fold into the count
233        // but the class is driven by the real verdict failure.
234        return Err(crate::error::DataIntegrityError::new(format!(
235            "rivet validate: {} export(s) failed verification",
236            hard_failures.len() + failed_verdicts
237        ))
238        .into());
239    }
240    if !hard_failures.is_empty() {
241        // Could-not-verify only (no verified-wrong verdict): operational, generic.
242        anyhow::bail!(
243            "rivet validate: {} export(s) failed verification",
244            hard_failures.len()
245        );
246    }
247    Ok(())
248}
249
250/// Exit-code predicate for one export's verdict: non-zero iff the verifier
251/// surfaced an explicit failure (`has_failures` — "a reason an orchestrator
252/// should refuse the run") on a verdict that did not pass.  Both documented
253/// exit-0 cases survive: legacy runs (M6 — `passed: false` with no failures
254/// is "cannot certify", not "found a problem") and advisory-only verdicts
255/// (`UntrackedObject` never flips `passed`).
256fn verdict_fails_exit(v: &ManifestVerification) -> bool {
257    !v.passed && v.has_failures()
258}
259
260/// Per-export verdict plus the resolved physical prefix the verifier
261/// looked at — surfaced in both pretty and JSON output so an operator can
262/// confirm at a glance which bytes were checked.
263struct ExportVerdict {
264    name: String,
265    resolved_prefix: String,
266    verification: ManifestVerification,
267}
268
269/// Render the destination's resolved prefix for human/JSON output.
270///
271/// Cloud backends carry the data location in `prefix`; the local backend
272/// uses `path`.  Falling back to `<unresolved>` should never fire under
273/// normal config (clap + Config::load enforce one of the two) but keeps
274/// `validate` from panicking if a future config shape lands here.
275fn resolved_prefix_for_display(dest: &crate::config::DestinationConfig) -> String {
276    dest.prefix
277        .clone()
278        .or_else(|| dest.path.clone())
279        .unwrap_or_else(|| "<unresolved>".into())
280}
281
282/// Verify one CDC table's output at `table_dest`: its initial-snapshot dataset
283/// (when `has_snapshot`, a nested `snapshot/` prefix with its OWN manifest) plus
284/// its change parts. Shared by each table of a `tables:` stream AND a
285/// single-table CDC export — both write the same `<prefix>/{snapshot/, cdc-*}`
286/// shape, so both certify the snapshot and drop its (separately-verified) files
287/// from the change prefix's untracked-surplus advisory.
288#[allow(clippy::too_many_arguments)]
289fn verify_cdc_table(
290    table_dest: crate::config::DestinationConfig,
291    display_name: String,
292    export: &crate::config::ExportConfig,
293    target: &ValidateTarget,
294    has_snapshot: bool,
295    all_results: &mut Vec<ExportVerdict>,
296    hard_failures: &mut Vec<String>,
297) {
298    // The snapshot is a batch dataset with its own manifest — verify it, but with
299    // no `__pos` continuity check (snapshot rows carry no log position).
300    if has_snapshot {
301        let snap = crate::pipeline::cdc_job::dest_for_table(&table_dest, "snapshot");
302        verify_one_prefix(
303            snap,
304            format!("{display_name}/snapshot"),
305            export,
306            target,
307            false,
308            false,
309            all_results,
310            hard_failures,
311        );
312    }
313    verify_one_prefix(
314        table_dest,
315        display_name,
316        export,
317        target,
318        true,
319        has_snapshot,
320        all_results,
321        hard_failures,
322    );
323}
324
325/// Verify ONE resolved prefix (an export's whole destination, or one table's
326/// sub-prefix of a `tables:` CDC stream) and append its verdict / failures.
327/// `display_name` labels the verdict (`export` or `export/table[/snapshot]`);
328/// `run_cdc_pos_check` gates the CDC `__pos`-continuity re-read (off for a
329/// snapshot sub-prefix, whose batch rows carry no log position).
330#[allow(clippy::too_many_arguments)]
331fn verify_one_prefix(
332    expanded_dest: crate::config::DestinationConfig,
333    display_name: String,
334    export: &crate::config::ExportConfig,
335    target: &ValidateTarget,
336    run_cdc_pos_check: bool,
337    drop_snapshot_untracked: bool,
338    all_results: &mut Vec<ExportVerdict>,
339    hard_failures: &mut Vec<String>,
340) {
341    let resolved_prefix = resolved_prefix_for_display(&expanded_dest);
342    let dest = match crate::destination::create_destination(&expanded_dest) {
343        Ok(d) => d,
344        Err(e) => {
345            hard_failures.push(format!(
346                "export '{}' (prefix: {}): could not open destination: {:#}",
347                display_name, resolved_prefix, e
348            ));
349            return;
350        }
351    };
352    // Streaming destinations have no prefix to verify — note and skip.
353    if dest.capabilities().commit_protocol == crate::destination::WriteCommitProtocol::Streaming {
354        log::info!(
355            "export '{}': streaming destination, skipping (nothing to verify)",
356            display_name
357        );
358        return;
359    }
360    match verify_at_destination(&*dest, "", target.depth) {
361        Ok(mut v) => {
362            // Apply this export's `verify` policy: `content` fails the
363            // verdict when any part is only size-verified (review D).
364            v.enforce_content_policy(export.verify.requires_content());
365            // A `tables:` CDC table prefix holds its initial snapshot under a
366            // nested `snapshot/` dataset with its OWN manifest (verified as a
367            // separate prefix above). The listing here therefore sees those files
368            // as "untracked" surplus — but they are a known, separately-certified
369            // dataset, not orphans. Drop that advisory so the operator isn't told
370            // real data is stray. (Non-fatal already, so `passed` is unaffected.)
371            if drop_snapshot_untracked {
372                v.failures.retain(|f| {
373                    !matches!(
374                        f,
375                        crate::pipeline::validate_manifest::Failure::UntrackedObject { key, .. }
376                            if key.starts_with("snapshot/")
377                    )
378                });
379            }
380            // Finding #20: when the operator pinned a literal `--prefix`,
381            // they asserted a real dataset lives here. An absent manifest is
382            // then NOT the benign M6 legacy-run case (exit 0) — it almost
383            // always means the prefix was never written (a misconfigured CI
384            // gate `rivet validate && deploy` sailing past nothing). Escalate
385            // that exact shape (no manifest, no other failure) to a fatal
386            // `ManifestRequiredButAbsent` so the exit gate refuses it loudly
387            // instead of silently passing. No-op for every other shape (a
388            // real manifest, or an absent one already carrying a read error).
389            if target.prefix_override.is_some() {
390                v.require_manifest_present(&resolved_prefix);
391            }
392            // Capture the verdict before `v` is moved into the result: the
393            // deeper Form B checksum re-read below must run *only* on a
394            // manifest that was found and passed the standard checks.
395            let manifest_verified = v.manifest_found && v.passed;
396            all_results.push(ExportVerdict {
397                name: display_name.clone(),
398                resolved_prefix,
399                verification: v,
400            });
401            // CDC-specific: re-read the parts and confirm `__pos` stayed in
402            // source-log order (no reorder / no part-boundary overlap). The
403            // manifest check above already covered per-part MD5 / size / _SUCCESS.
404            // Full-depth only — like Form B below it downloads every part, so a
405            // light/sample run skips it (keeps the depth contract consistent).
406            if target.depth.runs_part_download()
407                && run_cdc_pos_check
408                && export.format == crate::config::FormatType::Parquet
409            {
410                match crate::source::cdc::validate::check_positions(&*dest, "") {
411                    Ok(pc) if pc.is_ok() => log::info!(
412                        "export '{}': cdc __pos continuity OK — {} changes across {} parts, range {:?}..{:?}",
413                        display_name,
414                        pc.rows,
415                        pc.parts,
416                        pc.first,
417                        pc.last
418                    ),
419                    Ok(pc) => {
420                        for viol in &pc.violations {
421                            hard_failures
422                                .push(format!("export '{}': cdc __pos: {}", display_name, viol));
423                        }
424                    }
425                    Err(e) => hard_failures.push(format!(
426                        "export '{}': cdc __pos check failed: {:#}",
427                        display_name, e
428                    )),
429                }
430            }
431            // Form B: re-read the parts and verify the per-column value
432            // checksums recorded in the manifest (catches an Arrow→Parquet
433            // encode / post-write fault the in-process Form A cannot see).
434            // Gated on a found+passed manifest: an absent (legacy pass) or
435            // unreadable manifest is already accounted for above, so re-reading
436            // it here would either break the legacy pass or double-count.
437            //
438            // Graded depth: Form B is the **only** part-download step, so it
439            // runs at `--depth full` alone.  `light` and `sample` deliberately
440            // skip it — `sample` is "all structural checks, no part bodies".
441            if target.depth.runs_part_download()
442                && manifest_verified
443                && export.format == crate::config::FormatType::Parquet
444                && let Err(e) =
445                    crate::source::value_checksum::validate_manifest_checksums(&*dest, "")
446            {
447                hard_failures.push(format!(
448                    "export '{}': value checksum: {:#}",
449                    display_name, e
450                ));
451            }
452        }
453        Err(e) => {
454            hard_failures.push(format!(
455                "export '{}' (prefix: {}): verify_at_destination failed: {:#}",
456                display_name, resolved_prefix, e
457            ));
458        }
459    }
460}
461
462fn render_pretty(results: &[ExportVerdict], hard_failures: &[String]) {
463    use std::io::Write;
464    let stdout = std::io::stdout();
465    let mut h = stdout.lock();
466
467    for r in results {
468        let _ = writeln!(h, "── {} ──", r.name);
469        let _ = writeln!(h, "  prefix:    {}", r.resolved_prefix);
470        let v = &r.verification;
471        // Graded verify layer: surface how deep this pass went so a reader
472        // knows whether a PASSED verdict reconciled parts (sample/full) or
473        // only the manifest + _SUCCESS (light).
474        let _ = writeln!(h, "  depth:     {}", v.depth_level);
475        if v.legacy_run {
476            let _ = writeln!(
477                h,
478                "  status:    legacy_run (no manifest at destination — pre-0.7.0 prefix)"
479            );
480            continue;
481        }
482        if !v.manifest_found {
483            let _ = writeln!(h, "  status:    NO MANIFEST");
484            // A read-error verdict lands here (manifest present but
485            // unreadable, or head failed): its `failures` are the
486            // operator's only signal, so print them before bailing out
487            // of this export's section.  Each line carries its stable
488            // `RIVET_VERIFY_*` code in brackets so CI can grep it.
489            for failure in &v.failures {
490                let _ = writeln!(h, "  failure:   [{}] {}", failure.error_code(), failure);
491            }
492            continue;
493        }
494        let _ = writeln!(
495            h,
496            "  status:    {}",
497            if v.passed { "PASSED" } else { "FAILED" }
498        );
499        let _ = writeln!(
500            h,
501            "  parts:     {} verified ({} md5, {} size-only), {} failed",
502            v.parts_verified,
503            v.parts_md5_verified,
504            v.parts_verified.saturating_sub(v.parts_md5_verified),
505            v.parts_failed
506        );
507        let _ = writeln!(
508            h,
509            "  _SUCCESS:  {}",
510            if v.success_marker_consistent {
511                "consistent"
512            } else if v.failures.iter().any(|f| matches!(
513                f,
514                crate::pipeline::ManifestVerificationFailure::SuccessMarkerStale { .. }
515                    | crate::pipeline::ManifestVerificationFailure::SuccessMarkerMalformed { .. }
516                    | crate::pipeline::ManifestVerificationFailure::SuccessMarkerReadError { .. }
517            )) {
518                "INCONSISTENT (see failures)"
519            } else {
520                "absent (no signal)"
521            }
522        );
523        let _ = writeln!(
524            h,
525            "  manifest:  {}",
526            if v.manifest_self_consistent {
527                "self-consistent"
528            } else {
529                "INCONSISTENT (see failures)"
530            }
531        );
532        for failure in &v.failures {
533            // `Failure: Display` is the single source of truth for the message;
534            // same string the run report uses.  L14: advisory (non-fatal)
535            // entries — `UntrackedObject` surplus — are labelled "warning:" not
536            // "failure:".  They never flip `passed` and never change the exit
537            // code (cleanup is `--resume`'s job, M9), so rendering them as
538            // "failure:" beside exit 0 was contradictory.  Fatal failures keep
539            // the "failure:" label.
540            let label = if failure.is_fatal() {
541                "failure:"
542            } else {
543                "warning:"
544            };
545            // Stable `RIVET_VERIFY_*` code in brackets ahead of the human
546            // message so an orchestrator can branch on the code without
547            // parsing the prose.
548            let _ = writeln!(h, "  {}   [{}] {}", label, failure.error_code(), failure);
549        }
550    }
551
552    if !hard_failures.is_empty() {
553        let _ = writeln!(h);
554        let _ = writeln!(h, "── errors ──");
555        for e in hard_failures {
556            let _ = writeln!(h, "  {}", e);
557        }
558    }
559    let _ = h.flush();
560}
561
562/// Serialize one [`ManifestVerificationFailure`] to JSON with its stable
563/// `RIVET_VERIFY_*` code injected next to `kind`.
564///
565/// The derive emits `{ "kind": "...", <variant fields> }`; this adds `"code"`
566/// so a consumer can branch on the code without re-deriving it from `kind`.
567/// Returns the enriched object (or the unmodified serde value if, impossibly,
568/// the failure didn't serialize as a JSON object).
569fn failure_json(f: &crate::pipeline::ManifestVerificationFailure) -> serde_json::Value {
570    let mut value = serde_json::json!(f);
571    if let Some(obj) = value.as_object_mut() {
572        obj.insert(
573            "code".to_string(),
574            serde_json::Value::String(f.error_code().to_string()),
575        );
576    }
577    value
578}
579
580/// Serialize a [`ManifestVerification`] to JSON, replacing the derive's plain
581/// `failures` array with one whose entries each carry their `RIVET_VERIFY_*`
582/// `code`.  All other fields (`depth_level`, `passed`, counts, …) ride the
583/// derive unchanged, so the stable wire contract is preserved and only widened.
584fn verification_json(v: &ManifestVerification) -> serde_json::Value {
585    let mut value = serde_json::json!(v);
586    if let Some(obj) = value.as_object_mut() {
587        let failures: Vec<serde_json::Value> = v.failures.iter().map(failure_json).collect();
588        obj.insert("failures".to_string(), serde_json::Value::Array(failures));
589    }
590    value
591}
592
593fn render_json(
594    results: &[ExportVerdict],
595    hard_failures: &[String],
596    out_path: Option<String>,
597) -> Result<()> {
598    // L14: surface advisory (non-fatal) entries — `UntrackedObject` surplus —
599    // in a dedicated top-level `warnings` array so a consumer can tell at a
600    // glance that "failures means failures".  The per-export
601    // `verification.failures` array is the stable wire contract (consumers
602    // branch on `failures[].kind`), so advisory entries stay there too — this
603    // is an additive lens over the same data, not a relocation.  Each entry
604    // also carries its stable `RIVET_VERIFY_*` `code` next to `kind`.
605    let warnings: Vec<serde_json::Value> = results
606        .iter()
607        .flat_map(|r| {
608            r.verification
609                .failures
610                .iter()
611                .filter(|f| !f.is_fatal())
612                .map(move |f| {
613                    serde_json::json!({
614                        "export_name": r.name,
615                        "warning": failure_json(f),
616                    })
617                })
618        })
619        .collect();
620
621    let payload = serde_json::json!({
622        "exports": results
623            .iter()
624            .map(|r| {
625                serde_json::json!({
626                    "export_name": r.name,
627                    "resolved_prefix": r.resolved_prefix,
628                    "verification": verification_json(&r.verification),
629                })
630            })
631            .collect::<Vec<_>>(),
632        "warnings": warnings,
633        "errors": hard_failures,
634    });
635    let serialized = serde_json::to_string_pretty(&payload)?;
636    match out_path {
637        Some(p) => {
638            std::fs::write(Path::new(&p), &serialized)?;
639            log::info!("rivet validate: wrote JSON report to {}", p);
640        }
641        None => {
642            println!("{}", serialized);
643        }
644    }
645    Ok(())
646}
647
648#[cfg(test)]
649mod tests {
650    use super::*;
651
652    // ── ValidateTarget::placeholder_context ────────────────────────────────
653
654    #[test]
655    fn target_default_uses_today() {
656        let target = ValidateTarget::default();
657        let ctx = target.placeholder_context("orders");
658        assert_eq!(ctx.date, chrono::Utc::now().date_naive());
659        assert_eq!(ctx.export_name, "orders");
660        assert!(ctx.run_id.is_none());
661    }
662
663    #[test]
664    fn target_with_date_overrides_today() {
665        let target = ValidateTarget {
666            date: Some(NaiveDate::from_ymd_opt(2026, 5, 21).unwrap()),
667            ..Default::default()
668        };
669        let ctx = target.placeholder_context("orders");
670        assert_eq!(ctx.date, NaiveDate::from_ymd_opt(2026, 5, 21).unwrap());
671        assert!(ctx.run_id.is_none());
672    }
673
674    #[test]
675    fn target_composes_date_and_run_id() {
676        // Regression for the "run yesterday, validate today" scenario:
677        // operator passes both --date and --run-id; the resolver must see
678        // both.
679        let target = ValidateTarget {
680            date: Some(NaiveDate::from_ymd_opt(2026, 5, 21).unwrap()),
681            run_id: Some("r-abc123".into()),
682            prefix_override: None,
683            ..Default::default()
684        };
685        let ctx = target.placeholder_context("orders");
686        assert_eq!(ctx.date, NaiveDate::from_ymd_opt(2026, 5, 21).unwrap());
687        assert_eq!(ctx.run_id.as_deref(), Some("r-abc123"));
688    }
689
690    // ── resolved_prefix_for_display ────────────────────────────────────────
691
692    #[test]
693    fn resolved_prefix_prefers_cloud_prefix_over_path() {
694        let dest = crate::config::DestinationConfig {
695            destination_type: crate::config::DestinationType::S3,
696            prefix: Some("exports/2026-05-21/orders/".into()),
697            path: Some("/scratch".into()),
698            ..Default::default()
699        };
700        assert_eq!(
701            resolved_prefix_for_display(&dest),
702            "exports/2026-05-21/orders/",
703        );
704    }
705
706    #[test]
707    fn resolved_prefix_falls_back_to_path_when_prefix_missing() {
708        let dest = crate::config::DestinationConfig {
709            destination_type: crate::config::DestinationType::Local,
710            prefix: None,
711            path: Some("/data/out".into()),
712            ..Default::default()
713        };
714        assert_eq!(resolved_prefix_for_display(&dest), "/data/out");
715    }
716
717    // ── verdict_fails_exit (exit-code policy) ──────────────────────────────
718
719    use crate::pipeline::ManifestVerificationFailure as VFailure;
720
721    /// Verdict shape `verify_at_destination` returns when `manifest.json`
722    /// exists but cannot be read: not legacy, not passed, one explicit
723    /// `ManifestReadError`.
724    fn read_error_verdict() -> ManifestVerification {
725        ManifestVerification {
726            legacy_run: false,
727            failures: vec![VFailure::ManifestReadError {
728                detail: "permission denied".into(),
729            }],
730            ..ManifestVerification::legacy()
731        }
732    }
733
734    #[test]
735    fn exit_gate_counts_manifest_read_error_as_failure() {
736        assert!(verdict_fails_exit(&read_error_verdict()));
737    }
738
739    #[test]
740    fn exit_gate_keeps_legacy_run_at_zero() {
741        // M6: no manifest, no failures — "cannot certify" is not "found a
742        // problem".
743        assert!(!verdict_fails_exit(&ManifestVerification::legacy()));
744    }
745
746    #[test]
747    fn exit_gate_keeps_advisory_untracked_at_zero() {
748        let v = ManifestVerification {
749            manifest_found: true,
750            legacy_run: false,
751            passed: true,
752            parts_verified: 1,
753            failures: vec![VFailure::UntrackedObject {
754                key: "stray.parquet".into(),
755                size_bytes: 9,
756            }],
757            ..ManifestVerification::legacy()
758        };
759        assert!(!verdict_fails_exit(&v));
760    }
761
762    #[test]
763    fn exit_gate_counts_fatal_failure_on_found_manifest() {
764        let v = ManifestVerification {
765            manifest_found: true,
766            legacy_run: false,
767            failures: vec![VFailure::PartMissing {
768                part_id: 1,
769                path: "part-000001.parquet".into(),
770            }],
771            ..ManifestVerification::legacy()
772        };
773        assert!(verdict_fails_exit(&v));
774    }
775
776    // ── run_validate_command end-to-end (local destination; the source URL
777    //     is never dialed — see tests/validate_historical.rs) ──────────────
778
779    use crate::manifest::{
780        MANIFEST_VERSION, ManifestDestination, ManifestPart, ManifestSource, ManifestStatus,
781        PartStatus, RunManifest,
782    };
783
784    fn success_manifest(parts: Vec<ManifestPart>) -> RunManifest {
785        let row_count: i64 = parts.iter().map(|p| p.rows).sum();
786        let part_count = parts.len() as u32;
787        RunManifest {
788            mode: "batch".to_string(),
789            manifest_version: MANIFEST_VERSION,
790            run_id: "r-validate-cmd".into(),
791            export_name: "orders".into(),
792            started_at: "2026-06-09T12:00:00Z".into(),
793            finished_at: "2026-06-09T12:01:00Z".into(),
794            status: ManifestStatus::Success,
795            source: ManifestSource {
796                engine: "postgres".into(),
797                schema: Some("public".into()),
798                table: Some("orders".into()),
799                extraction: None,
800            },
801            destination: ManifestDestination {
802                kind: "local".into(),
803                uri: "file:///tmp/out".into(),
804            },
805            format: "parquet".into(),
806            compression: "zstd".into(),
807            schema_fingerprint: "xxh3:0123456789abcdef".into(),
808            row_count,
809            part_count,
810            parts,
811            column_checksums: None,
812            checksum_key_column: None,
813        }
814    }
815
816    /// Land `manifest.json` + `_SUCCESS` at `prefix` via the public writer
817    /// surface — same path the `rivet run` end-of-run writer takes.
818    fn stage_dataset(prefix: &Path, m: &RunManifest) {
819        std::fs::create_dir_all(prefix).unwrap();
820        let dest = crate::destination::create_destination(&crate::config::DestinationConfig {
821            destination_type: crate::config::DestinationType::Local,
822            path: Some(prefix.to_string_lossy().into_owned()),
823            ..Default::default()
824        })
825        .unwrap();
826        crate::pipeline::write_manifest(&*dest, m).unwrap();
827    }
828
829    /// Config with a single export pointing at `prefix`.  Written next to —
830    /// never inside — the prefix, so it can't surface as untracked surplus.
831    fn write_cfg(dir: &Path, prefix: &Path) -> std::path::PathBuf {
832        let cfg = dir.join("rivet.yaml");
833        let yaml = format!(
834            "source:\n  type: postgres\n  url: postgresql://nobody@localhost/nope\nexports:\n  - name: orders\n    query: \"SELECT 1\"\n    mode: full\n    format: parquet\n    destination:\n      type: local\n      path: \"{}\"\n",
835            prefix.to_string_lossy()
836        );
837        std::fs::write(&cfg, yaml).unwrap();
838        cfg
839    }
840
841    /// In-process twin of the live roast test (tests/roast_validate_exit.rs):
842    /// `manifest.json` present but unreadable must exit non-zero.  head()
843    /// (fs::metadata) succeeds, read() (fs::read) hits EACCES — exactly the
844    /// `ManifestReadError` verdict.
845    #[cfg(unix)]
846    #[test]
847    fn unreadable_manifest_fails_the_command() {
848        use std::os::unix::fs::PermissionsExt;
849
850        let dir = tempfile::tempdir().unwrap();
851        let prefix = dir.path().join("out");
852        stage_dataset(&prefix, &success_manifest(Vec::new()));
853        let cfg = write_cfg(dir.path(), &prefix);
854
855        let manifest_path = prefix.join(crate::manifest::MANIFEST_FILENAME);
856        std::fs::set_permissions(&manifest_path, std::fs::Permissions::from_mode(0o000)).unwrap();
857        if std::fs::read(&manifest_path).is_ok() {
858            // euid 0 ignores file modes — the degraded state can't be staged.
859            eprintln!("skipping unreadable_manifest_fails_the_command: running as root");
860            return;
861        }
862
863        let report = dir.path().join("report.json");
864        let err = run_validate_command(
865            cfg.to_str().unwrap(),
866            Some("orders"),
867            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
868            ValidateTarget::default(),
869        )
870        .expect_err("an unreadable manifest is an explicit failure, not exit 0");
871        assert!(
872            format!("{err:#}").contains("1 export(s) failed verification"),
873            "got: {err:#}"
874        );
875
876        // The JSON report (written before the bail) still carries the
877        // verdict so the operator sees why.
878        let json: serde_json::Value =
879            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
880        let verification = &json["exports"][0]["verification"];
881        assert_eq!(verification["manifest_found"], false);
882        assert_eq!(verification["legacy_run"], false);
883        assert_eq!(verification["failures"][0]["kind"], "manifest_read_error");
884    }
885
886    #[test]
887    fn untracked_surplus_alone_keeps_exit_zero() {
888        // The advisory neighbor of the read-error fix: gating on
889        // `has_failures()` alone would flip this verdict to non-zero, but
890        // surplus cleanup is `--resume`'s job (M9), not validate's.
891        let dir = tempfile::tempdir().unwrap();
892        let prefix = dir.path().join("out");
893        stage_dataset(&prefix, &success_manifest(Vec::new()));
894        std::fs::write(prefix.join("rogue.parquet"), b"XX").unwrap();
895        let cfg = write_cfg(dir.path(), &prefix);
896
897        let report = dir.path().join("report.json");
898        run_validate_command(
899            cfg.to_str().unwrap(),
900            Some("orders"),
901            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
902            ValidateTarget::default(),
903        )
904        .expect("advisory untracked surplus must not flip the exit code");
905
906        let json: serde_json::Value =
907            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
908        let verification = &json["exports"][0]["verification"];
909        assert_eq!(verification["passed"], true);
910        // The stable wire contract is preserved: untracked entries still ride
911        // `verification.failures` (consumers branch on `failures[].kind`).
912        assert_eq!(verification["failures"][0]["kind"], "untracked_object");
913
914        // L14: …and the same advisory entry is also surfaced in the top-level
915        // `warnings` array so "failures means failures" — an exit-0 verdict no
916        // longer hides a surplus object under a "failure" label.
917        let warnings = json["warnings"].as_array().expect("warnings array present");
918        assert_eq!(warnings.len(), 1, "the untracked surplus is one warning");
919        assert_eq!(warnings[0]["export_name"], "orders");
920        assert_eq!(warnings[0]["warning"]["kind"], "untracked_object");
921        assert_eq!(warnings[0]["warning"]["key"], "rogue.parquet");
922    }
923
924    #[test]
925    fn json_warnings_array_is_empty_when_no_advisory_failures() {
926        // A clean dataset with no surplus → no warnings.  Guards against the
927        // `warnings` lens accidentally picking up fatal failures.
928        let dir = tempfile::tempdir().unwrap();
929        let prefix = dir.path().join("out");
930        stage_dataset(&prefix, &success_manifest(Vec::new()));
931        let cfg = write_cfg(dir.path(), &prefix);
932
933        let report = dir.path().join("report.json");
934        run_validate_command(
935            cfg.to_str().unwrap(),
936            Some("orders"),
937            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
938            ValidateTarget::default(),
939        )
940        .expect("a clean dataset must pass");
941
942        let json: serde_json::Value =
943            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
944        assert_eq!(
945            json["warnings"]
946                .as_array()
947                .expect("warnings array present")
948                .len(),
949            0,
950            "no surplus → no warnings"
951        );
952    }
953
954    #[test]
955    fn multiplex_cdc_validates_each_table_sub_prefix() {
956        // Regression: a `tables:` CDC stream lands each table under its OWN
957        // sub-prefix (`<base>/<table>/`, via `cdc_job::dest_for_table`). The base
958        // prefix holds no manifest, so validating it alone read back as an empty
959        // "legacy_run" and produced a single base verdict — the operator could
960        // not certify the stream. The command must descend into each table.
961        let dir = tempfile::tempdir().unwrap();
962        let base = dir.path().join("cdc");
963        stage_dataset(&base.join("alpha"), &success_manifest(Vec::new()));
964        stage_dataset(&base.join("beta"), &success_manifest(Vec::new()));
965        let cfg = write_multiplex_cfg(dir.path(), &base);
966
967        let report = dir.path().join("report.json");
968        run_validate_command(
969            cfg.to_str().unwrap(),
970            None,
971            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
972            // Sample depth: exercise the per-table descent + manifest verify
973            // without the full-depth part download (empty datasets carry no
974            // parquet to __pos-check).
975            ValidateTarget {
976                depth: ValidateDepth::Sample,
977                ..Default::default()
978            },
979        )
980        .expect("both table sub-prefixes are complete — the stream must validate");
981
982        let json: serde_json::Value =
983            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
984        let names: Vec<&str> = json["exports"]
985            .as_array()
986            .unwrap()
987            .iter()
988            .map(|e| e["export_name"].as_str().unwrap())
989            .collect();
990        // ONE verdict per table, named `<export>/<table>` — proof the descent
991        // ran (the un-fixed command produced a single "cdc" verdict at the base).
992        assert_eq!(names, vec!["cdc/alpha", "cdc/beta"], "per-table descent");
993        for e in json["exports"].as_array().unwrap() {
994            assert_eq!(e["verification"]["passed"], true, "each table passes");
995        }
996    }
997
998    fn write_multiplex_cfg(dir: &Path, base: &Path) -> std::path::PathBuf {
999        let cfg = dir.join("rivet-cdc.yaml");
1000        let yaml = format!(
1001            "source:\n  type: mysql\n  url: mysql://nobody@localhost/nope\nexports:\n  - name: cdc\n    tables: [alpha, beta]\n    mode: cdc\n    format: parquet\n    cdc:\n      server_id: 1\n    destination:\n      type: local\n      path: \"{}\"\n",
1002            base.to_string_lossy()
1003        );
1004        std::fs::write(&cfg, yaml).unwrap();
1005        cfg
1006    }
1007
1008    #[test]
1009    fn single_table_cdc_certifies_snapshot_and_hides_its_untracked() {
1010        // A single-table CDC export with `initial: snapshot` lands its snapshot as
1011        // a nested `snapshot/` dataset under the change prefix. Validate must
1012        // (a) give the snapshot its OWN verdict (so a missing/broken snapshot is
1013        // caught) and (b) NOT flag the snapshot files as "untracked" at the change
1014        // prefix — they are a separately-verified dataset, not orphans.
1015        let dir = tempfile::tempdir().unwrap();
1016        let base = dir.path().join("cdc");
1017        stage_dataset(&base, &success_manifest(Vec::new()));
1018        stage_dataset(&base.join("snapshot"), &success_manifest(Vec::new()));
1019        let cfg = write_single_cdc_cfg(dir.path(), &base);
1020
1021        let report = dir.path().join("report.json");
1022        run_validate_command(
1023            cfg.to_str().unwrap(),
1024            None,
1025            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1026            ValidateTarget {
1027                depth: ValidateDepth::Sample,
1028                ..Default::default()
1029            },
1030        )
1031        .expect("snapshot + change datasets are complete");
1032
1033        let json: serde_json::Value =
1034            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1035        let exports = json["exports"].as_array().unwrap();
1036        let names: Vec<&str> = exports
1037            .iter()
1038            .map(|e| e["export_name"].as_str().unwrap())
1039            .collect();
1040        // The snapshot got its own verdict, and the change prefix its own.
1041        assert!(
1042            names.contains(&"cdc/snapshot"),
1043            "snapshot certified: {names:?}"
1044        );
1045        assert!(names.contains(&"cdc"), "change-prefix verdict: {names:?}");
1046        // The change-prefix verdict carries NO untracked-snapshot advisory.
1047        let change = exports.iter().find(|e| e["export_name"] == "cdc").unwrap();
1048        let failures = change["verification"]["failures"].as_array().unwrap();
1049        assert!(
1050            failures.iter().all(|f| f["kind"] != "untracked_object"),
1051            "snapshot files must not read as untracked surplus: {failures:?}"
1052        );
1053    }
1054
1055    fn write_single_cdc_cfg(dir: &Path, base: &Path) -> std::path::PathBuf {
1056        let cfg = dir.join("rivet-single-cdc.yaml");
1057        let yaml = format!(
1058            "source:\n  type: mysql\n  url: mysql://nobody@localhost/nope\nexports:\n  - name: cdc\n    table: t\n    mode: cdc\n    format: parquet\n    cdc:\n      initial: snapshot\n      server_id: 1\n      checkpoint: ./ck\n    destination:\n      type: local\n      path: \"{}\"\n",
1059            base.to_string_lossy()
1060        );
1061        std::fs::write(&cfg, yaml).unwrap();
1062        cfg
1063    }
1064
1065    #[test]
1066    fn missing_part_fails_the_command() {
1067        let dir = tempfile::tempdir().unwrap();
1068        let prefix = dir.path().join("out");
1069        let m = success_manifest(vec![ManifestPart {
1070            part_id: 1,
1071            path: "part-000001.parquet".into(),
1072            rows: 10,
1073            size_bytes: 4,
1074            content_fingerprint: "xxh3:1111111111111111".into(),
1075            content_md5: String::new(),
1076            status: PartStatus::Committed,
1077        }]);
1078        stage_dataset(&prefix, &m); // the part itself is never written
1079        let cfg = write_cfg(dir.path(), &prefix);
1080
1081        let err = run_validate_command(
1082            cfg.to_str().unwrap(),
1083            Some("orders"),
1084            ValidateOutputFormat::Json(None),
1085            ValidateTarget::default(),
1086        )
1087        .expect_err("a missing committed part must fail verification");
1088        assert!(
1089            format!("{err:#}").contains("1 export(s) failed verification"),
1090            "got: {err:#}"
1091        );
1092    }
1093
1094    // ── finding #20: operator-pinned --prefix requires a manifest ────────────
1095
1096    /// `--prefix` at a real, complete dataset still passes — the normal
1097    /// "validate exactly this prefix" case must not regress.
1098    #[test]
1099    fn prefix_override_with_real_manifest_passes() {
1100        let dir = tempfile::tempdir().unwrap();
1101        let prefix = dir.path().join("out");
1102        stage_dataset(&prefix, &success_manifest(Vec::new()));
1103        let cfg = write_cfg(dir.path(), &prefix);
1104
1105        run_validate_command(
1106            cfg.to_str().unwrap(),
1107            Some("orders"),
1108            ValidateOutputFormat::Json(None),
1109            ValidateTarget {
1110                prefix_override: Some(prefix.to_string_lossy().into_owned()),
1111                ..Default::default()
1112            },
1113        )
1114        .expect("a real dataset under a pinned --prefix must pass");
1115    }
1116
1117    /// `--prefix` at a never-written directory FAILS (exit non-zero): the
1118    /// operator asserted a dataset lives here, so an absent manifest is a
1119    /// refusal reason, not the benign legacy-run pass. This is the in-process
1120    /// twin of the live `audit_validate_absent_prefix_can_fail` roast.
1121    #[test]
1122    fn prefix_override_at_absent_manifest_fails() {
1123        let dir = tempfile::tempdir().unwrap();
1124        // The export's config destination is irrelevant — `--prefix` overrides
1125        // it. Point the override at a dir that exists but was never written.
1126        let cfg_prefix = dir.path().join("cfg_dest");
1127        std::fs::create_dir_all(&cfg_prefix).unwrap();
1128        let cfg = write_cfg(dir.path(), &cfg_prefix);
1129        let empty_prefix = dir.path().join("never_written");
1130        std::fs::create_dir_all(&empty_prefix).unwrap();
1131
1132        let report = dir.path().join("report.json");
1133        let err = run_validate_command(
1134            cfg.to_str().unwrap(),
1135            Some("orders"),
1136            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1137            ValidateTarget {
1138                prefix_override: Some(empty_prefix.to_string_lossy().into_owned()),
1139                ..Default::default()
1140            },
1141        )
1142        .expect_err("a never-written prefix pinned via --prefix must fail, not legacy-pass");
1143        assert!(
1144            format!("{err:#}").contains("1 export(s) failed verification"),
1145            "got: {err:#}"
1146        );
1147
1148        // The verdict (written before the bail) carries the explicit reason so
1149        // the operator sees why the gate refused, not a bare exit code.
1150        let json: serde_json::Value =
1151            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1152        let verification = &json["exports"][0]["verification"];
1153        assert_eq!(verification["manifest_found"], false);
1154        assert_eq!(verification["legacy_run"], false);
1155        assert_eq!(
1156            verification["failures"][0]["kind"],
1157            "manifest_required_but_absent"
1158        );
1159    }
1160
1161    /// Without `--prefix`, an absent manifest stays the benign M6 legacy-run
1162    /// pass (exit 0) — today's behaviour is preserved for config-resolved
1163    /// destinations that may legitimately be pre-0.7.0 prefixes.
1164    #[test]
1165    fn absent_manifest_without_prefix_override_stays_legacy_pass() {
1166        let dir = tempfile::tempdir().unwrap();
1167        let prefix = dir.path().join("out");
1168        std::fs::create_dir_all(&prefix).unwrap(); // exists, but no manifest
1169        let cfg = write_cfg(dir.path(), &prefix);
1170
1171        run_validate_command(
1172            cfg.to_str().unwrap(),
1173            Some("orders"),
1174            ValidateOutputFormat::Json(None),
1175            ValidateTarget::default(), // no --prefix
1176        )
1177        .expect("an absent manifest with no pinned --prefix is a legacy pass (exit 0)");
1178    }
1179
1180    // ── graded verify layer (--depth) end-to-end ─────────────────────────
1181
1182    /// Stage a dataset that passes sections 1-5 (manifest reads + is
1183    /// self-consistent, the single part is present at the recorded size,
1184    /// `_SUCCESS` matches) **but** records a non-empty `column_checksums`, so
1185    /// the Form B re-read is *reachable*. The part body is deliberately NOT
1186    /// valid Parquet, so if Form B runs it errors on the Parquet open — making
1187    /// "did Form B run?" observable as a pass/fail of the command.
1188    fn stage_dataset_form_b_would_fail(prefix: &Path) {
1189        std::fs::create_dir_all(prefix).unwrap();
1190        // 4-byte non-Parquet body; the manifest records size 4 so the part
1191        // reconcile (size-only, empty content_md5) passes.
1192        let part_body: &[u8] = b"AAAA";
1193        std::fs::write(prefix.join("part-000001.parquet"), part_body).unwrap();
1194
1195        let mut m = success_manifest(vec![ManifestPart {
1196            part_id: 1,
1197            path: "part-000001.parquet".into(),
1198            rows: 1,
1199            size_bytes: part_body.len() as u64,
1200            content_fingerprint: "xxh3:1111111111111111".into(),
1201            content_md5: String::new(),
1202            status: PartStatus::Committed,
1203        }]);
1204        // Non-empty → Form B does NOT early-return; it proceeds to read the
1205        // (garbage) part as Parquet and fail.
1206        m.column_checksums = Some(vec![crate::manifest::ColumnChecksum {
1207            name: "id".into(),
1208            checksum: "0".into(),
1209        }]);
1210        stage_dataset(prefix, &m);
1211    }
1212
1213    #[test]
1214    fn sample_depth_does_not_run_form_b() {
1215        // At `--depth sample` the structural checks (parts present, _SUCCESS,
1216        // self-consistency) all pass and the Form B value re-read is skipped —
1217        // so the command succeeds even though the part body is not real Parquet.
1218        let dir = tempfile::tempdir().unwrap();
1219        let prefix = dir.path().join("out");
1220        stage_dataset_form_b_would_fail(&prefix);
1221        let cfg = write_cfg(dir.path(), &prefix);
1222
1223        let report = dir.path().join("report.json");
1224        run_validate_command(
1225            cfg.to_str().unwrap(),
1226            Some("orders"),
1227            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1228            ValidateTarget {
1229                depth: ValidateDepth::Sample,
1230                ..Default::default()
1231            },
1232        )
1233        .expect("sample depth skips Form B, so a non-Parquet part still passes");
1234
1235        let json: serde_json::Value =
1236            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1237        let verification = &json["exports"][0]["verification"];
1238        assert_eq!(verification["passed"], true);
1239        assert_eq!(verification["parts_verified"], 1, "sample reconciles parts");
1240        assert_eq!(verification["depth_level"], "sample");
1241    }
1242
1243    #[test]
1244    fn full_depth_runs_form_b() {
1245        // The contrast: identical dataset, `--depth full`. Sections 1-5 still
1246        // pass (so `manifest_verified` is true and Form B is gated open), Form B
1247        // re-reads the part, fails to parse it as Parquet, and the command exits
1248        // non-zero. Proves Form B runs at full depth and only at full depth.
1249        let dir = tempfile::tempdir().unwrap();
1250        let prefix = dir.path().join("out");
1251        stage_dataset_form_b_would_fail(&prefix);
1252        let cfg = write_cfg(dir.path(), &prefix);
1253
1254        let err = run_validate_command(
1255            cfg.to_str().unwrap(),
1256            Some("orders"),
1257            ValidateOutputFormat::Json(None),
1258            ValidateTarget {
1259                depth: ValidateDepth::Full,
1260                ..Default::default()
1261            },
1262        )
1263        .expect_err("full depth runs Form B, which fails on a non-Parquet part");
1264        assert!(
1265            format!("{err:#}").contains("1 export(s) failed verification"),
1266            "got: {err:#}"
1267        );
1268    }
1269
1270    #[test]
1271    fn json_report_carries_failure_code_and_depth_level() {
1272        // render_json injects the stable `RIVET_VERIFY_*` code next to each
1273        // failure's `kind`, and the verdict carries the `depth_level` it ran at.
1274        // A missing committed part gives us a fatal failure to inspect.
1275        let dir = tempfile::tempdir().unwrap();
1276        let prefix = dir.path().join("out");
1277        let m = success_manifest(vec![ManifestPart {
1278            part_id: 1,
1279            path: "part-000001.parquet".into(),
1280            rows: 10,
1281            size_bytes: 4,
1282            content_fingerprint: "xxh3:1111111111111111".into(),
1283            content_md5: String::new(),
1284            status: PartStatus::Committed,
1285        }]);
1286        stage_dataset(&prefix, &m); // the part itself is never written → PartMissing
1287        let cfg = write_cfg(dir.path(), &prefix);
1288
1289        let report = dir.path().join("report.json");
1290        let _ = run_validate_command(
1291            cfg.to_str().unwrap(),
1292            Some("orders"),
1293            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1294            ValidateTarget {
1295                depth: ValidateDepth::Sample,
1296                ..Default::default()
1297            },
1298        )
1299        .expect_err("a missing part fails the command");
1300
1301        let json: serde_json::Value =
1302            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1303        let verification = &json["exports"][0]["verification"];
1304        // depth_level surfaces the level the pass ran at.
1305        assert_eq!(verification["depth_level"], "sample");
1306        // The PartMissing failure carries BOTH its stable kind and its code.
1307        let failure = &verification["failures"][0];
1308        assert_eq!(failure["kind"], "part_missing");
1309        assert_eq!(failure["code"], "RIVET_VERIFY_PART_MISSING");
1310        // The per-variant fields ride alongside (the derive output is widened,
1311        // not replaced).
1312        assert_eq!(failure["part_id"], 1);
1313    }
1314
1315    #[test]
1316    fn json_warning_entry_also_carries_its_code() {
1317        // The advisory `warnings` lens carries the code too — an untracked
1318        // surplus surfaces `RIVET_VERIFY_UNTRACKED_OBJECT` without flipping exit.
1319        let dir = tempfile::tempdir().unwrap();
1320        let prefix = dir.path().join("out");
1321        stage_dataset(&prefix, &success_manifest(Vec::new()));
1322        std::fs::write(prefix.join("rogue.parquet"), b"XX").unwrap();
1323        let cfg = write_cfg(dir.path(), &prefix);
1324
1325        let report = dir.path().join("report.json");
1326        run_validate_command(
1327            cfg.to_str().unwrap(),
1328            Some("orders"),
1329            ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1330            ValidateTarget::default(),
1331        )
1332        .expect("advisory untracked surplus must not flip the exit code");
1333
1334        let json: serde_json::Value =
1335            serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1336        let warning = &json["warnings"][0]["warning"];
1337        assert_eq!(warning["kind"], "untracked_object");
1338        assert_eq!(warning["code"], "RIVET_VERIFY_UNTRACKED_OBJECT");
1339        // And the default depth (full) is recorded on the verdict.
1340        assert_eq!(json["exports"][0]["verification"]["depth_level"], "full");
1341    }
1342}