diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs new file mode 100644 index 000000000..788b43d14 --- /dev/null +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -0,0 +1,1972 @@ +//! Temporary compatibility importer for legacy SlateDB run history. +//! +//! Keep this source reader, its reports, and its verification path through the +//! 30-day run-history compatibility window. Remove them only after the +//! production evidence gate for that window has been accepted. + +use std::collections::HashSet; +use std::error::Error as StdError; +use std::fmt; + +use fabro_types::{EventEnvelope, RunEvent, RunId, RunProjection}; +use sqlx::SqlitePool; +#[cfg(test)] +use tokio::sync::Barrier; +use tracing::debug; + +use crate::keys::SlateKey; +use crate::slate::CachedRunProjection; +use crate::{Database, EventPayload, ListRunsQuery, RunProjectionReducer, RunSummaryStore, keys}; + +/// Count-only observations about the legacy catalog and session indexes. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryDiagnostics { + pub catalog_markers: u64, + pub empty_catalog_markers: u64, + pub session_reverse_rows: u64, +} + +/// Durable progress from one legacy run-history import attempt. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryImportReport { + pub scanned_source_runs: u64, + pub scanned_source_events: u64, + pub imported_runs: u64, + pub imported_events: u64, + pub verified_existing_runs: u64, + pub verified_existing_events: u64, + pub discarded_projection_only_rows: u64, + pub committed_run_transactions: u64, + pub diagnostics: LegacyRunHistoryDiagnostics, +} + +/// Aggregate proof from legacy-prefix and full-SQL-destination verification. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryVerificationReport { + pub source_runs: u64, + pub source_events: u64, + pub matched_prefix_runs: u64, + pub matched_prefix_events: u64, + pub target_runs: u64, + pub target_events: u64, + pub sql_only_runs: u64, + pub sql_only_events: u64, + pub diagnostics: LegacyRunHistoryDiagnostics, +} + +/// An import failure plus the durable progress completed before it. +pub struct LegacyRunHistoryImportError { + report: LegacyRunHistoryImportReport, + failure: LegacyRunHistoryImportFailure, +} + +impl LegacyRunHistoryImportError { + #[must_use] + pub fn report(&self) -> &LegacyRunHistoryImportReport { + &self.report + } + + /// Returns secondary errors encountered while rolling back a failed run + /// import transaction. + /// + /// The standard error source chain preserves the failure that interrupted + /// the import. Because that chain is linear, rollback errors are exposed + /// separately. + pub fn cleanup_errors(&self) -> impl Iterator { + let mut errors = Vec::new(); + self.failure.collect_cleanup_errors(&mut errors); + errors.into_iter() + } +} + +impl fmt::Debug for LegacyRunHistoryImportError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryImportError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyRunHistoryImportError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy run-history import failed after scanning {} runs and committing {} run transactions: {}", + self.report.scanned_source_runs, self.report.committed_run_transactions, self.failure + ) + } +} + +impl StdError for LegacyRunHistoryImportError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(self.failure.primary_failure()) + } +} + +/// A verification failure plus the aggregate proof completed before it. +pub struct LegacyRunHistoryVerificationError { + report: LegacyRunHistoryVerificationReport, + failure: LegacyRunHistoryVerificationFailure, +} + +impl LegacyRunHistoryVerificationError { + #[must_use] + pub fn report(&self) -> &LegacyRunHistoryVerificationReport { + &self.report + } +} + +impl fmt::Debug for LegacyRunHistoryVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryVerificationError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyRunHistoryVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy run-history verification failed after checking {} source runs and {} target runs: {}", + self.report.source_runs, self.report.target_runs, self.failure + ) + } +} + +impl StdError for LegacyRunHistoryVerificationError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.failure) + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryImportFailure { + #[error("reading and validating the legacy run-history source")] + Source(#[source] LegacyRunHistorySourceFailure), + #[error("starting the projection-only row cleanup transaction")] + BeginCleanup(#[source] sqlx::Error), + #[error("deleting projection-only run rows")] + DeleteProjectionOnlyRows(#[source] sqlx::Error), + #[error("committing the projection-only row cleanup")] + CommitCleanup(#[source] sqlx::Error), + #[error("starting a per-run import transaction")] + BeginRunTransaction(#[source] sqlx::Error), + #[error("reading an existing destination run history")] + ReadDestination(#[source] crate::Error), + #[error("the destination history is partial or conflicts with the legacy prefix")] + DestinationConflict, + #[error("replaying an existing destination run history")] + ReplayDestination(#[source] crate::Error), + #[error("verifying an existing destination run row")] + VerifyDestination(#[source] crate::Error), + #[error("inserting the imported final run row")] + InsertRun(#[source] crate::Error), + #[error("inserting an imported run event")] + InsertEvent(#[source] crate::Error), + #[error("committing a per-run import transaction")] + CommitRunTransaction(#[source] sqlx::Error), + #[error("rolling back a failed per-run import transaction")] + RollbackRunTransaction { + #[source] + source: sqlx::Error, + prior: Box, + }, + #[error("collecting count-only legacy index diagnostics")] + Diagnostics(#[source] LegacyRunHistoryDiagnosticsFailure), + #[error("a legacy run-history import counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryImportFailure { + fn kind(&self) -> &'static str { + self.into() + } + + fn primary_failure(&self) -> &Self { + match self { + Self::RollbackRunTransaction { prior, .. } => prior.primary_failure(), + _ => self, + } + } + + fn collect_cleanup_errors<'a>(&'a self, errors: &mut Vec<&'a (dyn StdError + 'static)>) { + if let Self::RollbackRunTransaction { source, prior } = self { + errors.push(source); + prior.collect_cleanup_errors(errors); + } + } +} + +impl fmt::Debug for LegacyRunHistoryImportFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut debug = formatter.debug_struct("LegacyRunHistoryImportFailure"); + debug.field("kind", &self.kind()); + if let Self::RollbackRunTransaction { prior, .. } = self { + debug.field("prior_failure", &prior.kind()); + } + debug.finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryVerificationFailure { + #[error("reading and validating the legacy run-history source")] + Source(#[source] LegacyRunHistorySourceFailure), + #[error("acquiring a SQLite verification connection")] + AcquireConnection(#[source] sqlx::Error), + #[error("reading a destination run history")] + ReadDestination(#[source] crate::Error), + #[error("SQLite is missing all or part of a legacy run-history prefix")] + MissingDestinationPrefix, + #[error("a destination run-history prefix conflicts with legacy JSON or sequence identity")] + DestinationPrefixConflict, + #[error("enumerating destination run rows")] + ListDestinationRuns(#[source] sqlx::Error), + #[error("a destination run row has an invalid identity")] + InvalidDestinationRunId, + #[error("a destination run has no event history")] + EmptyDestinationHistory, + #[error("replaying a destination run history")] + ReplayDestination(#[source] crate::Error), + #[error("verifying a destination run row")] + VerifyDestination(#[source] crate::Error), + #[error("collecting count-only legacy index diagnostics")] + Diagnostics(#[source] LegacyRunHistoryDiagnosticsFailure), + #[error("a legacy run-history verification counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryVerificationFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistoryVerificationFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryVerificationFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistorySourceFailure { + #[error("opening the legacy run-history source")] + OpenSource(#[source] crate::Error), + #[error("opening the legacy run-history scan")] + OpenScan(#[source] slatedb::Error), + #[error("reading the legacy run-history scan")] + ReadScan(#[source] slatedb::Error), + #[error("a legacy run-event key is not UTF-8")] + KeyUtf8(#[source] std::str::Utf8Error), + #[error("a legacy run-event key is not canonical")] + InvalidKey, + #[error("a legacy run-event value is not UTF-8")] + ValueUtf8(#[source] std::str::Utf8Error), + #[error("a legacy run-event value is not valid JSON")] + DecodeEvent(#[source] serde_json::Error), + #[error("a legacy run-event value does not match its key or event contract")] + ValidateEvent(#[source] crate::Error), + #[error("a legacy run history has invalid sequence ordering")] + InvalidSequence, + #[error("a legacy run history does not begin with sequence 1 run.created")] + InvalidFirstEvent, + #[error("replaying a legacy run history")] + Replay(#[source] crate::Error), + #[error("a legacy run-history source counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistorySourceFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistorySourceFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistorySourceFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryDiagnosticsFailure { + #[error("reading the legacy run catalog")] + ReadCatalog(#[source] crate::Error), + #[error("opening the legacy run source for diagnostics")] + OpenSource(#[source] crate::Error), + #[error("opening a legacy run-event probe")] + OpenEventProbe(#[source] slatedb::Error), + #[error("reading a legacy run-event probe")] + ReadEventProbe(#[source] slatedb::Error), + #[error("opening the legacy session reverse-row scan")] + OpenSessionScan(#[source] slatedb::Error), + #[error("reading the legacy session reverse-row scan")] + ReadSessionScan(#[source] slatedb::Error), + #[error("a legacy run-history diagnostics counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryDiagnosticsFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistoryDiagnosticsFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryDiagnosticsFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(Default)] +struct ImportControls { + #[cfg(test)] + source_after_events: Option, + #[cfg(test)] + after_run_inserted: Option>, +} + +impl ImportControls { + fn source_scan_error(&self, observed_events: u64) -> Option { + #[cfg(test)] + if self.source_after_events == Some(observed_events) { + return Some(slatedb::Error::unavailable( + "injected legacy run-history source failure".to_owned(), + )); + } + let _ = (self, observed_events); + None + } + + #[cfg(test)] + async fn after_run_inserted(&self) { + if let Some(barrier) = &self.after_run_inserted { + barrier.wait().await; + barrier.wait().await; + } + } +} + +struct ValidatedLegacyRunEvent { + run_id: RunId, + payload: EventPayload, + envelope: EventEnvelope, + event_json: String, +} + +struct ValidatedLegacyRunHistory { + run_id: RunId, + events: Vec, + current: CachedRunProjection, +} + +struct LegacyRunHistorySource { + entries: slatedb::DbIterator, + buffered: Option, + observed_events: u64, +} + +impl LegacyRunHistorySource { + async fn open(database: &Database) -> Result { + let source = database + .open_db() + .await + .map_err(LegacyRunHistorySourceFailure::OpenSource)?; + let prefix = SlateKey::new("runs").into_prefix(); + let entries = source + .scan_prefix(prefix) + .await + .map_err(LegacyRunHistorySourceFailure::OpenScan)?; + Ok(Self { + entries, + buffered: None, + observed_events: 0, + }) + } + + async fn next_run( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + let first = if let Some(entry) = self.buffered.take() { + parse_source_event(&entry.key, &entry.value)? + .expect("the buffered source entry belongs to the event namespace") + } else { + let Some(event) = self.next_event(controls).await? else { + return Ok(None); + }; + event + }; + let run_id = first.run_id; + let run_id_text = run_id.to_string(); + let mut events = vec![first]; + loop { + let Some(entry) = self.next_event_entry(controls).await? else { + break; + }; + if event_run_segment(&entry.key) != Some(run_id_text.as_bytes()) { + self.buffered = Some(entry); + break; + } + let event = parse_source_event(&entry.key, &entry.value)? + .expect("an event-namespace entry parses as an event or fails"); + events.push(event); + } + + if events[0].envelope.seq != 1 || events[0].envelope.event.event_name() != "run.created" { + return Err(LegacyRunHistorySourceFailure::InvalidFirstEvent); + } + if events + .windows(2) + .any(|pair| pair[0].envelope.seq >= pair[1].envelope.seq) + { + return Err(LegacyRunHistorySourceFailure::InvalidSequence); + } + let envelopes = events + .iter() + .map(|event| event.envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes) + .map_err(LegacyRunHistorySourceFailure::Replay)?; + let last_seq = events + .last() + .expect("a validated history contains at least one event") + .envelope + .seq; + let current = CachedRunProjection::from_projection(run_id, projection, last_seq); + Ok(Some(ValidatedLegacyRunHistory { + run_id, + events, + current, + })) + } + + async fn next_event( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + let Some(entry) = self.next_event_entry(controls).await? else { + return Ok(None); + }; + parse_source_event(&entry.key, &entry.value) + } + + async fn next_event_entry( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + loop { + if let Some(source) = + controls.and_then(|controls| controls.source_scan_error(self.observed_events)) + { + return Err(LegacyRunHistorySourceFailure::ReadScan(source)); + } + let Some(entry) = self + .entries + .next() + .await + .map_err(LegacyRunHistorySourceFailure::ReadScan)? + else { + return Ok(None); + }; + if event_run_segment(&entry.key).is_none() { + continue; + } + self.observed_events = self + .observed_events + .checked_add(1) + .ok_or(LegacyRunHistorySourceFailure::CounterOverflow)?; + return Ok(Some(entry)); + } + } +} + +impl Database { + /// Strictly imports legacy SlateDB run history into the inactive SQLite + /// run store, committing one complete run at a time. + /// + /// The caller must prevent writes to both stores for the duration of the + /// import. This operation does not establish a cross-store snapshot. + pub async fn import_legacy_run_history_into( + &self, + pool: &SqlitePool, + ) -> Result { + self.import_legacy_run_history_with_controls(pool, &ImportControls::default()) + .await + } + + async fn import_legacy_run_history_with_controls( + &self, + pool: &SqlitePool, + controls: &ImportControls, + ) -> Result { + let mut report = LegacyRunHistoryImportReport::default(); + let result = self + .run_legacy_run_history_import(pool, controls, &mut report) + .await; + debug_import_outcome(result.as_ref().map_or("failed", |()| "complete"), &report); + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyRunHistoryImportError { report, failure }), + } + } + + async fn run_legacy_run_history_import( + &self, + pool: &SqlitePool, + controls: &ImportControls, + report: &mut LegacyRunHistoryImportReport, + ) -> Result<(), LegacyRunHistoryImportFailure> { + discard_projection_only_rows(pool, report).await?; + let mut source = LegacyRunHistorySource::open(self) + .await + .map_err(LegacyRunHistoryImportFailure::Source)?; + + loop { + let history = match source.next_run(Some(controls)).await { + Ok(history) => history, + Err(error) => { + report.scanned_source_events = source.observed_events; + return Err(LegacyRunHistoryImportFailure::Source(error)); + } + }; + report.scanned_source_events = source.observed_events; + let Some(history) = history else { + break; + }; + import_checked_add(&mut report.scanned_source_runs, 1)?; + import_one_run(pool, controls, history, report).await?; + } + + report.diagnostics = self + .legacy_run_history_diagnostics() + .await + .map_err(LegacyRunHistoryImportFailure::Diagnostics)?; + Ok(()) + } + + /// Verifies every legacy history as an exact SQLite prefix and then + /// independently replays and verifies every SQLite run. + /// + /// The caller must prevent writes to both stores for the duration of + /// verification. This operation does not establish a cross-store snapshot. + pub async fn verify_legacy_run_history_in( + &self, + pool: &SqlitePool, + ) -> Result { + let mut report = LegacyRunHistoryVerificationReport::default(); + let result = self + .run_legacy_run_history_verification(pool, &mut report) + .await; + debug_verification_outcome(result.as_ref().map_or("failed", |()| "complete"), &report); + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyRunHistoryVerificationError { report, failure }), + } + } + + async fn run_legacy_run_history_verification( + &self, + pool: &SqlitePool, + report: &mut LegacyRunHistoryVerificationReport, + ) -> Result<(), LegacyRunHistoryVerificationFailure> { + let mut source_ids = HashSet::new(); + let mut source = LegacyRunHistorySource::open(self) + .await + .map_err(LegacyRunHistoryVerificationFailure::Source)?; + loop { + let history = match source.next_run(None).await { + Ok(history) => history, + Err(error) => { + report.source_events = source.observed_events; + return Err(LegacyRunHistoryVerificationFailure::Source(error)); + } + }; + report.source_events = source.observed_events; + let Some(history) = history else { + break; + }; + verification_checked_add(&mut report.source_runs, 1)?; + source_ids.insert(history.run_id); + verify_source_prefix(pool, &history).await?; + verification_checked_add(&mut report.matched_prefix_runs, 1)?; + verification_checked_add( + &mut report.matched_prefix_events, + usize_to_verification_count(history.events.len())?, + )?; + } + + let stored_ids: Vec = sqlx::query_scalar("SELECT id FROM runs ORDER BY id ASC") + .fetch_all(pool) + .await + .map_err(LegacyRunHistoryVerificationFailure::ListDestinationRuns)?; + for stored_id in stored_ids { + let run_id = stored_id + .parse::() + .map_err(|_| LegacyRunHistoryVerificationFailure::InvalidDestinationRunId)?; + let mut connection = pool + .acquire() + .await + .map_err(LegacyRunHistoryVerificationFailure::AcquireConnection)?; + let events = + RunSummaryStore::list_events_with_json_on_connection(&mut connection, &run_id) + .await + .map_err(LegacyRunHistoryVerificationFailure::ReadDestination)?; + if events.is_empty() { + return Err(LegacyRunHistoryVerificationFailure::EmptyDestinationHistory); + } + let current = replay_destination(&run_id, &events) + .map_err(LegacyRunHistoryVerificationFailure::ReplayDestination)?; + RunSummaryStore::verify_current_run_on_connection(&mut connection, ¤t) + .await + .map_err(LegacyRunHistoryVerificationFailure::VerifyDestination)?; + + let event_count = usize_to_verification_count(events.len())?; + verification_checked_add(&mut report.target_runs, 1)?; + verification_checked_add(&mut report.target_events, event_count)?; + if !source_ids.contains(&run_id) { + verification_checked_add(&mut report.sql_only_runs, 1)?; + verification_checked_add(&mut report.sql_only_events, event_count)?; + } + } + + report.diagnostics = self + .legacy_run_history_diagnostics() + .await + .map_err(LegacyRunHistoryVerificationFailure::Diagnostics)?; + Ok(()) + } + + async fn legacy_run_history_diagnostics( + &self, + ) -> Result { + let catalog_ids = self + .catalog_index() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)? + .list(&ListRunsQuery::default()) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?; + let mut diagnostics = LegacyRunHistoryDiagnostics { + catalog_markers: u64::try_from(catalog_ids.len()) + .map_err(|_| LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?, + ..LegacyRunHistoryDiagnostics::default() + }; + let source = self + .open_db() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenSource)?; + for run_id in catalog_ids { + let mut events = source + .scan_prefix(keys::run_events_prefix(&run_id)) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenEventProbe)?; + if events + .next() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadEventProbe)? + .is_none() + { + diagnostics.empty_catalog_markers = diagnostics + .empty_catalog_markers + .checked_add(1) + .ok_or(LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?; + } + } + let mut sessions = source + .scan_prefix(keys::sessions_by_id_prefix()) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenSessionScan)?; + while sessions + .next() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadSessionScan)? + .is_some() + { + diagnostics.session_reverse_rows = diagnostics + .session_reverse_rows + .checked_add(1) + .ok_or(LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?; + } + Ok(diagnostics) + } +} + +fn parse_source_event( + key: &[u8], + value: &[u8], +) -> Result, LegacyRunHistorySourceFailure> { + let key_text = std::str::from_utf8(key).map_err(LegacyRunHistorySourceFailure::KeyUtf8)?; + let segments = SlateKey::segments(key_text).collect::>(); + if segments.get(2).copied() != Some("events") { + return Ok(None); + } + let ["runs", run_id_text, "events", leaf] = segments.as_slice() else { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + }; + let run_id = run_id_text + .parse::() + .map_err(|_| LegacyRunHistorySourceFailure::InvalidKey)?; + let Some((sequence_text, epoch_ms_text)) = leaf.split_once('-') else { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + }; + if sequence_text.len() != 6 || !sequence_text.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + } + let sequence = sequence_text + .parse::() + .ok() + .filter(|sequence| (1..=keys::MAX_EVENT_SEQ).contains(sequence)) + .ok_or(LegacyRunHistorySourceFailure::InvalidKey)?; + let epoch_ms = epoch_ms_text + .parse::() + .map_err(|_| LegacyRunHistorySourceFailure::InvalidKey)?; + if keys::run_event_key(&run_id, sequence, epoch_ms).as_ref() != key { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + } + + let event_json = std::str::from_utf8(value) + .map_err(LegacyRunHistorySourceFailure::ValueUtf8)? + .to_owned(); + let payload: EventPayload = + serde_json::from_str(&event_json).map_err(LegacyRunHistorySourceFailure::DecodeEvent)?; + payload + .validate(&run_id) + .map_err(LegacyRunHistorySourceFailure::ValidateEvent)?; + let event = + RunEvent::try_from(&payload).map_err(LegacyRunHistorySourceFailure::ValidateEvent)?; + if event.run_id != run_id { + return Err(LegacyRunHistorySourceFailure::ValidateEvent( + crate::Error::RunEventMismatch { + run_id: run_id.to_string(), + seq: sequence, + field: "run_id", + }, + )); + } + Ok(Some(ValidatedLegacyRunEvent { + run_id, + payload, + envelope: EventEnvelope { + seq: sequence, + event, + }, + event_json, + })) +} + +fn event_run_segment(key: &[u8]) -> Option<&[u8]> { + let mut segments = key.split(|byte| *byte == 0); + (segments.next()? == b"runs").then_some(())?; + let run_id = segments.next()?; + (segments.next()? == b"events").then_some(run_id) +} + +async fn discard_projection_only_rows( + pool: &SqlitePool, + report: &mut LegacyRunHistoryImportReport, +) -> Result<(), LegacyRunHistoryImportFailure> { + let mut transaction = pool + .begin() + .await + .map_err(LegacyRunHistoryImportFailure::BeginCleanup)?; + let result = sqlx::query( + r" +DELETE FROM runs +WHERE NOT EXISTS ( + SELECT 1 FROM run_events WHERE run_events.run_id = runs.id +) +", + ) + .execute(&mut *transaction) + .await + .map_err(LegacyRunHistoryImportFailure::DeleteProjectionOnlyRows)?; + let discarded = result.rows_affected(); + transaction + .commit() + .await + .map_err(LegacyRunHistoryImportFailure::CommitCleanup)?; + report.discarded_projection_only_rows = discarded; + Ok(()) +} + +async fn import_one_run( + pool: &SqlitePool, + controls: &ImportControls, + history: ValidatedLegacyRunHistory, + report: &mut LegacyRunHistoryImportReport, +) -> Result<(), LegacyRunHistoryImportFailure> { + #[cfg(not(test))] + let _ = controls; + let mut transaction = pool + .begin() + .await + .map_err(LegacyRunHistoryImportFailure::BeginRunTransaction)?; + let result = async { + let has_destination: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM run_events WHERE run_id = ?)") + .bind(history.run_id.to_string()) + .fetch_one(&mut *transaction) + .await + .map_err(|source| { + LegacyRunHistoryImportFailure::ReadDestination(crate::Error::Sqlite(source)) + })?; + let mut updated = *report; + if has_destination { + let destination = RunSummaryStore::list_events_with_json_on_connection( + &mut transaction, + &history.run_id, + ) + .await + .map_err(LegacyRunHistoryImportFailure::ReadDestination)?; + require_exact_prefix(&history, &destination) + .map_err(|()| LegacyRunHistoryImportFailure::DestinationConflict)?; + let current = replay_destination(&history.run_id, &destination) + .map_err(LegacyRunHistoryImportFailure::ReplayDestination)?; + RunSummaryStore::verify_current_run_on_connection(&mut transaction, ¤t) + .await + .map_err(LegacyRunHistoryImportFailure::VerifyDestination)?; + import_checked_add(&mut updated.verified_existing_runs, 1)?; + import_checked_add( + &mut updated.verified_existing_events, + usize_to_import_count(destination.len())?, + )?; + } else { + RunSummaryStore::insert_imported_run_on_connection(&mut transaction, &history.current) + .await + .map_err(LegacyRunHistoryImportFailure::InsertRun)?; + #[cfg(test)] + controls.after_run_inserted().await; + for event in &history.events { + RunSummaryStore::insert_imported_event_on_connection( + &mut transaction, + &history.run_id, + &event.payload, + &event.envelope, + &event.event_json, + ) + .await + .map_err(LegacyRunHistoryImportFailure::InsertEvent)?; + } + import_checked_add(&mut updated.imported_runs, 1)?; + import_checked_add( + &mut updated.imported_events, + usize_to_import_count(history.events.len())?, + )?; + } + import_checked_add(&mut updated.committed_run_transactions, 1)?; + Ok(updated) + } + .await; + + match result { + Ok(updated) => { + transaction + .commit() + .await + .map_err(LegacyRunHistoryImportFailure::CommitRunTransaction)?; + *report = updated; + Ok(()) + } + Err(prior) => match transaction.rollback().await { + Ok(()) => Err(prior), + Err(source) => Err(LegacyRunHistoryImportFailure::RollbackRunTransaction { + source, + prior: Box::new(prior), + }), + }, + } +} + +async fn verify_source_prefix( + pool: &SqlitePool, + history: &ValidatedLegacyRunHistory, +) -> Result<(), LegacyRunHistoryVerificationFailure> { + let mut connection = pool + .acquire() + .await + .map_err(LegacyRunHistoryVerificationFailure::AcquireConnection)?; + let has_destination: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM run_events WHERE run_id = ?)") + .bind(history.run_id.to_string()) + .fetch_one(&mut *connection) + .await + .map_err(|source| { + LegacyRunHistoryVerificationFailure::ReadDestination(crate::Error::Sqlite(source)) + })?; + if !has_destination { + return Err(LegacyRunHistoryVerificationFailure::MissingDestinationPrefix); + } + let destination = + RunSummaryStore::list_events_with_json_on_connection(&mut connection, &history.run_id) + .await + .map_err(LegacyRunHistoryVerificationFailure::ReadDestination)?; + if destination.len() < history.events.len() { + return Err(LegacyRunHistoryVerificationFailure::MissingDestinationPrefix); + } + require_exact_prefix(history, &destination) + .map_err(|()| LegacyRunHistoryVerificationFailure::DestinationPrefixConflict) +} + +fn require_exact_prefix( + history: &ValidatedLegacyRunHistory, + destination: &[(EventEnvelope, String)], +) -> Result<(), ()> { + if destination.len() < history.events.len() { + return Err(()); + } + if history + .events + .iter() + .zip(destination) + .any(|(source, (target, target_json))| { + source.envelope.seq != target.seq || source.event_json != *target_json + }) + { + return Err(()); + } + Ok(()) +} + +fn replay_destination( + run_id: &RunId, + events: &[(EventEnvelope, String)], +) -> crate::Result { + let Some((first, _event_json)) = events.first() else { + return Err(crate::Error::InvalidEvent( + "run projection requires an event".to_owned(), + )); + }; + if first.seq != 1 { + return Err(crate::Error::RunEventMismatch { + run_id: run_id.to_string(), + seq: first.seq, + field: "seq", + }); + } + let envelopes = events + .iter() + .map(|(envelope, _event_json)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let last_seq = envelopes + .last() + .expect("a destination history validated as nonempty") + .seq; + Ok(CachedRunProjection::from_projection( + *run_id, projection, last_seq, + )) +} + +fn usize_to_import_count(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyRunHistoryImportFailure::CounterOverflow) +} + +fn import_checked_add(value: &mut u64, amount: u64) -> Result<(), LegacyRunHistoryImportFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyRunHistoryImportFailure::CounterOverflow)?; + Ok(()) +} + +fn usize_to_verification_count(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyRunHistoryVerificationFailure::CounterOverflow) +} + +fn verification_checked_add( + value: &mut u64, + amount: u64, +) -> Result<(), LegacyRunHistoryVerificationFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyRunHistoryVerificationFailure::CounterOverflow)?; + Ok(()) +} + +fn debug_import_outcome(outcome: &'static str, report: &LegacyRunHistoryImportReport) { + debug!( + outcome, + scanned_source_runs = report.scanned_source_runs, + scanned_source_events = report.scanned_source_events, + imported_runs = report.imported_runs, + imported_events = report.imported_events, + verified_existing_runs = report.verified_existing_runs, + verified_existing_events = report.verified_existing_events, + discarded_projection_only_rows = report.discarded_projection_only_rows, + committed_run_transactions = report.committed_run_transactions, + catalog_markers = report.diagnostics.catalog_markers, + empty_catalog_markers = report.diagnostics.empty_catalog_markers, + session_reverse_rows = report.diagnostics.session_reverse_rows, + "Legacy run-history import finished" + ); +} + +fn debug_verification_outcome(outcome: &'static str, report: &LegacyRunHistoryVerificationReport) { + debug!( + outcome, + source_runs = report.source_runs, + source_events = report.source_events, + matched_prefix_runs = report.matched_prefix_runs, + matched_prefix_events = report.matched_prefix_events, + target_runs = report.target_runs, + target_events = report.target_events, + sql_only_runs = report.sql_only_runs, + sql_only_events = report.sql_only_events, + catalog_markers = report.diagnostics.catalog_markers, + empty_catalog_markers = report.diagnostics.empty_catalog_markers, + session_reverse_rows = report.diagnostics.session_reverse_rows, + "Legacy run-history verification finished" + ); +} + +#[cfg(test)] +mod tests { + use std::error::Error as _; + use std::sync::Arc; + use std::time::Duration; + + use chrono::{TimeZone as _, Utc}; + use fabro_types::{ + Graph, RunEvent, RunId, RunProjection, SessionId, WorkflowSettings, test_support, + }; + use fabro_util::error; + use object_store::memory::InMemory; + use tokio::sync::Barrier; + use ulid::Ulid; + + use super::{ + ImportControls, LegacyRunHistoryDiagnostics, LegacyRunHistoryImportError, + LegacyRunHistoryImportFailure, LegacyRunHistoryImportReport, + LegacyRunHistoryVerificationFailure, LegacyRunHistoryVerificationReport, + parse_source_event, + }; + use crate::keys::SlateKey; + use crate::slate::CachedRunProjection; + use crate::{ + Database, EventEnvelope, EventPayload, RunProjectionReducer, RunSummaryStore, keys, + test_support as store_test_support, + }; + + type TestResult = std::result::Result>; + + struct TestContext { + _directory: tempfile::TempDir, + source: Database, + source_db: slatedb::Db, + sqlite: sqlx::SqlitePool, + } + + impl TestContext { + async fn new() -> TestResult { + let directory = tempfile::tempdir()?; + let sqlite = + fabro_db::Database::connect(directory.path().join("fabro.sqlite3")).await?; + sqlite.migrate().await?; + let sqlite = sqlite.clone_pool(); + let source = Database::new( + Arc::new(InMemory::new()), + "legacy-run-history-import-tests", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + store_test_support::test_run_summary_store(), + ); + let source_db = source.open_db().await?; + Ok(Self { + _directory: directory, + source, + source_db, + sqlite, + }) + } + + async fn put_event( + &self, + run_id: &RunId, + seq: u32, + epoch_ms: i64, + event_json: &str, + ) -> TestResult<()> { + self.source_db + .put( + keys::run_event_key(run_id, seq, epoch_ms), + event_json.as_bytes(), + ) + .await?; + Ok(()) + } + + async fn put_raw(&self, key: impl AsRef<[u8]>, value: &[u8]) -> TestResult<()> { + self.source_db.put(key, value).await?; + Ok(()) + } + + async fn source_entries(&self) -> TestResult, Vec)>> { + let mut entries = self.source_db.scan_prefix(Vec::::new()).await?; + let mut snapshot = Vec::new(); + while let Some(entry) = entries.next().await? { + snapshot.push((entry.key.to_vec(), entry.value.to_vec())); + } + Ok(snapshot) + } + + async fn import(&self) -> TestResult { + Ok(self + .source + .import_legacy_run_history_into(&self.sqlite) + .await?) + } + } + + fn run_id(index: u128) -> RunId { + RunId::from(Ulid::from_parts( + 1_788_000_000_000 + u64::try_from(index).unwrap(), + index, + )) + } + + fn event_value( + run_id: &RunId, + seq: u32, + event: &str, + properties: &serde_json::Value, + ) -> serde_json::Value { + serde_json::json!({ + "id": format!("evt-{seq}-{event}"), + "ts": Utc + .timestamp_millis_opt(1_788_000_000_000 + i64::from(seq)) + .single() + .unwrap() + .to_rfc3339(), + "run_id": run_id.to_string(), + "event": event, + "properties": properties, + }) + } + + fn created_value(run_id: &RunId, title: &str) -> serde_json::Value { + event_value( + run_id, + 1, + "run.created", + &serde_json::json!({ + "title": title, + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "workflow_slug": "test-workflow", + "labels": {}, + "provenance": test_support::test_run_provenance(), + }), + ) + } + + fn submitted_value(run_id: &RunId, seq: u32) -> serde_json::Value { + event_value(run_id, seq, "run.submitted", &serde_json::json!({})) + } + + fn decode_event( + run_id: &RunId, + seq: u32, + event_json: &str, + ) -> TestResult<(EventPayload, EventEnvelope)> { + let payload: EventPayload = serde_json::from_str(event_json)?; + payload.validate(run_id)?; + let event = RunEvent::try_from(&payload)?; + Ok((payload, EventEnvelope { seq, event })) + } + + async fn seed_destination_history( + pool: &sqlx::SqlitePool, + run_id: &RunId, + events: &[(u32, String)], + ) -> TestResult<()> { + let decoded = events + .iter() + .map(|(seq, event_json)| decode_event(run_id, *seq, event_json)) + .collect::>>()?; + let envelopes = decoded + .iter() + .map(|(_payload, envelope)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let current = + CachedRunProjection::from_projection(*run_id, projection, events.last().unwrap().0); + let mut transaction = pool.begin().await?; + RunSummaryStore::insert_imported_run_on_connection(&mut transaction, ¤t).await?; + for ((_, event_json), (payload, envelope)) in events.iter().zip(&decoded) { + RunSummaryStore::insert_imported_event_on_connection( + &mut transaction, + run_id, + payload, + envelope, + event_json, + ) + .await?; + } + transaction.commit().await?; + Ok(()) + } + + async fn append_destination_event( + pool: &sqlx::SqlitePool, + run_id: &RunId, + prior: &[(u32, String)], + next: (u32, String), + ) -> TestResult<()> { + let mut all = prior.to_vec(); + all.push(next.clone()); + let decoded = all + .iter() + .map(|(seq, event_json)| decode_event(run_id, *seq, event_json)) + .collect::>>()?; + let envelopes = decoded + .iter() + .map(|(_payload, envelope)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let current = CachedRunProjection::from_projection(*run_id, projection, next.0); + let mut transaction = pool.begin().await?; + RunSummaryStore::append_event_on_connection( + &mut transaction, + prior.last().unwrap().0, + ¤t, + &decoded.last().unwrap().0, + ) + .await?; + transaction.commit().await?; + Ok(()) + } + + #[test] + fn legacy_run_history_import_exposes_rollback_cleanup_errors() { + let error = LegacyRunHistoryImportError { + report: LegacyRunHistoryImportReport::default(), + failure: LegacyRunHistoryImportFailure::RollbackRunTransaction { + source: sqlx::Error::Protocol("injected rollback failure".to_owned()), + prior: Box::new(LegacyRunHistoryImportFailure::DestinationConflict), + }, + }; + + assert_eq!( + error.source().map(ToString::to_string), + Some( + "the destination history is partial or conflicts with the legacy prefix".to_owned() + ) + ); + assert!( + error + .cleanup_errors() + .any(|source| source.downcast_ref::().is_some()), + "rollback source was absent from cleanup errors" + ); + } + + #[tokio::test] + async fn legacy_run_history_imports_exact_json_gaps_and_count_only_diagnostics() + -> TestResult<()> { + let context = TestContext::new().await?; + let first = run_id(1); + let second = run_id(2); + let empty_marker = run_id(3); + let stale = run_id(4); + let first_created = serde_json::to_string_pretty(&created_value(&first, "first"))?; + let first_submitted = serde_json::to_string(&submitted_value(&first, 4))?; + let second_created = serde_json::to_string(&created_value(&second, "second"))?; + context.put_event(&first, 1, 10, &first_created).await?; + context.put_event(&first, 4, 40, &first_submitted).await?; + context.put_event(&second, 1, 20, &second_created).await?; + context.source.catalog_index().await?.add(&first).await?; + context + .source + .catalog_index() + .await? + .add(&empty_marker) + .await?; + context + .source + .put_session_run_index(&SessionId::new(), &first) + .await?; + + let stale_created = serde_json::to_string(&created_value(&stale, "stale"))?; + seed_destination_history(&context.sqlite, &stale, &[(1, stale_created)]).await?; + sqlx::query("DELETE FROM run_events WHERE run_id = ?") + .bind(stale.to_string()) + .execute(&context.sqlite) + .await?; + sqlx::query("UPDATE runs SET summary_json = '{}' WHERE id = ?") + .bind(stale.to_string()) + .execute(&context.sqlite) + .await?; + let source_before = context.source_entries().await?; + + let report = context.import().await?; + + assert_eq!(report, LegacyRunHistoryImportReport { + scanned_source_runs: 2, + scanned_source_events: 3, + imported_runs: 2, + imported_events: 3, + discarded_projection_only_rows: 1, + committed_run_transactions: 2, + diagnostics: LegacyRunHistoryDiagnostics { + catalog_markers: 2, + empty_catalog_markers: 1, + session_reverse_rows: 1, + }, + ..LegacyRunHistoryImportReport::default() + }); + let stored: Vec<(i64, String)> = + sqlx::query_as("SELECT seq, event_json FROM run_events WHERE run_id = ? ORDER BY seq") + .bind(first.to_string()) + .fetch_all(&context.sqlite) + .await?; + assert_eq!(stored, vec![(1, first_created), (4, first_submitted)]); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(first.to_string()) + .fetch_one(&context.sqlite) + .await?, + 4 + ); + assert_eq!(context.source_entries().await?, source_before); + let verification = context + .source + .verify_legacy_run_history_in(&context.sqlite) + .await?; + assert_eq!(verification.target_runs, 2); + assert_eq!(verification.target_events, 3); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_retry_uses_complete_destination_histories_as_progress() + -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(10); + context + .put_event( + &run_id, + 1, + 10, + &serde_json::to_string(&created_value(&run_id, "retry"))?, + ) + .await?; + let first = context.import().await?; + let second = context.import().await?; + + assert_eq!(first.imported_runs, 1); + assert_eq!(second, LegacyRunHistoryImportReport { + scanned_source_runs: 1, + scanned_source_events: 1, + verified_existing_runs: 1, + verified_existing_events: 1, + committed_run_transactions: 1, + ..LegacyRunHistoryImportReport::default() + }); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_retry_and_verification_compare_summary_json_semantically() + -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(11); + let mut created = created_value(&run_id, "labeled"); + created["properties"]["labels"] = serde_json::json!({ + "alpha": "one", + "beta": "two", + "gamma": "three", + }); + context + .put_event(&run_id, 1, 10, &serde_json::to_string(&created)?) + .await?; + context.import().await?; + + let compact: String = sqlx::query_scalar("SELECT summary_json FROM runs WHERE id = ?") + .bind(run_id.to_string()) + .fetch_one(&context.sqlite) + .await?; + let reformatted = + serde_json::to_string_pretty(&serde_json::from_str::(&compact)?)?; + assert_ne!(compact, reformatted); + sqlx::query("UPDATE runs SET summary_json = ? WHERE id = ?") + .bind(reformatted) + .bind(run_id.to_string()) + .execute(&context.sqlite) + .await?; + + let retry = context.import().await?; + assert_eq!(retry.verified_existing_runs, 1); + assert_eq!(retry.verified_existing_events, 1); + let verification = context + .source + .verify_legacy_run_history_in(&context.sqlite) + .await?; + assert_eq!(verification.target_runs, 1); + assert_eq!(verification.target_events, 1); + Ok(()) + } + + #[test] + fn legacy_run_history_rejects_noncanonical_event_key_shapes() -> TestResult<()> { + let run_id = run_id(20); + let event_json = serde_json::to_string(&created_value(&run_id, "key"))?; + let invalid = [ + format!("runs\0{run_id}\0events\0000001-1\0extra").into_bytes(), + b"runs\0not-a-run-id\0events\x00000001-1".to_vec(), + format!("runs\0{run_id}\0events\000001-1").into_bytes(), + format!("runs\0{run_id}\0events\0000000-1").into_bytes(), + format!("runs\0{run_id}\0events\01000000-1").into_bytes(), + format!("runs\0{run_id}\0events\0000001-nope").into_bytes(), + format!("runs\0{run_id}\0events\0000001-01").into_bytes(), + ]; + for key in invalid { + assert!(parse_source_event(&key, event_json.as_bytes()).is_err()); + } + let mut invalid_utf8 = b"runs\0".to_vec(); + invalid_utf8.extend_from_slice(&[0xff, 0xfe]); + invalid_utf8.extend_from_slice(b"\0events\x00000001-1"); + assert!(parse_source_event(&invalid_utf8, event_json.as_bytes()).is_err()); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rejects_invalid_values_sequences_and_replay() -> TestResult<()> { + let invalid_utf8 = TestContext::new().await?; + let first = run_id(30); + invalid_utf8 + .put_raw(keys::run_event_key(&first, 1, 1), &[0xff]) + .await?; + assert!( + invalid_utf8 + .source + .import_legacy_run_history_into(&invalid_utf8.sqlite) + .await + .is_err() + ); + + let invalid_json = TestContext::new().await?; + invalid_json.put_event(&first, 1, 1, "{not-json").await?; + assert!( + invalid_json + .source + .import_legacy_run_history_into(&invalid_json.sqlite) + .await + .is_err() + ); + + let missing_first = TestContext::new().await?; + missing_first + .put_event( + &first, + 2, + 2, + &serde_json::to_string(&submitted_value(&first, 2))?, + ) + .await?; + assert!( + missing_first + .source + .import_legacy_run_history_into(&missing_first.sqlite) + .await + .is_err() + ); + + let wrong_first_event = TestContext::new().await?; + wrong_first_event + .put_event( + &first, + 1, + 1, + &serde_json::to_string(&submitted_value(&first, 1))?, + ) + .await?; + assert!( + wrong_first_event + .source + .import_legacy_run_history_into(&wrong_first_event.sqlite) + .await + .is_err() + ); + + let mismatched_payload = TestContext::new().await?; + let other_run = run_id(31); + mismatched_payload + .put_event( + &first, + 1, + 1, + &serde_json::to_string(&created_value(&other_run, "mismatch"))?, + ) + .await?; + assert!( + mismatched_payload + .source + .import_legacy_run_history_into(&mismatched_payload.sqlite) + .await + .is_err() + ); + + let duplicate = TestContext::new().await?; + let created = serde_json::to_string(&created_value(&first, "duplicate"))?; + duplicate.put_event(&first, 1, 1, &created).await?; + duplicate.put_event(&first, 1, 2, &created).await?; + assert!( + duplicate + .source + .import_legacy_run_history_into(&duplicate.sqlite) + .await + .is_err() + ); + + let unreplayable = TestContext::new().await?; + unreplayable.put_event(&first, 1, 1, &created).await?; + unreplayable.put_event(&first, 2, 2, &created).await?; + assert!( + unreplayable + .source + .import_legacy_run_history_into(&unreplayable.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rolls_back_a_run_when_event_insertion_fails() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(40); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "rollback"))?, + ) + .await?; + sqlx::query( + "CREATE TRIGGER reject_imported_event BEFORE INSERT ON run_events BEGIN SELECT RAISE(FAIL, 'injected'); END", + ) + .execute(&context.sqlite) + .await?; + + let error = context + .source + .import_legacy_run_history_into(&context.sqlite) + .await + .unwrap_err(); + + assert_eq!(error.report().imported_runs, 0); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_interruption_preserves_complete_runs_and_retry_converges() + -> TestResult<()> { + let context = TestContext::new().await?; + let first = run_id(50); + let second = run_id(51); + for run_id in [first, second] { + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "interrupted"))?, + ) + .await?; + } + let controls = ImportControls { + source_after_events: Some(2), + ..ImportControls::default() + }; + + let error = context + .source + .import_legacy_run_history_with_controls(&context.sqlite, &controls) + .await + .unwrap_err(); + assert_eq!(error.report().imported_runs, 1); + assert_eq!(error.report().committed_run_transactions, 1); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 1 + ); + + let retry = context.import().await?; + assert_eq!(retry.imported_runs, 1); + assert_eq!(retry.verified_existing_runs, 1); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 2 + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_cancellation_leaves_no_half_imported_run() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(60); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "cancel"))?, + ) + .await?; + let barrier = Arc::new(Barrier::new(2)); + let source = context.source.clone(); + let sqlite = context.sqlite.clone(); + let task_barrier = Arc::clone(&barrier); + let task = tokio::spawn(async move { + let controls = ImportControls { + after_run_inserted: Some(task_barrier), + ..ImportControls::default() + }; + source + .import_legacy_run_history_with_controls(&sqlite, &controls) + .await + }); + barrier.wait().await; + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + assert_eq!(context.import().await?.imported_runs, 1); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rejects_partial_conflicting_and_corrupt_destinations() + -> TestResult<()> { + let run_id = run_id(70); + let source_created = serde_json::to_string(&created_value(&run_id, "source"))?; + let source_submitted = serde_json::to_string(&submitted_value(&run_id, 2))?; + + let partial = TestContext::new().await?; + partial.put_event(&run_id, 1, 1, &source_created).await?; + partial.put_event(&run_id, 2, 2, &source_submitted).await?; + seed_destination_history(&partial.sqlite, &run_id, &[(1, source_created.clone())]).await?; + assert!( + partial + .source + .import_legacy_run_history_into(&partial.sqlite) + .await + .is_err() + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events") + .fetch_one(&partial.sqlite) + .await?, + 1 + ); + + let conflicting = TestContext::new().await?; + conflicting + .put_event(&run_id, 1, 1, &source_created) + .await?; + let target_created = serde_json::to_string(&created_value(&run_id, "target"))?; + seed_destination_history(&conflicting.sqlite, &run_id, &[(1, target_created)]).await?; + assert!( + conflicting + .source + .import_legacy_run_history_into(&conflicting.sqlite) + .await + .is_err() + ); + + let corrupt_event = TestContext::new().await?; + corrupt_event + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_event.sqlite, &run_id, &[( + 1, + source_created.clone(), + )]) + .await?; + sqlx::query("UPDATE run_events SET event_name = 'run.failed' WHERE run_id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_event.sqlite) + .await?; + assert!( + corrupt_event + .source + .import_legacy_run_history_into(&corrupt_event.sqlite) + .await + .is_err() + ); + + let corrupt_summary = TestContext::new().await?; + corrupt_summary + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_summary.sqlite, &run_id, &[( + 1, + source_created.clone(), + )]) + .await?; + sqlx::query("UPDATE runs SET title = 'corrupt' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_summary.sqlite) + .await?; + assert!( + corrupt_summary + .source + .import_legacy_run_history_into(&corrupt_summary.sqlite) + .await + .is_err() + ); + + let corrupt_head = TestContext::new().await?; + corrupt_head + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_head.sqlite, &run_id, &[(1, source_created)]).await?; + sqlx::query("UPDATE runs SET source_last_seq = 2 WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_head.sqlite) + .await?; + assert!( + corrupt_head + .source + .import_legacy_run_history_into(&corrupt_head.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_verification_accepts_sql_suffixes_and_sql_only_runs() + -> TestResult<()> { + let context = TestContext::new().await?; + let source_run = run_id(80); + let sql_only_run = run_id(81); + let source_created = serde_json::to_string(&created_value(&source_run, "source"))?; + context + .put_event(&source_run, 1, 1, &source_created) + .await?; + context.import().await?; + let suffix = serde_json::to_string(&submitted_value(&source_run, 2))?; + append_destination_event( + &context.sqlite, + &source_run, + &[(1, source_created)], + (2, suffix), + ) + .await?; + let sql_only_created = serde_json::to_string(&created_value(&sql_only_run, "sql-only"))?; + seed_destination_history(&context.sqlite, &sql_only_run, &[(1, sql_only_created)]).await?; + + let report = context + .source + .verify_legacy_run_history_in(&context.sqlite) + .await?; + + assert_eq!(report, LegacyRunHistoryVerificationReport { + source_runs: 1, + source_events: 1, + matched_prefix_runs: 1, + matched_prefix_events: 1, + target_runs: 2, + target_events: 3, + sql_only_runs: 1, + sql_only_events: 1, + ..LegacyRunHistoryVerificationReport::default() + }); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_verification_rejects_invalid_sql_only_histories() -> TestResult<()> + { + let missing_first = TestContext::new().await?; + let missing_first_id = run_id(82); + let created = serde_json::to_string(&created_value(&missing_first_id, "missing-first"))?; + seed_destination_history(&missing_first.sqlite, &missing_first_id, &[(2, created)]).await?; + let error = missing_first + .source + .verify_legacy_run_history_in(&missing_first.sqlite) + .await + .unwrap_err(); + assert!(matches!( + error.failure, + LegacyRunHistoryVerificationFailure::ReplayDestination( + crate::Error::RunEventMismatch { field: "seq", .. } + ) + )); + + let noncanonical = TestContext::new().await?; + let noncanonical_id = run_id(83); + let canonical_json = + serde_json::to_string(&created_value(&noncanonical_id, "noncanonical"))?; + seed_destination_history(&noncanonical.sqlite, &noncanonical_id, &[( + 1, + canonical_json.clone(), + )]) + .await?; + let canonical_id = noncanonical_id.to_string(); + let lowercase_id = canonical_id.to_lowercase(); + assert_ne!(canonical_id, lowercase_id); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ?") + .bind(canonical_json.replace(&canonical_id, &lowercase_id)) + .bind(canonical_id) + .execute(&noncanonical.sqlite) + .await?; + let error = noncanonical + .source + .verify_legacy_run_history_in(&noncanonical.sqlite) + .await + .unwrap_err(); + assert!(matches!( + error.failure, + LegacyRunHistoryVerificationFailure::ReadDestination(crate::Error::RunEventMismatch { + field: "run_id", + .. + }) + )); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_verification_fails_on_prefix_and_current_row_corruption() + -> TestResult<()> { + let run_id = run_id(90); + let created = serde_json::to_string(&created_value(&run_id, "verify"))?; + + let changed_prefix = TestContext::new().await?; + changed_prefix.put_event(&run_id, 1, 1, &created).await?; + let changed = serde_json::to_string(&created_value(&run_id, "changed"))?; + seed_destination_history(&changed_prefix.sqlite, &run_id, &[(1, changed)]).await?; + assert!( + changed_prefix + .source + .verify_legacy_run_history_in(&changed_prefix.sqlite) + .await + .is_err() + ); + + let corrupt_row = TestContext::new().await?; + corrupt_row.put_event(&run_id, 1, 1, &created).await?; + seed_destination_history(&corrupt_row.sqlite, &run_id, &[(1, created)]).await?; + sqlx::query("UPDATE runs SET workflow_slug = 'corrupt' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_row.sqlite) + .await?; + assert!( + corrupt_row + .source + .verify_legacy_run_history_in(&corrupt_row.sqlite) + .await + .is_err() + ); + + let corrupt_json = TestContext::new().await?; + let created = serde_json::to_string(&created_value(&run_id, "verify"))?; + corrupt_json.put_event(&run_id, 1, 1, &created).await?; + seed_destination_history(&corrupt_json.sqlite, &run_id, &[(1, created)]).await?; + sqlx::query("UPDATE runs SET summary_json = '{}' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_json.sqlite) + .await?; + assert!( + corrupt_json + .source + .verify_legacy_run_history_in(&corrupt_json.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_errors_expose_safe_counts_without_event_contents() -> TestResult<()> + { + let context = TestContext::new().await?; + let run_id = run_id(100); + let secret = "DO-NOT-RENDER-THIS-TITLE"; + let mut value = created_value(&run_id, secret); + value + .get_mut("properties") + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("settings"); + context + .put_event(&run_id, 1, 1, &serde_json::to_string(&value)?) + .await?; + + let error = context + .source + .import_legacy_run_history_into(&context.sqlite) + .await + .unwrap_err(); + let rendered = format!("{error:?} {error}"); + let chain = error::collect_chain(&error).join(": "); + assert!(!rendered.contains(secret)); + assert!(!chain.contains(secret)); + assert!(error.source().is_some()); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_scan_ignores_non_event_run_namespaces() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(110); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "event"))?, + ) + .await?; + context + .put_raw( + SlateKey::new("runs").with(run_id).with("state"), + b"not event JSON", + ) + .await?; + assert_eq!(context.import().await?.imported_events, 1); + Ok(()) + } +} diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 3fd34521d..ad957424d 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -8,6 +8,7 @@ mod error; mod keyed_mutex; mod keys; mod legacy_blob_import; +mod legacy_run_history_import; mod record; mod run_sessions; mod run_state; @@ -37,6 +38,10 @@ pub use legacy_blob_import::{ LegacyBlobImportError, LegacyBlobImportReport, LegacyBlobInventory, LegacyBlobInventoryError, LegacyBlobVerificationError, LegacyBlobVerificationReport, }; +pub use legacy_run_history_import::{ + LegacyRunHistoryDiagnostics, LegacyRunHistoryImportError, LegacyRunHistoryImportReport, + LegacyRunHistoryVerificationError, LegacyRunHistoryVerificationReport, +}; pub use run_sessions::{ ProjectedRunSession, project_run_session, project_run_session_with_context, project_run_sessions, diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index c8fc9716e..079c01fc2 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -50,6 +50,31 @@ impl RunProjectionReducer for RunProjection { }; let mut state = projection_from_created(first)?; for event in rest { + // Runs written before the runnable state was introduced can move + // directly from submitted to starting. Replay that historical + // shape through the equivalent current transition while keeping + // live single-event transitions strict. + if matches!(event.event.body, EventBody::RunStarting(_)) + && matches!(state.status, RunStatus::Submitted) + { + state.try_apply_status(RunStatus::Runnable, event.event.ts)?; + } + if let EventBody::RunFailed(props) = &event.event.body { + let failed = RunStatus::Failed { + reason: props.failure.reason, + }; + if !state.status.can_transition_to(failed) { + if matches!( + state.status, + RunStatus::Submitted | RunStatus::Pending { .. } + ) { + state.try_apply_status(RunStatus::Runnable, event.event.ts)?; + } + if matches!(state.status, RunStatus::Runnable) { + state.try_apply_status(RunStatus::Starting, event.event.ts)?; + } + } + } state.apply_event(event)?; } Ok(state) @@ -2664,6 +2689,71 @@ mod tests { } } + fn historical_created_event() -> EventEnvelope { + test_raw_event( + 1, + "run.created", + &json!({ + "settings": WorkflowSettings::default(), + "graph": Graph::new("historical"), + "labels": {}, + "provenance": test_support::test_run_provenance() + }), + None, + ) + } + + #[test] + fn historical_submitted_to_starting_transition_replays() { + let state = RunProjection::apply_events(&[ + historical_created_event(), + test_raw_event(2, "run.submitted", &json!({}), None), + test_raw_event(3, "run.starting", &json!({}), None), + ]) + .unwrap(); + + assert_eq!(state.status, RunStatus::Starting); + } + + #[test] + fn historical_runnable_to_terminated_transition_replays() { + let state = RunProjection::apply_events(&[ + historical_created_event(), + test_raw_event(2, "run.submitted", &json!({}), None), + test_raw_event( + 3, + "run.runnable", + &json!({ "source": "start_requested" }), + None, + ), + test_raw_event( + 4, + "run.failed", + &json!({ + "failure": { + "reason": "terminated", + "detail": { + "message": "worker stopped before startup", + "category": "deterministic" + } + }, + "timing": { + "wall_time_ms": 1, + "inference_time_ms": 0, + "tool_time_ms": 0, + "active_time_ms": 0 + } + }), + None, + ), + ]) + .unwrap(); + + assert_eq!(state.status, RunStatus::Failed { + reason: FailureReason::Terminated, + }); + } + #[test] fn live_run_timing_returns_none_before_run_starts() { let state = initialized_projection(); diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 56ea142a4..ea5e0b134 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -384,6 +384,19 @@ impl RunSummaryStore { connection: &mut SqliteConnection, run_id: &RunId, ) -> Result> { + Ok( + Self::list_events_with_json_on_connection(connection, run_id) + .await? + .into_iter() + .map(|(envelope, _event_json)| envelope) + .collect(), + ) + } + + pub(crate) async fn list_events_with_json_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + ) -> Result> { let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); query .push(" WHERE run_id = ") @@ -405,7 +418,143 @@ impl RunSummaryStore { actual_last_seq, }); } - decode_event_rows(&rows, run_id) + decode_event_rows_with_json(&rows, run_id) + } + + pub(crate) async fn insert_imported_run_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + ) -> Result<()> { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, entry.last_seq)?; + if !(1..=keys::MAX_EVENT_SEQ).contains(&entry.last_seq) { + return Err(Error::RunHeadMismatch { + run_id: entry.run_id.to_string(), + expected_last_seq: entry.last_seq, + actual_last_seq: None, + }); + } + insert_run_on_connection(connection, &record).await + } + + pub(crate) async fn insert_imported_event_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + payload: &EventPayload, + envelope: &EventEnvelope, + event_json: &str, + ) -> Result<()> { + payload.validate(run_id)?; + let decoded = RunEvent::try_from(payload)?; + if envelope.event != decoded || envelope.event.run_id != *run_id { + return Err(run_event_mismatch(run_id, envelope.seq, "event_json")); + } + if !(1..=keys::MAX_EVENT_SEQ).contains(&envelope.seq) { + return Err(run_event_mismatch(run_id, envelope.seq, "seq")); + } + insert_event_json_on_connection(connection, run_id, envelope, event_json).await + } + + pub(crate) async fn verify_current_run_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + ) -> Result<()> { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, entry.last_seq)?; + let row = sqlx::query( + r" +SELECT id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, + status, archived_at_ms, parent_id, title, workflow_slug, workflow_name, + repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions, + input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, + total_usd_micros, summary_json +FROM runs +WHERE id = ? +", + ) + .bind(entry.run_id.to_string()) + .fetch_optional(connection) + .await? + .ok_or_else(|| Error::RunNotFound(entry.run_id.to_string()))?; + + let run = &record.run; + let diff = run.diff.unwrap_or_default(); + verify_run_field(&row, run, "id", &run.id.to_string())?; + verify_run_field(&row, run, "source_last_seq", &i64::from(record.last_seq))?; + verify_run_field( + &row, + run, + "created_at_ms", + &run.timestamps.created_at.timestamp_millis(), + )?; + verify_run_field( + &row, + run, + "started_at_ms", + &run.timestamps + .started_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "last_event_at_ms", + &run.timestamps + .last_event_at + .unwrap_or(run.timestamps.created_at) + .timestamp_millis(), + )?; + verify_run_field( + &row, + run, + "completed_at_ms", + &run.timestamps + .completed_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "status", + &run.lifecycle.status.kind().to_string(), + )?; + verify_run_field( + &row, + run, + "archived_at_ms", + &run.lifecycle + .archived_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "parent_id", + &run.parent_id.map(|value| value.to_string()), + )?; + verify_run_field(&row, run, "title", &run.title)?; + verify_run_field(&row, run, "workflow_slug", &run.workflow.slug)?; + verify_run_field(&row, run, "workflow_name", &record.workflow_name)?; + verify_run_field(&row, run, "repository_name", &record.repository_name)?; + verify_run_field( + &row, + run, + "automation_id", + &run.automation + .as_ref() + .map(|automation| automation.id.clone()), + )?; + verify_run_field(&row, run, "diff_files_changed", &diff.files_changed)?; + verify_run_field(&row, run, "diff_additions", &diff.additions)?; + verify_run_field(&row, run, "diff_deletions", &diff.deletions)?; + verify_run_field(&row, run, "input_tokens", &record.input_tokens)?; + verify_run_field(&row, run, "output_tokens", &record.output_tokens)?; + verify_run_field(&row, run, "reasoning_tokens", &record.reasoning_tokens)?; + verify_run_field(&row, run, "cache_read_tokens", &record.cache_read_tokens)?; + verify_run_field(&row, run, "cache_write_tokens", &record.cache_write_tokens)?; + verify_run_field(&row, run, "total_usd_micros", &record.total_usd_micros)?; + verify_run_json_field(&row, run)?; + Ok(()) } pub(crate) async fn list_events_from_with_limit_on_connection( @@ -625,8 +774,17 @@ async fn insert_event_on_connection( envelope: &EventEnvelope, ) -> Result<()> { let event_json = serde_json::to_string(payload)?; + insert_event_json_on_connection(connection, &record.run.id, envelope, &event_json).await +} + +async fn insert_event_json_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + envelope: &EventEnvelope, + event_json: &str, +) -> Result<()> { sqlx::query(INSERT_EVENT_SQL) - .bind(record.run.id.to_string()) + .bind(run_id.to_string()) .bind(i64::from(envelope.seq)) .bind(envelope.event.event_name()) .bind(envelope.event.node_id.as_deref()) @@ -666,6 +824,47 @@ fn decode_event_rows(rows: &[SqliteRow], run_id: &RunId) -> Result Result> { + let run_id_text = run_id.to_string(); + rows.iter() + .map(|row| { + let event_json: String = row.try_get("event_json")?; + let envelope = decode_event_row(row, run_id, &run_id_text)?; + Ok((envelope, event_json)) + }) + .collect() +} + +fn verify_run_field(row: &SqliteRow, run: &Run, field: &'static str, expected: &T) -> Result<()> +where + T: for<'row> sqlx::Decode<'row, Sqlite> + sqlx::Type + PartialEq, +{ + let stored: T = row.try_get(field)?; + if &stored != expected { + return Err(Error::RunSummaryMismatch { + run_id: run.id.to_string(), + field, + }); + } + Ok(()) +} + +fn verify_run_json_field(row: &SqliteRow, run: &Run) -> Result<()> { + let stored_json: String = row.try_get("summary_json")?; + let stored: serde_json::Value = serde_json::from_str(&stored_json)?; + let expected = serde_json::to_value(run)?; + if stored != expected { + return Err(Error::RunSummaryMismatch { + run_id: run.id.to_string(), + field: "summary_json", + }); + } + Ok(()) +} + fn decode_event_row( row: &SqliteRow, expected_run_id: &RunId, @@ -680,6 +879,15 @@ fn decode_event_row( let event_json: String = row.try_get("event_json")?; let payload: EventPayload = serde_json::from_str(&event_json)?; + if payload + .as_value() + .get("run_id") + .and_then(serde_json::Value::as_str) + != Some(expected_run_id_text) + { + return Err(run_event_mismatch(expected_run_id, seq, "run_id")); + } + payload.validate(expected_run_id)?; let event = RunEvent::try_from(&payload)?; if event.run_id != *expected_run_id { return Err(run_event_mismatch(expected_run_id, seq, "run_id")); diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 39a2129b3..1885b2869 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1621,13 +1621,7 @@ mod tests { .add(&bad_run_id) .await .unwrap(); - let mut run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap(); - let run_settings = run_spec - .get_mut("settings") - .and_then(|settings| settings.get_mut("run")) - .and_then(serde_json::Value::as_object_mut) - .unwrap(); - run_settings.remove("integrations"); + let run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap(); store .put_unvalidated_run_event( &bad_run_id, @@ -1658,8 +1652,8 @@ mod tests { assert_eq!(unreadable[0].run_id, bad_run_id); assert_eq!(unreadable[0].created_at, bad_run_id.created_at()); assert!( - unreadable[0].error.contains("missing field `integrations`"), - "expected missing integrations error, got: {}", + unreadable[0].error.contains("missing field `provenance`"), + "expected missing provenance error, got: {}", unreadable[0].error ); } diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs index 4dc457455..a0450dea6 100644 --- a/lib/foundation/fabro-types/src/run_event/mod.rs +++ b/lib/foundation/fabro-types/src/run_event/mod.rs @@ -20,7 +20,7 @@ pub use session::*; pub use stage::*; pub use todo::*; -use crate::{ParallelBranchId, Principal, RunId, StageId}; +use crate::{ParallelBranchId, Principal, RunId, StageId, UsdMicros}; /// Maximum accepted body size for `POST /runs/{id}/events`. /// @@ -759,7 +759,8 @@ fn is_known_event_name(event: &str) -> bool { } impl RunEvent { - pub fn from_value(value: Value) -> serde_json::Result { + pub fn from_value(mut value: Value) -> serde_json::Result { + normalize_legacy_event(&mut value); let raw: RunEventRaw = serde_json::from_value(value)?; Self::from_parts(RunEventParts { id: raw.id, @@ -808,10 +809,11 @@ impl RunEvent { let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| { ::custom("missing or non-string field: event") })?; - let properties = obj + let mut properties = obj .get("properties") .cloned() .unwrap_or_else(default_properties); + normalize_legacy_event_properties(event, &mut properties); Self::from_parts(RunEventParts { id: id.to_string(), ts, @@ -918,6 +920,291 @@ impl RunEvent { } } +/// Upgrades historical wire shapes only in the value being decoded. Legacy +/// importers still retain and compare the original stored JSON. +fn normalize_legacy_event(value: &mut Value) { + let Some(event) = value + .get("event") + .and_then(Value::as_str) + .map(str::to_owned) + else { + return; + }; + let Some(properties) = value.get_mut("properties") else { + return; + }; + normalize_legacy_event_properties(&event, properties); +} + +fn normalize_legacy_event_properties(event: &str, properties: &mut Value) { + let Some(object) = properties.as_object_mut() else { + return; + }; + match event { + "agent.message" => normalize_legacy_agent_message(object), + "run.completed" => normalize_legacy_timing(object, false), + "run.failed" => { + normalize_legacy_run_failure(object); + normalize_legacy_timing(object, false); + } + "stage.completed" => { + normalize_legacy_usage_field(object); + normalize_legacy_billing_field(object, "billing"); + normalize_legacy_timing(object, true); + } + "stage.failed" => normalize_legacy_billing_field(object, "billing"), + "prompt.completed" => { + normalize_legacy_usage_field(object); + normalize_legacy_billing_field(object, "billing"); + } + "checkpoint.completed" => normalize_legacy_checkpoint_billing(object), + "sandbox.initialized" => normalize_legacy_sandbox_id(object), + _ => {} + } +} + +fn normalize_legacy_agent_message(properties: &mut Map) { + let speed = properties + .get("usage") + .and_then(Value::as_object) + .and_then(|usage| usage.get("speed")) + .and_then(Value::as_str) + .map(str::to_owned); + if let Some(model_id) = properties.get("model").and_then(Value::as_str) { + let mut model = Map::from_iter([ + ( + "provider".to_owned(), + Value::String(legacy_provider_for_model(model_id).to_owned()), + ), + ("model_id".to_owned(), Value::String(model_id.to_owned())), + ]); + if let Some(speed @ ("standard" | "fast")) = speed.as_deref() { + model.insert("speed".to_owned(), Value::String(speed.to_owned())); + } + properties.insert("model".to_owned(), Value::Object(model)); + } + if !properties.contains_key("billing") { + if let Some(usage) = properties.remove("usage") { + properties.insert("billing".to_owned(), usage); + } + } +} + +fn normalize_legacy_usage_field(properties: &mut Map) { + if !properties.contains_key("billing") { + if let Some(usage) = properties.remove("usage") { + properties.insert("billing".to_owned(), usage); + } + } +} + +fn normalize_legacy_billing_field(properties: &mut Map, field: &str) { + if let Some(billing) = properties.get_mut(field) { + normalize_legacy_billing_values(billing); + } +} + +fn normalize_legacy_checkpoint_billing(properties: &mut Map) { + let Some(outcomes) = properties + .get_mut("node_outcomes") + .and_then(Value::as_object_mut) + else { + return; + }; + for outcome in outcomes.values_mut().filter_map(Value::as_object_mut) { + normalize_legacy_billing_field(outcome, "usage"); + } +} + +fn normalize_legacy_timing(properties: &mut Map, stage: bool) { + if properties.contains_key("timing") { + return; + } + let Some(wall_time_ms) = properties.get("duration_ms").and_then(Value::as_u64) else { + return; + }; + let timing = json!({ + "wall_time_ms": wall_time_ms, + "inference_time_ms": 0, + "tool_time_ms": 0, + "active_time_ms": 0, + }); + properties.insert("timing".to_owned(), timing); + if stage { + properties.remove("duration_ms"); + } +} + +fn normalize_legacy_run_failure(properties: &mut Map) { + if !properties.contains_key("failure") { + if let (Some(message), Some(reason)) = ( + properties.get("error").cloned(), + properties.get("reason").cloned(), + ) { + let causes = properties + .get("causes") + .cloned() + .unwrap_or_else(|| Value::Array(Vec::new())); + properties.insert( + "failure".to_owned(), + json!({ + "reason": reason, + "detail": { + "message": message, + "causes": causes, + "category": "deterministic", + } + }), + ); + } + } + if !properties.contains_key("final_git_commit_sha") { + if let Some(commit) = properties.get("git_commit_sha").cloned() { + properties.insert("final_git_commit_sha".to_owned(), commit); + } + } +} + +fn normalize_legacy_sandbox_id(properties: &mut Map) { + if properties.contains_key("id") { + return; + } + let id = properties + .get("identifier") + .and_then(Value::as_str) + .unwrap_or_default(); + properties.insert("id".to_owned(), Value::String(id.to_owned())); +} + +fn normalize_legacy_billing_values(value: &mut Value) { + if legacy_stage_usage(value) { + let legacy = std::mem::take(value); + *value = normalized_legacy_stage_usage(&legacy); + return; + } + match value { + Value::Array(values) => { + for value in values { + normalize_legacy_billing_values(value); + } + } + Value::Object(object) => { + if let Some(facts) = object.get_mut("facts").and_then(Value::as_object_mut) { + if !facts.contains_key("algorithm") { + let provider = facts + .get("provider") + .and_then(Value::as_str) + .map(str::to_owned); + if let Some(provider) = provider { + facts.remove("provider"); + facts.insert( + "algorithm".to_owned(), + Value::String(legacy_billing_algorithm(&provider).to_owned()), + ); + } + } + } + for value in object.values_mut() { + normalize_legacy_billing_values(value); + } + } + _ => {} + } +} + +fn legacy_stage_usage(value: &Value) -> bool { + let Some(object) = value.as_object() else { + return false; + }; + object.get("model").is_some_and(Value::is_string) + && object.get("input_tokens").is_some_and(Value::is_number) + && object.get("output_tokens").is_some_and(Value::is_number) +} + +fn normalized_legacy_stage_usage(legacy: &Value) -> Value { + let object = legacy + .as_object() + .expect("legacy stage usage was validated as an object"); + let model_id = object + .get("model") + .and_then(Value::as_str) + .expect("legacy stage usage was validated with a string model"); + let provider = legacy_provider_for_model(model_id); + let mut model = json!({ + "provider": provider, + "model_id": model_id, + }); + if let Some(speed @ ("standard" | "fast")) = object.get("speed").and_then(Value::as_str) { + model["speed"] = Value::String(speed.to_owned()); + } + let input_tokens = object + .get("input_tokens") + .and_then(Value::as_i64) + .unwrap_or(0); + let output_tokens = object + .get("output_tokens") + .and_then(Value::as_i64) + .unwrap_or(0); + let reasoning_tokens = object + .get("reasoning_tokens") + .and_then(Value::as_i64) + .unwrap_or(0); + let cache_read_tokens = object + .get("cache_read_tokens") + .and_then(Value::as_i64) + .unwrap_or(0); + let cache_write_tokens = object + .get("cache_write_tokens") + .and_then(Value::as_i64) + .unwrap_or(0); + let mut normalized = json!({ + "input": { + "usage": { + "model": model, + "tokens": { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "reasoning_tokens": reasoning_tokens, + "cache_read_tokens": cache_read_tokens, + "cache_write_tokens": cache_write_tokens, + } + }, + "facts": { + "algorithm": legacy_billing_algorithm(provider), + } + } + }); + if let Some(cost) = object.get("cost").and_then(Value::as_f64) { + normalized["total_usd_micros"] = Value::from(UsdMicros::from_usd(cost).0); + } + normalized +} + +fn legacy_provider_for_model(model_id: &str) -> &'static str { + if model_id.starts_with("claude-") { + "anthropic" + } else if model_id.starts_with("gemini-") { + "gemini" + } else if model_id.starts_with("gpt-") + || model_id.starts_with("chatgpt-") + || model_id.starts_with("o1") + || model_id.starts_with("o3") + || model_id.starts_with("o4") + { + "openai" + } else { + "legacy" + } +} + +fn legacy_billing_algorithm(provider: &str) -> &'static str { + match provider { + "anthropic" => "anthropic", + "gemini" => "gemini", + _ => "openai", + } +} + impl Serialize for RunEvent { fn serialize(&self, serializer: S) -> Result where @@ -957,6 +1244,16 @@ mod tests { ) } + fn stored_event(event: &str, properties: &Value) -> Value { + json!({ + "id": format!("evt_{event}"), + "ts": "2026-04-04T12:00:00.000Z", + "run_id": fixtures::RUN_1, + "event": event, + "properties": properties, + }) + } + #[test] fn run_event_round_trips_json() { let event = RunEvent { @@ -1054,6 +1351,212 @@ mod tests { assert_eq!(props.effective_reasoning_effort, None); } + #[test] + fn historical_run_created_defaults_new_run_settings() { + let mut settings = serde_json::to_value(WorkflowSettings::default()).unwrap(); + let run = settings["run"].as_object_mut().unwrap(); + for field in ["clone", "run_branch", "integrations"] { + run.remove(field); + } + let line = stored_event( + "run.created", + &json!({ + "settings": settings, + "graph": Graph::new("test"), + "labels": {}, + "source_directory": "/tmp/run", + "provenance": test_support::test_run_provenance() + }), + ); + + let parsed = RunEvent::from_value(line).unwrap(); + let EventBody::RunCreated(props) = parsed.body else { + panic!("expected run.created"); + }; + + assert_eq!(props.settings.run, WorkflowSettings::default().run); + } + + #[test] + fn historical_agent_message_accepts_string_model() { + let line = stored_event( + "agent.message", + &json!({ + "text": "done", + "model": "gemini-3.1-pro-preview", + "billing": { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15 + }, + "tool_call_count": 0, + "visit": 1 + }), + ); + + let parsed = RunEvent::from_value(line).unwrap(); + let normalized = parsed.to_value().unwrap(); + + assert_eq!(normalized["properties"]["model"]["provider"], "gemini"); + assert_eq!( + normalized["properties"]["model"]["model_id"], + "gemini-3.1-pro-preview" + ); + } + + #[test] + fn historical_stage_usage_and_duration_are_upgraded() { + let line = stored_event( + "stage.completed", + &json!({ + "index": 0, + "duration_ms": 42, + "status": "succeeded", + "usage": { + "model": "claude-sonnet-4-6", + "input_tokens": 100, + "output_tokens": 20, + "cache_read_tokens": 7, + "cache_write_tokens": 3, + "reasoning_tokens": 2, + "speed": "fast", + "cost": 0.012_345 + }, + "attempt": 1, + "max_attempts": 1 + }), + ); + + let parsed = RunEvent::from_value(line).unwrap(); + let normalized = parsed.to_value().unwrap(); + let properties = &normalized["properties"]; + + assert_eq!(properties["timing"]["wall_time_ms"], 42); + assert_eq!( + properties["billing"]["input"]["facts"]["algorithm"], + "anthropic" + ); + assert_eq!( + properties["billing"]["input"]["usage"]["model"]["speed"], + "fast" + ); + assert_eq!(properties["billing"]["total_usd_micros"], 12_345); + assert!(properties.get("duration_ms").is_none()); + assert!(properties.get("usage").is_none()); + } + + #[test] + fn historical_billing_provider_tags_are_upgraded() { + let legacy_billing = json!({ + "input": { + "usage": { + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6" + }, + "tokens": { + "input_tokens": 100, + "output_tokens": 20, + "reasoning_tokens": 0, + "cache_read_tokens": 7, + "cache_write_tokens": 3 + } + }, + "facts": { + "provider": "anthropic", + "cache_write_5m_tokens": 3, + "cache_write_1h_tokens": 0 + } + }, + "total_usd_micros": 123 + }); + let prompt = stored_event( + "prompt.completed", + &json!({ + "response": "done", + "model": "claude-sonnet-4-6", + "provider": "anthropic", + "billing": legacy_billing.clone() + }), + ); + let checkpoint = stored_event( + "checkpoint.completed", + &json!({ + "status": "succeeded", + "current_node": "build", + "node_outcomes": { + "build": { + "status": "succeeded", + "usage": legacy_billing + } + } + }), + ); + + let prompt = RunEvent::from_value(prompt).unwrap().to_value().unwrap(); + let checkpoint = RunEvent::from_value(checkpoint) + .unwrap() + .to_value() + .unwrap(); + + assert_eq!( + prompt["properties"]["billing"]["input"]["facts"]["algorithm"], + "anthropic" + ); + assert_eq!( + checkpoint["properties"]["node_outcomes"]["build"]["usage"]["input"]["facts"]["algorithm"], + "anthropic" + ); + } + + #[test] + fn historical_terminal_and_sandbox_events_are_upgraded() { + let completed = stored_event( + "run.completed", + &json!({ + "duration_ms": 123, + "artifact_count": 0, + "status": "succeeded", + "reason": "completed" + }), + ); + let failed = stored_event( + "run.failed", + &json!({ + "error": "cancelled by user", + "causes": ["interrupt requested"], + "duration_ms": 456, + "reason": "cancelled", + "git_commit_sha": "abc123" + }), + ); + let sandbox = stored_event( + "sandbox.initialized", + &json!({ + "provider": "local", + "working_directory": "/tmp/run" + }), + ); + + let completed = RunEvent::from_value(completed).unwrap().to_value().unwrap(); + let failed = RunEvent::from_value(failed).unwrap().to_value().unwrap(); + let sandbox = RunEvent::from_value(sandbox).unwrap().to_value().unwrap(); + + assert_eq!(completed["properties"]["timing"]["wall_time_ms"], 123); + assert_eq!(failed["properties"]["failure"]["reason"], "cancelled"); + assert_eq!( + failed["properties"]["failure"]["detail"]["message"], + "cancelled by user" + ); + assert_eq!( + failed["properties"]["failure"]["detail"]["causes"], + json!(["interrupt requested"]) + ); + assert_eq!(failed["properties"]["timing"]["wall_time_ms"], 456); + assert_eq!(failed["properties"]["final_git_commit_sha"], "abc123"); + assert_eq!(sandbox["properties"]["id"], ""); + } + #[test] fn run_created_round_trip_preserves_manifest_blob() { let line = json!({ diff --git a/lib/foundation/fabro-types/src/settings/run.rs b/lib/foundation/fabro-types/src/settings/run.rs index 6d2b3ebeb..0d9833a38 100644 --- a/lib/foundation/fabro-types/src/settings/run.rs +++ b/lib/foundation/fabro-types/src/settings/run.rs @@ -22,6 +22,7 @@ use super::size::Size; /// A structurally resolved `[run]` view for consumers. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(default)] pub struct RunNamespace { pub goal: Option, pub working_dir: Option,