use super::super::args::{CiArgs, JudgeArgs, RunArgs};
use super::pipeline_error::{elapsed_ms, PipelineError};
use super::run_output::decide_run_outcome;
use super::runner_builder::{build_runner, ensure_parent_dir};
use crate::exit_codes::{
reason_for_unloadable_explicit_config, ExitCodeVersion, ReasonCode, RunOutcome,
};
use std::path::PathBuf;
use std::time::Instant;
#[derive(Clone)]
pub(crate) struct PipelineInput {
pub config: PathBuf,
pub db: PathBuf,
pub trace_file: Option<PathBuf>,
pub baseline: Option<PathBuf>,
pub export_baseline: Option<PathBuf>,
pub strict: bool,
pub rerun_failures: u32,
pub quarantine_mode: String,
pub embedder: String,
pub embedding_model: String,
pub refresh_embeddings: bool,
pub incremental: bool,
pub refresh_cache: bool,
pub no_cache: bool,
pub judge: JudgeArgs,
pub replay_strict: bool,
pub deny_deprecations: bool,
pub allow_ineffective_assertions: bool,
pub redact_prompts: bool,
pub exit_codes: ExitCodeVersion,
pub ingest_trace_on_replay_strict: bool,
pub strict_zero_reruns: bool,
}
impl PipelineInput {
pub(crate) fn from_run(args: &RunArgs) -> Self {
Self {
config: args.config.clone(),
db: args.db.clone(),
trace_file: args.trace_file.clone(),
baseline: args.baseline.clone(),
export_baseline: args.export_baseline.clone(),
strict: args.strict,
rerun_failures: args.rerun_failures,
quarantine_mode: args.quarantine_mode.clone(),
embedder: args.embedder.clone(),
embedding_model: args.embedding_model.clone(),
refresh_embeddings: args.refresh_embeddings,
incremental: args.incremental,
refresh_cache: args.refresh_cache,
no_cache: args.no_cache,
judge: args.judge.clone(),
replay_strict: args.replay_strict,
deny_deprecations: args.deny_deprecations,
allow_ineffective_assertions: args.allow_ineffective_assertions,
redact_prompts: args.redact_prompts,
exit_codes: args.exit_codes,
ingest_trace_on_replay_strict: false,
strict_zero_reruns: false,
}
}
pub(crate) fn from_ci(args: &CiArgs) -> Self {
Self {
config: args.config.clone(),
db: args.db.clone(),
trace_file: args.trace_file.clone(),
baseline: args.baseline.clone(),
export_baseline: args.export_baseline.clone(),
strict: args.strict,
rerun_failures: args.rerun_failures,
quarantine_mode: args.quarantine_mode.clone(),
embedder: args.embedder.clone(),
embedding_model: args.embedding_model.clone(),
refresh_embeddings: args.refresh_embeddings,
incremental: args.incremental,
refresh_cache: args.refresh_cache,
no_cache: args.no_cache,
judge: args.judge.clone(),
replay_strict: args.replay_strict,
deny_deprecations: args.deny_deprecations,
allow_ineffective_assertions: args.allow_ineffective_assertions,
redact_prompts: args.redact_prompts,
exit_codes: args.exit_codes,
ingest_trace_on_replay_strict: true,
strict_zero_reruns: true,
}
}
}
pub(crate) struct PipelineSuccess {
pub cfg: assay_core::model::EvalConfig,
pub artifacts: assay_core::report::RunArtifacts,
pub outcome: RunOutcome,
pub timings: PipelineTimings,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct PipelineTimings {
pub total_ms: u64,
pub config_load_ms: Option<u64>,
pub ingest_ms: Option<u64>,
pub runner_build_ms: Option<u64>,
pub run_suite_ms: Option<u64>,
}
pub(crate) async fn execute_pipeline(
input: &PipelineInput,
legacy_mode: bool,
) -> Result<PipelineSuccess, PipelineError> {
let pipeline_start = Instant::now();
let mut timings = PipelineTimings::default();
if let Err(e) = ensure_parent_dir(&input.db) {
return Err(PipelineError::cfg_parse(
input.db.display().to_string(),
format!("Failed to create DB dir: {}", e),
));
}
if input.baseline.is_some() && input.export_baseline.is_some() {
return Err(PipelineError::invalid_args(
"Cannot use --baseline and --export-baseline together",
));
}
let cfg = {
let config_start = Instant::now();
match assay_core::config::load_config_with_cause(
&input.config,
assay_core::config::LoadOptions {
legacy_mode,
strict_unknown_fields: input.deny_deprecations,
allow_ineffective_assertions: input.allow_ineffective_assertions,
},
) {
Ok(c) => {
timings.config_load_ms = Some(elapsed_ms(config_start));
c
}
Err(e) => {
let path = input.config.display().to_string();
let msg = e.to_string();
return Err(match reason_for_unloadable_explicit_config(&e) {
ReasonCode::EMissingConfig => PipelineError::missing_cfg(path, msg),
_ => PipelineError::cfg_parse(path, msg),
});
}
}
};
if !cfg.is_legacy() && cfg.has_legacy_usage() {
let msg = format!(
"Deprecated policy file usage detected in version {} config. Run 'assay migrate' to inline policies.",
cfg.version
);
if input.deny_deprecations {
return Err(PipelineError::cfg_parse(
input.config.display().to_string(),
msg,
));
}
eprintln!("WARN: {}", msg);
}
if cfg.model == "trace" && input.trace_file.is_none() {
return Err(PipelineError::invalid_args(
"config uses model: trace, so --trace-file <PATH> is required",
));
}
let store = match assay_core::storage::Store::open(&input.db) {
Ok(s) => s,
Err(e) => {
return Err(PipelineError::cfg_parse(
input.db.display().to_string(),
format!("Failed to open DB: {}", e),
));
}
};
if input.ingest_trace_on_replay_strict {
if let Err(e) = store.init_schema() {
return Err(PipelineError::cfg_parse(
input.db.display().to_string(),
format!("Failed to init DB schema: {}", e),
));
}
if input.replay_strict {
if let Some(trace_path) = &input.trace_file {
let ingest_start = Instant::now();
match assay_core::trace::ingest::ingest_into_store(&store, trace_path) {
Ok(stats) => {
timings.ingest_ms = Some(elapsed_ms(ingest_start));
eprintln!(
"auto-ingest: loaded {} events into {} (from {})",
stats.event_count,
input.db.display(),
trace_path.display()
);
}
Err(e) => {
let msg = format!("Failed to ingest trace: {}", e);
return Err(if trace_path.exists() {
PipelineError::cfg_parse(trace_path.display().to_string(), msg)
} else {
PipelineError::from_run_error(
assay_core::errors::RunError::trace_not_found(
trace_path.display().to_string(),
"trace path does not exist",
),
)
});
}
}
}
}
}
let reruns = if input.strict_zero_reruns && input.strict {
0
} else {
input.rerun_failures
};
let runner_build_start = Instant::now();
let runner = build_runner(
store,
&input.trace_file,
&cfg,
reruns,
&input.quarantine_mode,
&input.embedder,
&input.embedding_model,
input.refresh_embeddings,
input.incremental,
input.refresh_cache || input.no_cache,
&input.judge,
&input.baseline,
input.config.clone(),
input.replay_strict,
)
.await;
timings.runner_build_ms = Some(elapsed_ms(runner_build_start));
let runner = match runner {
Ok(r) => r,
Err(e) => {
if let Some(diag) = assay_core::errors::try_map_error(&e) {
return Err(PipelineError::from_diagnostic(
input.config.display().to_string(),
&diag,
));
}
return Err(PipelineError::from_run_error(
assay_core::errors::RunError::from_anyhow(&e),
));
}
};
let total = cfg.tests.len();
if total > 0 {
eprintln!("Running {} tests...", total);
}
let progress = assay_core::report::console::default_progress_sink(total);
let run_suite_start = Instant::now();
let mut artifacts = runner.run_suite(&cfg, progress).await.map_err(|error| {
error
.downcast_ref::<assay_core::errors::RunError>()
.cloned()
.map(PipelineError::from_run_error)
.unwrap_or(PipelineError::Fatal(error))
})?;
timings.run_suite_ms = Some(elapsed_ms(run_suite_start));
if input.redact_prompts {
let policy = assay_core::redaction::RedactionPolicy::new(true);
for row in &mut artifacts.results {
policy.redact_judge_metadata(&mut row.details);
}
}
let outcome = decide_run_outcome(&artifacts.results, input.strict, input.exit_codes);
timings.total_ms = elapsed_ms(pipeline_start);
Ok(PipelineSuccess {
cfg,
artifacts,
outcome,
timings,
})
}