diff --git a/cli/src/services/agent_trace_dwh_sync.rs b/cli/src/services/agent_trace_dwh_sync.rs new file mode 100644 index 00000000..b9afed68 --- /dev/null +++ b/cli/src/services/agent_trace_dwh_sync.rs @@ -0,0 +1,1395 @@ +//! The single orchestration boundary composing an `AgentTraceDwhReplica` with +//! the three independent ETL bridges into one sync call. +//! +//! `AgentTraceDwhSync::run()` owns exactly one sequence — open the replica, +//! pull once, run `AgentTraceEtl`, `ConversationEtl`, and `CodeChangesEtl` in +//! that order through their existing `run(repository_id, source, &replica)` +//! APIs unmodified, then push once on full success — behind one bridge-lock- +//! held Turso Sync connection. It extends nothing in `agent_trace_dwh_replica` +//! or any of the three ETL modules, and it deliberately does not resolve +//! credentials, discover paths, or wrap the whole sequence in a global +//! transaction: each ETL still commits its own watermark independently, and a +//! failed push leaves those commits durable in the local replica. + +use std::fmt; + +use crate::services::{ + agent_trace_db::repository::RepositoryAgentTraceDb, + agent_trace_dwh_replica::{ + AgentTraceDwhReplica, AgentTraceDwhReplicaConfig, AgentTraceDwhReplicaError, + }, + agent_trace_etl::{AgentTraceEtl, AgentTraceEtlStats}, + code_changes_etl::{CodeChangesEtl, CodeChangesEtlStats}, + conversation_etl::{ConversationEtl, ConversationEtlStats}, +}; + +/// One combined sync run: `open` → `pull` → `AgentTraceEtl` → +/// `ConversationEtl` → `CodeChangesEtl` → `push`. +/// +/// Configuration reuses each ETL's own defaults/batch sizing; this type adds +/// no configuration of its own beyond composing the three runners. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +#[allow(clippy::struct_field_names)] +pub struct AgentTraceDwhSync { + agent_trace_etl: AgentTraceEtl, + conversation_etl: ConversationEtl, + code_changes_etl: CodeChangesEtl, +} + +impl AgentTraceDwhSync { + /// Open the replica, pull once, run the three ETLs in order, and push + /// once on full success. + /// + /// The bridge lock stays held for the whole sequence: `replica_config` is + /// consumed by exactly one `AgentTraceDwhReplica::open()` call, and the + /// opened replica is dropped (releasing the lock) when this call returns, + /// whether it succeeds or fails. On any stage failure, the sequence stops + /// immediately and `push()` is never invoked; any ETL stage that already + /// committed within this call remains durable in the local replica. + /// + /// `stats.pulled_changes` reflects Turso Sync's own `pull()` semantics: + /// because each call opens a fresh replica connection, the pull that + /// follows any prior session's successful `push()` (including this + /// orchestrator's own immediately preceding `run()`) observes that push + /// as unreconciled and reports `true`, even though the pulled data + /// already matches what is on disk. It settles to `false` only once a + /// `run()` observes no push from any source since the previous `run()`'s + /// own pull. + pub fn run( + &self, + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica_config: AgentTraceDwhReplicaConfig, + ) -> Result { + let replica = AgentTraceDwhReplica::open(replica_config) + .map_err(AgentTraceDwhSyncError::ReplicaOpen)?; + + let pulled_changes = replica.pull().map_err(AgentTraceDwhSyncError::Pull)?; + + let agent_traces = self + .agent_trace_etl + .run(repository_id, source, &replica) + .map_err(AgentTraceDwhSyncError::AgentTraceEtl)?; + + let conversation = self + .conversation_etl + .run(repository_id, source, &replica) + .map_err(AgentTraceDwhSyncError::ConversationEtl)?; + + let code_changes = self + .code_changes_etl + .run(repository_id, source, &replica) + .map_err(AgentTraceDwhSyncError::CodeChangesEtl)?; + + replica.push().map_err(AgentTraceDwhSyncError::Push)?; + + Ok(AgentTraceDwhSyncStats { + pulled_changes, + agent_traces, + conversation, + code_changes, + }) + } +} + +/// Combined stats for one complete `AgentTraceDwhSync::run()` call. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct AgentTraceDwhSyncStats { + /// Whether `pull()` applied any remote changes to the local replica. + pub pulled_changes: bool, + pub agent_traces: AgentTraceEtlStats, + pub conversation: ConversationEtlStats, + pub code_changes: CodeChangesEtlStats, +} + +/// A stage-tagged failure from one `AgentTraceDwhSync::run()` call. +/// +/// Every variant identifies exactly which stage of the sequence failed, so a +/// caller can tell that no stage after it ran. `ReplicaOpen`/`Pull`/`Push` +/// wrap [`AgentTraceDwhReplicaError`], which already redacts the caller's +/// auth token; the three ETL stages wrap `anyhow::Error`, which never +/// observes the token in the first place. +#[derive(Debug)] +pub enum AgentTraceDwhSyncError { + /// Opening the replica (including remote bootstrap and schema + /// classification) failed. + ReplicaOpen(AgentTraceDwhReplicaError), + /// Pulling remote changes into the replica failed. + Pull(AgentTraceDwhReplicaError), + /// The `AgentTraceEtl` stage failed. + AgentTraceEtl(anyhow::Error), + /// The `ConversationEtl` stage failed. + ConversationEtl(anyhow::Error), + /// The `CodeChangesEtl` stage failed. + CodeChangesEtl(anyhow::Error), + /// Pushing local changes to the remote failed after all three ETLs + /// committed locally. + Push(AgentTraceDwhReplicaError), +} + +impl fmt::Display for AgentTraceDwhSyncError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ReplicaOpen(source) => { + write!(f, "agent trace DWH sync replica open failed: {source}") + } + Self::Pull(source) => write!(f, "agent trace DWH sync pull failed: {source}"), + Self::AgentTraceEtl(source) => { + write!( + f, + "agent trace DWH sync agent trace ETL stage failed: {source}" + ) + } + Self::ConversationEtl(source) => { + write!( + f, + "agent trace DWH sync conversation ETL stage failed: {source}" + ) + } + Self::CodeChangesEtl(source) => { + write!( + f, + "agent trace DWH sync code changes ETL stage failed: {source}" + ) + } + Self::Push(source) => write!(f, "agent trace DWH sync push failed: {source}"), + } + } +} + +impl std::error::Error for AgentTraceDwhSyncError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::ReplicaOpen(source) | Self::Pull(source) | Self::Push(source) => Some(source), + Self::AgentTraceEtl(source) + | Self::ConversationEtl(source) + | Self::CodeChangesEtl(source) => Some(source.as_ref()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_composes_default_etl_configuration() { + assert_eq!( + AgentTraceDwhSync::default(), + AgentTraceDwhSync { + agent_trace_etl: AgentTraceEtl::default(), + conversation_etl: ConversationEtl::default(), + code_changes_etl: CodeChangesEtl::default(), + } + ); + } + + #[test] + fn stats_default_to_zeroed_stage_stats_and_no_pulled_changes() { + let stats = AgentTraceDwhSyncStats::default(); + assert!(!stats.pulled_changes); + assert_eq!(stats.agent_traces, AgentTraceEtlStats::default()); + assert_eq!(stats.conversation, ConversationEtlStats::default()); + assert_eq!(stats.code_changes, CodeChangesEtlStats::default()); + } + + #[test] + fn each_error_stage_display_names_its_stage_and_never_contains_the_sentinel_token() { + let sentinel_token = "sentinel-must-never-leak-token"; + let replica_error = || AgentTraceDwhReplicaError::Pull { + message: format!("boom containing {sentinel_token}") + .replace(sentinel_token, ""), + }; + let etl_error = || anyhow::anyhow!("boom, no token involved"); + + let cases: Vec<(AgentTraceDwhSyncError, &str)> = vec![ + ( + AgentTraceDwhSyncError::ReplicaOpen(replica_error()), + "replica open", + ), + (AgentTraceDwhSyncError::Pull(replica_error()), "pull"), + ( + AgentTraceDwhSyncError::AgentTraceEtl(etl_error()), + "agent trace ETL", + ), + ( + AgentTraceDwhSyncError::ConversationEtl(etl_error()), + "conversation ETL", + ), + ( + AgentTraceDwhSyncError::CodeChangesEtl(etl_error()), + "code changes ETL", + ), + (AgentTraceDwhSyncError::Push(replica_error()), "push"), + ]; + + for (error, expected_fragment) in cases { + let display = error.to_string(); + let debug = format!("{error:?}"); + assert!( + display.contains(expected_fragment), + "expected {display:?} to name stage {expected_fragment:?}" + ); + assert!( + !display.contains(sentinel_token), + "Display output must never contain the auth token: {display}" + ); + assert!( + !debug.contains(sentinel_token), + "Debug output must never contain the auth token: {debug}" + ); + } + } +} + +/// Turso Sync integration harness proving AC1/AC2 against a real disposable +/// remote: a fresh empty-remote sync bootstraps the schema, runs all three +/// ETLs, and pushes once; a second run against the same source/remote is a +/// visible no-op. Runs only when a `tursodb` binary supporting +/// `--sync-server` is discoverable on `PATH`, matching the +/// `agent_trace_dwh_replica` integration harness convention exactly. +#[cfg(test)] +mod integration_tests { + use std::{ + net::TcpStream, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, + }; + + use super::*; + use crate::services::agent_trace_db::{ + AgentTraceInsert, DiffTraceInsert, InsertMessageInsert, InsertPartInsert, MessageRole, + PartType, PAYLOAD_TYPE_PATCH, + }; + use crate::services::agent_trace_dwh_db::AgentTraceDwhDb; + + fn unique_path(label: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after Unix epoch") + .as_nanos(); + let dir = std::env::temp_dir().join(format!( + "sce-agent-trace-dwh-sync-integration-{label}-{}-{nonce}", + std::process::id() + )); + std::fs::create_dir_all(&dir).expect("create unique test directory"); + dir + } + + fn clean(dir: &Path) { + let _ = std::fs::remove_dir_all(dir); + } + + fn find_tursodb() -> Option { + let path_var = std::env::var_os("PATH")?; + std::env::split_paths(&path_var).find_map(|dir| { + let candidate = dir.join("tursodb"); + candidate.is_file().then_some(candidate) + }) + } + + struct LocalSyncServer { + child: Child, + url: String, + } + + impl LocalSyncServer { + fn spawn(tursodb_path: &Path) -> Self { + Self::spawn_with_database(tursodb_path, None) + } + + /// Spawn against a specific on-disk `DATABASE` file instead of the + /// default `:memory:`, so this remote's data survives this process + /// being killed and a fresh `tursodb --sync-server` process later + /// starting against the same file. Used only by the push-failure + /// recovery test, which needs to force a real remote outage and then + /// "restart remote availability" without losing what was already + /// pushed before the outage. + fn spawn_persistent(tursodb_path: &Path, db_path: &Path) -> Self { + Self::spawn_with_database(tursodb_path, Some(db_path)) + } + + fn spawn_with_database(tursodb_path: &Path, db_path: Option<&Path>) -> Self { + let port = { + let listener = std::net::TcpListener::bind("127.0.0.1:0") + .expect("bind an ephemeral port to pick a free one for the sync server"); + listener + .local_addr() + .expect("resolve the bound ephemeral port") + .port() + }; + let addr = format!("127.0.0.1:{port}"); + + let mut command = Command::new(tursodb_path); + command.args(["--sync-server", &addr]); + if let Some(db_path) = db_path { + command.arg(db_path); + } + + let child = command + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn tursodb --sync-server"); + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if TcpStream::connect(&addr).is_ok() { + break; + } + assert!( + Instant::now() < deadline, + "tursodb sync server did not become ready in time" + ); + std::thread::sleep(Duration::from_millis(50)); + } + + Self { + child, + url: format!("http://{addr}"), + } + } + + /// Kill this server's process immediately, so a caller can force and + /// observe a real remote-unavailable failure (e.g. from `push()`) + /// before optionally spawning a replacement. + fn kill(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } + + impl Drop for LocalSyncServer { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } + + fn valid_patch(path: &str, added: &str) -> String { + format!( + "diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -1,1 +1,1 @@\n+{added}\n" + ) + } + + /// Seed the source `RepositoryAgentTraceDb` with one row for each of the + /// four tables the three ETLs extract from, so a first `run()` produces + /// non-zero `inserted` counts across every stage. + fn seed_source(source: &RepositoryAgentTraceDb) { + source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-1", + commit_time_ms: 1_000, + trace_json: r#"{"id":"trace-1"}"#, + agent_trace_id: "trace-1", + url: "https://sce.crocoder.dev/agent-trace/trace-1", + remote_url: "https://github.com/acme/widgets", + }) + .expect("agent trace insert should succeed"); + + source + .insert_message(InsertMessageInsert { + session_id: String::from("session-1"), + message_id: String::from("message-1"), + role: MessageRole::User, + generated_at_unix_ms: 1_000, + }) + .expect("message insert should succeed"); + + source + .insert_part(InsertPartInsert { + part_type: PartType::Text, + text: String::from("hello"), + session_id: String::from("session-1"), + message_id: String::from("message-1"), + generated_at_unix_ms: 1_000, + }) + .expect("part insert should succeed"); + + source + .insert_diff_trace(DiffTraceInsert { + time_ms: 1_000, + session_id: "session-1", + patch: &valid_patch("file-1.rs", "added"), + model_id: Some("provider/model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("diff trace insert should succeed"); + } + + #[test] + #[allow(clippy::too_many_lines)] + fn agent_trace_dwh_sync_turso_sync_integration() { + let Some(tursodb_path) = find_tursodb() else { + println!( + "SKIPPING agent_trace_dwh_sync_turso_sync_integration: no `tursodb` binary on \ + PATH. Run `nix develop .#database -c ./scripts/run-cli-cargo.sh test \ + --manifest-path cli/Cargo.toml agent_trace_dwh_sync` to exercise the real \ + Turso Sync harness against the pinned local `tursodb --sync-server`." + ); + return; + }; + + let repository_id = "repo-dwh-sync"; + let sentinel_token = "sentinel-dwh-sync-integration-auth-token-must-not-leak"; + + let server = LocalSyncServer::spawn(&tursodb_path); + + let source_dir = unique_path("source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let replica_dir = unique_path("replica"); + let replica_path = replica_dir.join("agent-trace-sync.db"); + + let sync = AgentTraceDwhSync::default(); + + let first = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("first sync against a truly empty remote should succeed"); + + assert!( + first.agent_traces.inserted > 0, + "first sync should insert agent trace rows: {first:?}" + ); + assert!( + first.conversation.messages.inserted > 0, + "first sync should insert message rows: {first:?}" + ); + assert!( + first.conversation.parts.inserted > 0, + "first sync should insert part rows: {first:?}" + ); + assert!( + first.code_changes.inserted > 0, + "first sync should insert code change rows: {first:?}" + ); + let first_debug = format!("{first:?}"); + assert!( + !first_debug.contains(sentinel_token), + "sync stats must never contain the auth token: {first_debug}" + ); + + let second = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("second sync with no new source rows should succeed as a visible no-op"); + + // Observed real Turso Sync behavior: a freshly opened replica's first + // `pull()` after ANY session (including this orchestrator's own + // immediately preceding `run()`) has pushed reports `pulled_changes + // == true`, because that push was never locally marked "already + // observed" by this new connection — it must pull once to reconcile, + // even though the pulled data exactly matches what is already on + // disk. This is why `second` is not asserted here: AC2's real + // contract is "no new source rows means no new extraction/insertion," + // which holds regardless of this reconciliation pull. `third` below + // proves the `pulled_changes == false` steady state once no push has + // happened since the previous `run()`'s own reconciliation pull. + assert_eq!(second.agent_traces.extracted, 0); + assert_eq!(second.agent_traces.inserted, 0); + assert_eq!(second.conversation.messages.extracted, 0); + assert_eq!(second.conversation.messages.inserted, 0); + assert_eq!(second.conversation.parts.extracted, 0); + assert_eq!(second.conversation.parts.inserted, 0); + assert_eq!(second.code_changes.extracted, 0); + assert_eq!(second.code_changes.inserted, 0); + let second_debug = format!("{second:?}"); + assert!( + !second_debug.contains(sentinel_token), + "sync stats must never contain the auth token: {second_debug}" + ); + + let third = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("third sync with no new source rows should succeed as a visible no-op"); + + assert!( + !third.pulled_changes, + "a run with no push since the previous run's reconciliation pull should observe no \ + pulled changes: {third:?}" + ); + assert_eq!(third.agent_traces.extracted, 0); + assert_eq!(third.agent_traces.inserted, 0); + assert_eq!(third.conversation.messages.extracted, 0); + assert_eq!(third.conversation.messages.inserted, 0); + assert_eq!(third.conversation.parts.extracted, 0); + assert_eq!(third.conversation.parts.inserted, 0); + assert_eq!(third.code_changes.extracted, 0); + assert_eq!(third.code_changes.inserted, 0); + let third_debug = format!("{third:?}"); + assert!( + !third_debug.contains(sentinel_token), + "sync stats must never contain the auth token: {third_debug}" + ); + + clean(&source_dir); + clean(&replica_dir); + drop(server); + } + + /// Count of `agent_traces`/`messages`/`message_parts`/`code_changes` rows + /// for `repository_id` on the DWH surface a replica exposes. + fn count_repository_rows(db: &AgentTraceDwhDb, table: &str, repository_id: &str) -> i64 { + let sql = format!("SELECT COUNT(*) FROM {table} WHERE repository_id = ?1"); + db.query_map(&sql, (repository_id,), |row| { + row.get::(0).map_err(Into::into) + }) + .expect("row count query should succeed") + .into_iter() + .next() + .unwrap_or(0) + } + + /// `[agent_traces, messages, message_parts, code_changes]` row counts for + /// `repository_id`, in that fixed order, matching every assertion below. + fn stage_row_counts(replica: &AgentTraceDwhReplica, repository_id: &str) -> [i64; 4] { + [ + count_repository_rows(replica.db(), "agent_traces", repository_id), + count_repository_rows(replica.db(), "messages", repository_id), + count_repository_rows(replica.db(), "message_parts", repository_id), + count_repository_rows(replica.db(), "code_changes", repository_id), + ] + } + + /// Open a disposable peer replica against the live remote, pull, and read + /// its row counts. Proves what has actually been pushed, independent of + /// whatever any other replica holds locally. + fn remote_row_counts( + server_url: &str, + sentinel_token: &str, + repository_id: &str, + peer_label: &str, + ) -> [i64; 4] { + let peer_dir = unique_path(peer_label); + let replica = AgentTraceDwhReplica::open(AgentTraceDwhReplicaConfig { + local_path: peer_dir.join("agent-trace-sync.db"), + database_url: server_url.to_string(), + auth_token: sentinel_token.to_string(), + }) + .expect("peer replica open for remote inspection should succeed"); + replica.pull().expect("peer pull should succeed"); + let counts = stage_row_counts(&replica, repository_id); + drop(replica); + clean(&peer_dir); + counts + } + + /// Reopen the same local replica path directly to inspect exactly what is + /// durable in the local spool after a failed run. + /// + /// Empirically confirmed (see the observed-behavior note on + /// `AgentTraceDwhSync::run`'s pull-failure test below): once a replica's + /// local schema has classified `Ready`, `AgentTraceDwhReplica::open()` + /// never needs to reach the network, so an unreachable `database_url` + /// here is safe and does not affect what is read. + fn local_row_counts(local_path: &Path, repository_id: &str) -> [i64; 4] { + let replica = AgentTraceDwhReplica::open(AgentTraceDwhReplicaConfig { + local_path: local_path.to_path_buf(), + database_url: String::from("http://127.0.0.1:1"), + auth_token: String::from("unused-for-local-only-inspection"), + }) + .expect( + "reopening an already-Ready local replica for direct spool inspection should \ + succeed without contacting the network", + ); + let counts = stage_row_counts(&replica, repository_id); + drop(replica); + counts + } + + /// Covers AC3(a): opening against an unreachable remote fails with + /// `ReplicaOpen`, the auth token never leaks, and the live remote (never + /// actually addressed by this call) is left with zero rows for this + /// repository. + fn assert_replica_open_failure_stops_before_any_stage( + tursodb_path: &Path, + sentinel_token: &str, + ) { + let server = LocalSyncServer::spawn(tursodb_path); + let repository_id = "repo-stage-open-failure"; + + let source_dir = unique_path("stage-open-source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let unreachable_dir = unique_path("stage-open-replica"); + let sync = AgentTraceDwhSync::default(); + let error = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: unreachable_dir.join("agent-trace-sync.db"), + database_url: String::from("http://127.0.0.1:0"), + auth_token: sentinel_token.to_string(), + }, + ) + .expect_err("open against an unreachable remote should fail"); + assert!( + matches!(error, AgentTraceDwhSyncError::ReplicaOpen(_)), + "expected ReplicaOpen, got {error:?}" + ); + assert!(!error.to_string().contains(sentinel_token)); + + let after_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-open-check", + ); + assert_eq!( + after_remote, + [0, 0, 0, 0], + "a replica-open failure must never reach any ETL or push" + ); + + clean(&source_dir); + clean(&unreachable_dir); + drop(server); + } + + /// Covers AC3(b): once a replica is locally `Ready`, opening it again + /// against an unreachable `database_url` still succeeds (no network + /// access is needed to read an already-synced local schema), but the + /// following `pull()` fails with `Pull`, no ETL runs, and the live + /// remote — addressed only by the earlier, successful baseline run, never + /// by the failing one — is unaffected. + fn assert_pull_failure_stops_before_any_stage(tursodb_path: &Path, sentinel_token: &str) { + let server = LocalSyncServer::spawn(tursodb_path); + let repository_id = "repo-stage-pull-failure"; + let sync = AgentTraceDwhSync::default(); + + let source_dir = unique_path("stage-pull-source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let replica_dir = unique_path("stage-pull-replica"); + let replica_path = replica_dir.join("agent-trace-sync.db"); + sync.run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("baseline sync should succeed and populate the remote"); + + let baseline = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-pull-baseline", + ); + assert_eq!(baseline, [1, 1, 1, 1]); + + // A new source row that would be extracted if the sequence wrongly + // proceeded past the pull failure, so a false no-op can't hide a + // stop-on-failure bypass. + source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-2", + commit_time_ms: 2_000, + trace_json: r#"{"id":"trace-2"}"#, + agent_trace_id: "trace-2", + url: "https://sce.crocoder.dev/agent-trace/trace-2", + remote_url: "https://github.com/acme/widgets", + }) + .expect("second agent trace insert should succeed"); + + let error = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + // Deliberately different from `server.url`: the live + // remote above is never touched by this call, so any + // change there would prove a real stop-on-failure bypass + // rather than an artifact of killing the real server. + database_url: String::from("http://127.0.0.1:1"), + auth_token: sentinel_token.to_string(), + }, + ) + .expect_err("pull against an unreachable remote should fail"); + assert!( + matches!(error, AgentTraceDwhSyncError::Pull(_)), + "expected Pull, got {error:?}" + ); + assert!(!error.to_string().contains(sentinel_token)); + + let after_local = local_row_counts(&replica_path, repository_id); + assert_eq!( + after_local, + [1, 1, 1, 1], + "no ETL should run past a pull failure; the new source row must not appear locally" + ); + + let after_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-pull-after", + ); + assert_eq!( + after_remote, baseline, + "the live remote must be unaffected by a run that failed at pull using a different, \ + unreachable url" + ); + + clean(&source_dir); + clean(&replica_dir); + drop(server); + } + + /// Covers AC3(c): a same-repository, cross-source `agent_trace_id` + /// collision with a different `trace_json` fails `AgentTraceEtl` with an + /// identity-integrity conflict. `AgentTraceEtl` is both the first stage + /// and internally atomic per batch, so the failing run commits nothing + /// locally, and push never runs. + #[allow(clippy::too_many_lines)] + fn assert_agent_trace_etl_failure_stops_before_conversation_code_changes_and_push( + tursodb_path: &Path, + sentinel_token: &str, + ) { + let server = LocalSyncServer::spawn(tursodb_path); + let repository_id = "repo-stage-agent-trace-failure"; + let sync = AgentTraceDwhSync::default(); + + let first_writer_source_dir = unique_path("stage-agent-trace-source-a"); + let first_writer_source = + RepositoryAgentTraceDb::new_at(first_writer_source_dir.join("agent-trace.db")) + .expect("source A DB should open"); + first_writer_source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-a", + commit_time_ms: 1_000, + trace_json: r#"{"id":"conflict","variant":"a"}"#, + agent_trace_id: "trace-conflict", + url: "https://sce.crocoder.dev/agent-trace/trace-conflict", + remote_url: "https://github.com/acme/widgets", + }) + .expect("source A agent trace insert should succeed"); + + let first_writer_replica_dir = unique_path("stage-agent-trace-replica-a"); + sync.run( + repository_id, + &first_writer_source, + AgentTraceDwhReplicaConfig { + local_path: first_writer_replica_dir.join("agent-trace-sync.db"), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("source A's sync should succeed and publish trace-conflict"); + + let baseline = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-agent-trace-baseline", + ); + assert_eq!(baseline, [1, 0, 0, 0]); + + let conflicting_writer_source_dir = unique_path("stage-agent-trace-source-b"); + let conflicting_writer_source = + RepositoryAgentTraceDb::new_at(conflicting_writer_source_dir.join("agent-trace.db")) + .expect("source B DB should open"); + conflicting_writer_source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-b", + commit_time_ms: 2_000, + trace_json: r#"{"id":"conflict","variant":"b"}"#, + agent_trace_id: "trace-conflict", + url: "https://sce.crocoder.dev/agent-trace/trace-conflict", + remote_url: "https://github.com/acme/widgets", + }) + .expect("source B agent trace insert should succeed"); + // Later-stage source rows that would be extracted if the sequence + // wrongly proceeded past the AgentTraceEtl failure. + conflicting_writer_source + .insert_message(InsertMessageInsert { + session_id: String::from("session-b"), + message_id: String::from("message-b"), + role: MessageRole::User, + generated_at_unix_ms: 2_000, + }) + .expect("source B message insert should succeed"); + conflicting_writer_source + .insert_part(InsertPartInsert { + part_type: PartType::Text, + text: String::from("hello from b"), + session_id: String::from("session-b"), + message_id: String::from("message-b"), + generated_at_unix_ms: 2_000, + }) + .expect("source B part insert should succeed"); + conflicting_writer_source + .insert_diff_trace(DiffTraceInsert { + time_ms: 2_000, + session_id: "session-b", + patch: &valid_patch("file-b.rs", "added-b"), + model_id: Some("provider/model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("source B diff trace insert should succeed"); + + let conflicting_writer_replica_dir = unique_path("stage-agent-trace-replica-b"); + let conflicting_writer_replica_path = + conflicting_writer_replica_dir.join("agent-trace-sync.db"); + let error = sync + .run( + repository_id, + &conflicting_writer_source, + AgentTraceDwhReplicaConfig { + local_path: conflicting_writer_replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect_err( + "a conflicting agent_trace_id with a different trace_json should fail the \ + AgentTraceEtl stage", + ); + assert!( + matches!(error, AgentTraceDwhSyncError::AgentTraceEtl(_)), + "expected AgentTraceEtl, got {error:?}" + ); + assert!(!error.to_string().contains(sentinel_token)); + + let after_local = local_row_counts(&conflicting_writer_replica_path, repository_id); + assert_eq!( + after_local, baseline, + "AgentTraceEtl's atomic per-batch failure must leave the local replica exactly as \ + pull() left it (source A's already-published row, pulled before the conflict was \ + detected), with no partial facts from source B and nothing from later stages" + ); + + let after_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-agent-trace-after", + ); + assert_eq!( + after_remote, baseline, + "a failed first-stage run must never push" + ); + + clean(&first_writer_source_dir); + clean(&first_writer_replica_dir); + clean(&conflicting_writer_source_dir); + clean(&conflicting_writer_replica_dir); + drop(server); + } + + /// Covers AC3(d): an invalid `parts.type` value (no source-side CHECK + /// constraint permits inserting it through raw SQL) fails the parts half + /// of `ConversationEtl`. The messages half commits independently within + /// the same `ConversationEtl::run()` call, and `AgentTraceEtl` (run + /// before it) commits too; `CodeChangesEtl` never runs and push never + /// runs. + #[allow(clippy::too_many_lines)] + fn assert_conversation_etl_failure_leaves_agent_trace_committed_and_stops_before_code_changes_and_push( + tursodb_path: &Path, + sentinel_token: &str, + ) { + let server = LocalSyncServer::spawn(tursodb_path); + let repository_id = "repo-stage-conversation-failure"; + let sync = AgentTraceDwhSync::default(); + + let source_dir = unique_path("stage-conversation-source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let replica_dir = unique_path("stage-conversation-replica"); + let replica_path = replica_dir.join("agent-trace-sync.db"); + sync.run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("baseline sync should succeed"); + + let baseline = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-conversation-baseline", + ); + assert_eq!(baseline, [1, 1, 1, 1]); + + source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-2", + commit_time_ms: 2_000, + trace_json: r#"{"id":"trace-2"}"#, + agent_trace_id: "trace-2", + url: "https://sce.crocoder.dev/agent-trace/trace-2", + remote_url: "https://github.com/acme/widgets", + }) + .expect("second agent trace insert should succeed"); + source + .insert_message(InsertMessageInsert { + session_id: String::from("session-1"), + message_id: String::from("message-2"), + role: MessageRole::Assistant, + generated_at_unix_ms: 2_000, + }) + .expect("second message insert should succeed"); + source + .execute( + "INSERT INTO parts (type, text, message_id, session_id, generated_at_unix_ms) \ + VALUES (?1, ?2, ?3, ?4, ?5)", + ( + "bogus-type", + "invalid part", + "message-2", + "session-1", + 2_000_i64, + ), + ) + .expect("raw invalid parts insert should succeed: parts.type has no CHECK constraint"); + // A later-stage source row that would be extracted if the sequence + // wrongly proceeded past the ConversationEtl failure. + source + .insert_diff_trace(DiffTraceInsert { + time_ms: 2_000, + session_id: "session-1", + patch: &valid_patch("file-2.rs", "added-2"), + model_id: Some("provider/model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("second diff trace insert should succeed"); + + let error = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect_err("an invalid parts.type value should fail the ConversationEtl stage"); + assert!( + matches!(error, AgentTraceDwhSyncError::ConversationEtl(_)), + "expected ConversationEtl, got {error:?}" + ); + assert!(!error.to_string().contains(sentinel_token)); + + let after_local = local_row_counts(&replica_path, repository_id); + assert_eq!( + after_local, + [2, 2, 1, 1], + "AgentTraceEtl and the messages half of ConversationEtl should commit locally even \ + though parts failed and CodeChangesEtl never ran" + ); + + let after_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-conversation-after", + ); + assert_eq!( + after_remote, baseline, + "a ConversationEtl failure must never push" + ); + + clean(&source_dir); + clean(&replica_dir); + drop(server); + } + + /// Covers AC3(e): a malformed `diff_traces.patch` payload fails + /// `CodeChangesEtl`'s strict transformation. Both prior ETLs commit + /// locally within the same run, and push never runs. + #[allow(clippy::too_many_lines)] + fn assert_code_changes_etl_failure_leaves_prior_etls_committed_and_stops_before_push( + tursodb_path: &Path, + sentinel_token: &str, + ) { + let server = LocalSyncServer::spawn(tursodb_path); + let repository_id = "repo-stage-code-changes-failure"; + let sync = AgentTraceDwhSync::default(); + + let source_dir = unique_path("stage-code-changes-source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let replica_dir = unique_path("stage-code-changes-replica"); + let replica_path = replica_dir.join("agent-trace-sync.db"); + sync.run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("baseline sync should succeed"); + + let baseline = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-code-changes-baseline", + ); + assert_eq!(baseline, [1, 1, 1, 1]); + + source + .insert_agent_trace(AgentTraceInsert { + commit_id: "commit-2", + commit_time_ms: 2_000, + trace_json: r#"{"id":"trace-2"}"#, + agent_trace_id: "trace-2", + url: "https://sce.crocoder.dev/agent-trace/trace-2", + remote_url: "https://github.com/acme/widgets", + }) + .expect("second agent trace insert should succeed"); + source + .insert_message(InsertMessageInsert { + session_id: String::from("session-1"), + message_id: String::from("message-2"), + role: MessageRole::Assistant, + generated_at_unix_ms: 2_000, + }) + .expect("second message insert should succeed"); + source + .insert_part(InsertPartInsert { + part_type: PartType::Text, + text: String::from("second part"), + session_id: String::from("session-1"), + message_id: String::from("message-2"), + generated_at_unix_ms: 2_000, + }) + .expect("second part insert should succeed"); + source + .insert_diff_trace(DiffTraceInsert { + time_ms: 2_000, + session_id: "session-1", + patch: "Index: notes/malformed.md\n===================================================================\n--- notes/malformed.md\n+++ notes/malformed.md\n@@ malformed @@\n+bad\n", + model_id: None, + tool_name: "opencode", + tool_version: None, + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect( + "malformed diff trace insert should succeed at the source layer; strict \ + validation happens during ETL transformation", + ); + + let error = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect_err( + "a malformed diff_traces.patch payload should fail the CodeChangesEtl stage", + ); + assert!( + matches!(error, AgentTraceDwhSyncError::CodeChangesEtl(_)), + "expected CodeChangesEtl, got {error:?}" + ); + assert!(!error.to_string().contains(sentinel_token)); + + let after_local = local_row_counts(&replica_path, repository_id); + assert_eq!( + after_local, + [2, 2, 2, 1], + "AgentTraceEtl and ConversationEtl should commit locally even though CodeChangesEtl \ + failed and push never ran" + ); + + let after_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "stage-code-changes-after", + ); + assert_eq!( + after_remote, baseline, + "a CodeChangesEtl failure must never push" + ); + + clean(&source_dir); + clean(&replica_dir); + drop(server); + } + + /// Proves AC3: each of the five sequence stages, when it fails, is + /// identifiable through a distinct `AgentTraceDwhSyncError` stage + /// variant, and none of them ever reaches `push()`. + #[test] + fn agent_trace_dwh_sync_stage_failure_turso_sync_integration() { + let Some(tursodb_path) = find_tursodb() else { + println!( + "SKIPPING agent_trace_dwh_sync_stage_failure_turso_sync_integration: no \ + `tursodb` binary on PATH. Run `nix develop .#database -c \ + ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml \ + agent_trace_dwh_sync` to exercise the real Turso Sync harness against the \ + pinned local `tursodb --sync-server`." + ); + return; + }; + + let sentinel_token = "sentinel-dwh-sync-stage-failure-token-must-not-leak"; + + assert_replica_open_failure_stops_before_any_stage(&tursodb_path, sentinel_token); + assert_pull_failure_stops_before_any_stage(&tursodb_path, sentinel_token); + assert_agent_trace_etl_failure_stops_before_conversation_code_changes_and_push( + &tursodb_path, + sentinel_token, + ); + assert_conversation_etl_failure_leaves_agent_trace_committed_and_stops_before_code_changes_and_push( + &tursodb_path, + sentinel_token, + ); + assert_code_changes_etl_failure_leaves_prior_etls_committed_and_stops_before_push( + &tursodb_path, + sentinel_token, + ); + } + + /// Proves AC4: a push failure that happens after all three ETLs have + /// committed locally leaves those commits durable in the local spool; a + /// subsequent successful run reaches the remote with every previously + /// committed fact and watermark, with no lost rows and no duplicated + /// logical rows. + /// + /// This also exercises the exact scenario the plan calls out as needing + /// empirical discovery: the recovery run's first step, `pull()`, runs + /// against a replica that itself holds pending local commits (the three + /// ETL stages that committed locally during the failed-push run, never + /// pushed). `AgentTraceDwhSync::run()`'s own open→pull→ETLs→push ordering + /// composes this scenario for free — the recovery run is a plain second + /// `AgentTraceDwhSync::run()` call, not special-cased in any way — so + /// nothing about `run()`'s internal sequencing needs to change: + /// + /// Observed real Turso Sync behavior: `pull()` against a replica holding + /// pending local commits does not discard, corrupt, or roll back those + /// commits. It applies remote changes (here, none beyond the schema + /// already published before the outage) without touching the pending + /// local writes, exactly like the empty-remote/no-op-run behavior already + /// observed in `agent_trace_dwh_sync_turso_sync_integration`. The + /// recovery run's three ETL stages then see the source rows already + /// extracted and correctly report a no-op (zero `extracted`/`inserted` + /// everywhere), and the final `push()` succeeds, publishing exactly the + /// facts committed during the failed run — no duplication, because + /// nothing was ever re-extracted, and no loss, because every commit + /// survived the outage in the local spool. + #[test] + #[allow(clippy::too_many_lines)] + fn agent_trace_dwh_sync_push_failure_recovery_turso_sync_integration() { + let Some(tursodb_path) = find_tursodb() else { + println!( + "SKIPPING agent_trace_dwh_sync_push_failure_recovery_turso_sync_integration: no \ + `tursodb` binary on PATH. Run `nix develop .#database -c \ + ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml \ + agent_trace_dwh_sync` to exercise the real Turso Sync harness against the \ + pinned local `tursodb --sync-server`." + ); + return; + }; + + let repository_id = "repo-push-failure-recovery"; + let sentinel_token = "sentinel-dwh-sync-push-failure-recovery-token-must-not-leak"; + + let remote_dir = unique_path("push-failure-remote"); + let remote_db_path = remote_dir.join("remote.db"); + let mut server = LocalSyncServer::spawn_persistent(&tursodb_path, &remote_db_path); + + let source_dir = unique_path("push-failure-source"); + let source = RepositoryAgentTraceDb::new_at(source_dir.join("agent-trace.db")) + .expect("source DB should open"); + seed_source(&source); + + let replica_dir = unique_path("push-failure-replica"); + let replica_path = replica_dir.join("agent-trace-sync.db"); + + // Manually reproduce AgentTraceDwhSync::run()'s own + // open→pull→AgentTraceEtl→ConversationEtl→CodeChangesEtl sequence, so + // the test can force a deterministic push failure at the exact point + // `run()` itself would call push() — instead of racing a + // background kill against an opaque `run()` call. + let replica = AgentTraceDwhReplica::open(AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }) + .expect("replica open against the live remote should succeed and publish the schema"); + + replica + .pull() + .expect("pull against the freshly schema-initialized remote should succeed"); + + let agent_traces = AgentTraceEtl::default() + .run(repository_id, &source, &replica) + .expect("agent trace ETL should commit locally"); + let conversation = ConversationEtl::default() + .run(repository_id, &source, &replica) + .expect("conversation ETL should commit locally"); + let code_changes = CodeChangesEtl::default() + .run(repository_id, &source, &replica) + .expect("code changes ETL should commit locally"); + assert!(agent_traces.inserted > 0, "{agent_traces:?}"); + assert!(conversation.messages.inserted > 0, "{conversation:?}"); + assert!(conversation.parts.inserted > 0, "{conversation:?}"); + assert!(code_changes.inserted > 0, "{code_changes:?}"); + + // Force the push to fail deterministically: the remote is gone by + // the time push() runs. + server.kill(); + + let push_error = replica + .push() + .expect_err("push against a killed remote should fail"); + assert!(!push_error.to_string().contains(sentinel_token)); + + // Release the bridge lock so the recovery run below can reopen this + // same local replica path. + drop(replica); + + let after_failed_push = local_row_counts(&replica_path, repository_id); + assert_eq!( + after_failed_push, + [1, 1, 1, 1], + "all three ETL commits must remain durable in the local spool after a failed push" + ); + + // Restart remote availability: a fresh process on a new ephemeral + // port, backed by the same on-disk DATABASE file, so the schema + // published before the outage is still there and nothing else is. + let server = LocalSyncServer::spawn_persistent(&tursodb_path, &remote_db_path); + + let sync = AgentTraceDwhSync::default(); + let recovery = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect( + "the recovery run's pull() against a replica holding pending local commits, \ + followed by three no-op ETL stages and a push, should converge", + ); + assert_eq!( + recovery.agent_traces.extracted, 0, + "the recovery run must not re-extract rows already committed before the outage" + ); + assert_eq!(recovery.agent_traces.inserted, 0); + assert_eq!(recovery.conversation.messages.extracted, 0); + assert_eq!(recovery.conversation.messages.inserted, 0); + assert_eq!(recovery.conversation.parts.extracted, 0); + assert_eq!(recovery.conversation.parts.inserted, 0); + assert_eq!(recovery.code_changes.extracted, 0); + assert_eq!(recovery.code_changes.inserted, 0); + let recovery_debug = format!("{recovery:?}"); + assert!(!recovery_debug.contains(sentinel_token)); + + let after_recovery_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "push-failure-after-recovery", + ); + assert_eq!( + after_recovery_remote, + [1, 1, 1, 1], + "the recovery push must reach the remote with every fact committed during the \ + failed-push run, no lost rows and no duplicated logical rows" + ); + + // A further run from the same replica must be a stable no-op: no + // re-extraction, no re-push, no unbounded growth. + let stable = sync + .run( + repository_id, + &source, + AgentTraceDwhReplicaConfig { + local_path: replica_path.clone(), + database_url: server.url.clone(), + auth_token: sentinel_token.to_string(), + }, + ) + .expect("a further run with no new source rows should succeed as a stable no-op"); + assert_eq!(stable.agent_traces.extracted, 0); + assert_eq!(stable.agent_traces.inserted, 0); + assert_eq!(stable.conversation.messages.extracted, 0); + assert_eq!(stable.conversation.messages.inserted, 0); + assert_eq!(stable.conversation.parts.extracted, 0); + assert_eq!(stable.conversation.parts.inserted, 0); + assert_eq!(stable.code_changes.extracted, 0); + assert_eq!(stable.code_changes.inserted, 0); + + let after_stable_remote = remote_row_counts( + &server.url, + sentinel_token, + repository_id, + "push-failure-after-stable", + ); + assert_eq!( + after_stable_remote, + [1, 1, 1, 1], + "a stable no-op run must not change the remote's row counts" + ); + + clean(&remote_dir); + clean(&source_dir); + clean(&replica_dir); + drop(server); + } +} diff --git a/cli/src/services/mod.rs b/cli/src/services/mod.rs index 2358bcf4..2be9a6d0 100644 --- a/cli/src/services/mod.rs +++ b/cli/src/services/mod.rs @@ -5,6 +5,8 @@ pub mod agent_trace_dwh_db; #[allow(dead_code)] pub mod agent_trace_dwh_replica; #[allow(dead_code)] +pub mod agent_trace_dwh_sync; +#[allow(dead_code)] pub mod agent_trace_etl; #[allow(dead_code)] pub mod agent_trace_storage; diff --git a/context/context-map.md b/context/context-map.md index 79aada33..0db1d978 100644 --- a/context/context-map.md +++ b/context/context-map.md @@ -63,6 +63,7 @@ Feature/domain context: - `context/sce/auth-db.md` (encrypted `AuthDb = EncryptedTursoDb` adapter, canonical `/sce/auth.db` path, build-time generated `AUTH_MIGRATIONS` from `cli/migrations/auth/`, auth credential schema and updated-at trigger baseline, lifecycle setup/doctor integration, encrypted token-storage persistence, and `SCE_AUTH_DB_ENCRYPTION_KEY`/OS credential-store key handling) - `context/sce/agent-trace-db.md` (implemented Agent Trace database adapter: the sole repository-scoped `RepositoryAgentTraceDb` backed by one fresh multi-statement baseline schema file plus an additive `002_repository_source_instance_id` migration, typed `RepositoryMetadata { repository_id, source_instance_id }` with atomic once-only source-instance initialization, `repository_metadata` validation, narrow concurrent-first-open repair for missing one-file baseline migration metadata after all required schema tables exist, no trace-table `checkout_id` columns, repository-level typed insert helpers for diff traces, post-commit intersections, Agent Trace rows, messages, and parts, repository-level recent diff-trace reads without checkout filtering, on-demand command/hook initialization with no daemon/background service, and the never-touch on-disk boundary for any pre-migration checkout-scoped/global DB files; the checkout-scoped `AgentTraceDb` adapter, its `agent_trace_db_path()`/`agent_trace_db_path_for_checkout()` helpers, and the 15-file `cli/migrations/agent-trace/` chain were removed by the `retire-legacy-agent-trace-db` plan; active hook writers/readers and Agent Trace setup/lifecycle resolve repository storage through `agent_trace_storage`) - `context/sce/agent-trace-dwh-replica.md` (Agent Trace DWH Turso Sync replica boundary: `AgentTraceDwhReplica` in `cli/src/services/agent_trace_dwh_replica/replica.rs`, the sole owner of a Turso Sync connection to the repository-scoped `agent-trace-sync.db`; acquires the `BridgeLock` before any Turso access, opens via `turso::sync::Builder` without enabling multiprocess WAL, then classifies the DWH schema via `AgentTraceDwhDb::classify_schema_state()` — a `Ready` schema is left untouched, a genuinely `Empty` schema is initialized locally with `run_migrations()` and published with `push()`, and an `Incompatible` schema fails loudly without repair — exposes lock-lifetime-bound `run_agent_trace_etl()` and `run_code_changes_etl()` plus explicit `pull()`/`push()`, redacts the caller-supplied auth token from every error, and reuses a new `TursoDb::from_connection`/`block_on` seam; credential discovery/persistence and CLI/lifecycle wiring remain deferred) +- `context/sce/agent-trace-dwh-sync.md` (`AgentTraceDwhSync` orchestration boundary in `cli/src/services/agent_trace_dwh_sync.rs` composing `AgentTraceDwhReplica` with `AgentTraceEtl`, `ConversationEtl`, and `CodeChangesEtl` into one `run()` call — open replica, pull once, run the three ETLs in order, push once on full success — returning combined `AgentTraceDwhSyncStats` and a stage-tagged `AgentTraceDwhSyncError`; documents the observed Turso Sync `pulled_changes` reconciliation-echo behavior after any fresh open following a push; core service and empty-remote/no-op proof only so far, full failure/recovery/convergence semantics land with later `agent-trace-dwh-sync` plan tasks) - `context/sce/agent-trace-dwh-db.md` (Agent Trace DWH: a separate append-oriented destination schema for the CLI-independent ETL consumer, distinct from the repository-scoped source schema above. `AgentTraceDwhDb = TursoDb` in `cli/src/services/agent_trace_dwh_db/mod.rs`, explicit-path only, no lifecycle/CLI wiring yet. One fresh baseline `cli/migrations/agent-trace-dwh/001_dwh_schema.sql` creates `repositories`, `source_instances`, `etl_watermarks`, `messages`, `message_parts`, `agent_traces`, and `code_changes` with no foreign keys; every fact table denormalizes `repository_id`/`source_instance_id` lineage as plain text. Deterministic logical identities (messages, Agent Traces) are unique excluding `source_instance_id` for cross-source-database idempotency; raw local source row IDs (`source_part_id`, `source_diff_trace_id`) are unique per source instance so the same local integer coexists across sources/repositories) - `context/sce/agent-trace-etl.md` (shared Agent Trace ETL mechanics and the `agent_traces`, `messages`, `parts`, and `code_changes` bridges between the `agent-trace.db` source and DWH replica; covers bounded extraction, exact-content transformation/hashing, table-specific identity validation, atomic per-lineage watermark advancement, source contention handling, stats, replica-owned orchestration, and the code-change session-only relationship) - `context/sce/code-changes-etl.md` (CLI-independent `CodeChangesEtl` contract for the exact ordered `diff_traces` projection, short source snapshots, strict `patch`/`structured` normalization, source-lineage watermarks, exact payload hashing, checked patch metrics, atomic replay/conflict handling, replica-owned execution, and the session-only relationship to DWH conversations without message-level causality) diff --git a/context/plans/agent-trace-dwh-sync.md b/context/plans/agent-trace-dwh-sync.md new file mode 100644 index 00000000..4abcb5de --- /dev/null +++ b/context/plans/agent-trace-dwh-sync.md @@ -0,0 +1,544 @@ +# Plan: agent-trace-dwh-sync + +## Change summary + +Add `AgentTraceDwhSync`, the single orchestration service that connects the +already-implemented pieces: `AgentTraceDwhReplica` (PR #189, single-owner Turso +Sync replica lifecycle for `agent-trace-sync.db`), `AgentTraceEtl` (PR #190/#191), +`ConversationEtl` (PR #191), and `CodeChangesEtl` (PR #192). Today each of those +exists and is independently tested, but nothing yet drives them together: a +caller would have to hand-sequence `AgentTraceDwhReplica::open()`, `pull()`, +three separate `etl.run(repository_id, source, &replica)` calls, and `push()` +themselves, with no combined stats or stage-identified error type. + +This plan adds one new service, `cli/src/services/agent_trace_dwh_sync.rs`, +whose `run()` owns exactly that sequence — open → pull → `AgentTraceEtl` → +`ConversationEtl` → `CodeChangesEtl` → push — behind one bridge-lock-held +Turso Sync connection, returning one combined stats type and a stage-tagged +error. It extends nothing in `agent_trace_dwh_replica`, `agent_trace_etl`, +`conversation_etl`, or `code_changes_etl`: all three ETLs already expose +`run(repository_id, &RepositoryAgentTraceDb, &AgentTraceDwhReplica)`, which is +exactly the shape this orchestrator needs to call unmodified. + +The plan also proves, empirically against the real local Turso Sync harness +already established in `agent_trace_dwh_replica`'s `integration_tests` module, +that the local sync spool survives interruption at every stage (replica-open +failure, pull failure, each ETL failure, push failure) without losing source +rows, duplicating facts, or skipping watermarks — and documents whatever the +real Turso Sync SDK is observed to do when `pull()` runs against a replica +holding committed-but-unpushed ETL changes, since the request is explicit that +this observed behavior should override the proposed pull-before-ETL ordering +if it turns out to be unsafe. + +No CLI wiring, credential discovery, or `sce trace sync` command is added — +this plan makes exactly one Rust API, so that a future thin CLI adapter has +nothing left to design. + +## Acceptance criteria + +- [ ] AC1: A fresh sync against a genuinely empty remote succeeds in one + `AgentTraceDwhSync::run()` call: the remote DWH schema is initialized via + `AgentTraceDwhReplica::open()`'s existing empty-remote path, all three ETLs + run in order, and non-zero stats are returned for every table with source + rows. + - Validate: `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` +- [ ] AC2: A second `run()` against the same source and remote, with no new + source rows and no remote-side changes, succeeds and returns stats showing + zero extracted/inserted rows in every table (a visible no-op, not an error). + - Validate: same integration test, no-op-run assertion +- [ ] AC3: Replica-open failure, pull failure, and each of the three ETL + failures are each identifiable through a distinct `AgentTraceDwhSyncError` + stage variant, and each one leaves the final `push()` uninvoked. + - Validate: integration tests covering each failure stage (T02) +- [ ] AC4: A push failure that occurs after all three ETLs have committed + locally leaves those commits durable in the local `agent-trace-sync.db` + spool; the sync call returns an error; and a subsequent successful `run()` + reaches the remote with the previously committed facts and watermarks, with + no lost rows, no duplicated logical rows, and no skipped watermarks. + - Validate: integration test proving push-failure recovery (T03) +- [ ] AC5: Deleting the local `agent-trace-sync.db` after a successful sync and + running `AgentTraceDwhSync::run()` again reconstructs the replica from the + remote and performs only genuinely incremental ETL work (a no-op when no new + source rows exist since the deleted replica's last push). + - Validate: integration test proving fresh-replica reconstruction (T04) +- [ ] AC6: Two different `repository_id`s, each with its own source DB and + local replica path, syncing against the same remote DWH both appear in the + remote afterward, and neither sync corrupts or removes the other's rows. + - Validate: integration test proving multi-repository convergence (T05) +- [ ] AC7: Two source instances of the same `repository_id` syncing against + the same remote maintain independent per-source-instance watermarks, and + their overlapping local row IDs (parts, diff traces) do not collide in the + DWH. + - Validate: integration test proving multi-source-instance independence (T06) +- [ ] AC8: Two independently operated sync clients against the same remote + DWH converge: client A observes client B's remote additions after its own + `pull()`, neither destroys the other's committed facts, and repeated runs + from both sides stabilize (no unbounded growth in inserted counts once both + sides are current). + - Validate: integration test proving cross-client convergence (T07) +- [ ] AC9: No `AgentTraceDwhSyncError` variant, `Debug`/`Display` output, or + `AgentTraceDwhSyncStats` value ever contains the caller-supplied auth token, + including on a push/pull failure against a real remote. + - Validate: covered by the sentinel-auth-token assertions embedded in the + T02–T04 integration tests, matching the existing `redact_token` pattern in + `agent_trace_dwh_replica/replica.rs` + +### Full validation + +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml` +- `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml` +- `nix develop -c ./scripts/run-cli-cargo.sh clippy --manifest-path cli/Cargo.toml --all-targets -- -D warnings` +- `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path cli/Cargo.toml -- --check` +- `nix flake check` +- `nix run .#pkl-check-generated` + +### Context sync + +- `context/sce/agent-trace-dwh-sync.md` (new): the full sync lifecycle, + ownership, failure/recovery semantics, and the observed pull-with-pending- + local-changes Turso behavior. +- `context/context-map.md`: register the new domain context file. +- `context/glossary.md`: add an `AgentTraceDwhSync` entry; extend the existing + `Agent Trace DWH sync replica` entry's cross-links. +- `context/sce/agent-trace-dwh-replica.md`: note that `AgentTraceDwhSync` is + now the orchestration boundary that composes `run_agent_trace_etl()`/ + `run_code_changes_etl()`/`ConversationEtl::run()`/`pull()`/`push()`, without + changing anything about the replica's own ownership contract. + +## Constraints and non-goals + +- **In scope:** one new `cli/src/services/agent_trace_dwh_sync.rs` module + (struct, config reuse, stats, stage-tagged error, `run()`), its unit and + integration tests, and the context-sync files listed above. +- **Out of scope:** `AgentTraceDwhReplica`, `AgentTraceEtl`, `ConversationEtl`, + `CodeChangesEtl`, and their existing tests — call them through their current + public APIs unmodified unless a genuine correctness problem is found while + building this orchestrator (none is anticipated; the three `run()` methods + already take `&AgentTraceDwhReplica` directly). +- **Constraints:** exactly one Turso Sync connection per `run()` invocation; + the bridge lock stays held for the whole pull+ETLs+push sequence; no global + transaction wraps the three ETLs; auth tokens must never appear in errors, + `Debug`, `Display`, stats, or logs; reuse the existing `LocalSyncServer` + Turso Sync integration harness pattern rather than a fake replication + implementation. +- **Non-goals:** control-plane calls, workspace DWH provisioning, WorkOS auth, + token refresh/persistence, `sce trace sync` CLI wiring, scheduled/background + sync, automatic retry loops around the whole sync operation, new ETL tables, + post-commit intersection ETL, reverse remote-to-source hydration, schema + ownership changes, automatic DWH schema upgrades, analytics/query APIs, UI, + and any new ADR/decision record (context sync in this plan is limited to + current-state `context/sce/*.md`/glossary/context-map prose; a decision + record, if warranted, is a separate later call for `/validate`'s + context-synchronization gate or an explicit `/decision` invocation, not this + plan). + +## Assumptions + +- The branch already contains all work through PR #192 (`etl-code-change`) — + confirmed by `git log` on the current `etl-orchestrator` branch, which is + built directly on top of it. No rebase or branch change is needed before + starting T01. +- Item 8 of the request ("explicitly test pull with pending local changes") is + treated as an empirical discovery task (T03), not a pre-decided design + choice: T01 implements the literal open → pull → ETLs → push order the + request proposes, and T03 is authorized to adjust that internal ordering — + documenting exactly why — if the real local Turso Sync harness demonstrates + that ordering is unsafe. This mirrors the request's own instruction that the + observed-behavior test outranks the proposed sequence. +- New integration tests follow the existing `agent_trace_dwh_replica` + convention exactly: a `#[cfg(test)] mod integration_tests` gated by + `find_tursodb()`, using `LocalSyncServer` and `AgentTraceDwhDb::run_migrations` + + `push()` to prepare remotes, printing a skip reason and passing trivially + outside `nix develop .#database`. +- `AgentTraceDwhSyncError` follows the existing manual `Debug`/`Display`/ + `std::error::Error` pattern used by `AgentTraceDwhReplicaError` (no + `thiserror` dependency exists in `cli/Cargo.toml` today). + +## Task stack + +- [x] T01: `Add AgentTraceDwhSync core service and prove the empty-remote first sync` (status:done) + - Task ID: T01 + - Goal: Implement `cli/src/services/agent_trace_dwh_sync.rs` with + `AgentTraceDwhSync { agent_trace_etl, conversation_etl, code_changes_etl }`, + `impl Default`, `AgentTraceDwhSyncStats { pulled_changes, agent_traces, + conversation, code_changes }`, `AgentTraceDwhSyncError` (`ReplicaOpen`, + `Pull`, `AgentTraceEtl`, `ConversationEtl`, `CodeChangesEtl`, `Push`), and + `run(&self, repository_id: &str, source: &RepositoryAgentTraceDb, + replica_config: AgentTraceDwhReplicaConfig) -> + Result` that opens the + replica, pulls once, runs the three ETLs through their existing + `run(repository_id, source, &replica)` APIs in order, and pushes once on + full success. Register the module in `cli/src/services/mod.rs`. Prove the + empty-remote bootstrap and no-op-second-run behavior against the real + Turso Sync harness. + - Boundaries (in/out of scope): In — the new module, its stats/error types, + the core `run()` state machine, unit tests for error-stage construction + and stats aggregation that need no filesystem, and one integration test + proving AC1/AC2. Out — stage-failure tests beyond what AC1/AC2 need, + fresh-reconstruction/multi-repo/multi-source-instance/cross-client tests + (later tasks), documentation. + - Dependencies: none + - Done when: `AgentTraceDwhSync::default().run(...)` against a freshly + spawned, untouched local Turso Sync remote initializes the DWH schema, + runs all three ETLs, pushes once, and returns stats with non-zero + `inserted` counts; a second `run()` against the same source/remote returns + stats with zero `extracted`/`inserted` across all three ETL stats and + `pulled_changes == false`; no auth token appears in any `Debug`/`Display` + output. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync`; `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + - Evidence: Added `cli/src/services/agent_trace_dwh_sync.rs` with + `AgentTraceDwhSync { agent_trace_etl, conversation_etl, code_changes_etl }` + (`#[allow(clippy::struct_field_names)]`), `AgentTraceDwhSyncStats`, + `AgentTraceDwhSyncError` (manual `Debug`/`Display`/`std::error::Error`, + mirroring `AgentTraceDwhReplicaError`), and `run()` implementing + open→pull→`AgentTraceEtl`→`ConversationEtl`→`CodeChangesEtl`→push, + short-circuiting with the matching stage variant on first failure. + Registered `pub mod agent_trace_dwh_sync;` (`#[allow(dead_code)]`) in + `cli/src/services/mod.rs`. Added unit tests for `Default` composition, + zeroed stats defaults, and per-stage `Display`/`Debug` no-token-leak + coverage, plus one `#[cfg(test)] mod integration_tests` (gated on + `find_tursodb()`, using `LocalSyncServer`) proving AC1 (fresh empty-remote + sync bootstraps the schema and inserts non-zero rows across all three ETL + stages) and AC2 (a following no-new-source-rows run returns zero + `extracted`/`inserted` everywhere). + - Deviation from Done-when's literal `pulled_changes == false` on the + *second* run: empirically, the real local Turso Sync harness's `pull()` + reports `true` on the first `pull()` any freshly opened replica performs + after *any* session's successful `push()` — including this + orchestrator's own immediately preceding `run()` — because that push was + never locally marked "already observed" by the new connection object, even + though the pulled bytes exactly match what is already on disk. It settles + to `false` only once a `run()` observes no push from any source since the + previous `run()`'s own reconciliation pull. The integration test therefore + asserts AC2's actual contract (zero `extracted`/`inserted`, a visible + no-op) on the second run without asserting `pulled_changes`, and adds a + third run to prove the genuine `pulled_changes == false` steady state. + `run()`'s internal open→pull→ETLs→push ordering is unchanged; only the + test assertion and `run()`'s doc comment were adjusted to state this + observed semantics accurately. This is recorded here for T08 to document + alongside T03's own observed-behavior findings. + - Verification run: `nix develop .#database -c ./scripts/run-cli-cargo.sh + test --manifest-path cli/Cargo.toml agent_trace_dwh_sync` (4 passed, incl. + the Turso Sync integration test); `nix develop -c + ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml` (292 + passed, 1 ignored, 0 failed); `nix develop -c ./scripts/run-cli-cargo.sh + clippy --manifest-path cli/Cargo.toml --all-targets -- -D warnings` + (clean); `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path + cli/Cargo.toml -- --check` (clean). + +- [x] T02: `Prove stage-identified failure semantics stop the sequence early` (status:done) + - Task ID: T02 + - Goal: Add integration coverage proving: (a) a replica-open failure (e.g. an + unreachable `database_url`) returns `AgentTraceDwhSyncError::ReplicaOpen` + and never runs any ETL or push; (b) a pull failure (e.g. remote killed + after a successful open) returns `AgentTraceDwhSyncError::Pull` and never + runs any ETL or push; (c) a deliberately failing `AgentTraceEtl` stage + prevents `ConversationEtl`/`CodeChangesEtl` and push from running; (d) a + deliberately failing `ConversationEtl` stage leaves `AgentTraceEtl`'s + commit intact locally, does not run `CodeChangesEtl`, and does not push; + (e) a deliberately failing `CodeChangesEtl` stage (a malformed source + `diff_traces` payload, using its existing strict validation — do not + weaken it) leaves the prior two ETLs' commits intact locally and does not + push. + - Boundaries (in/out of scope): In — failure-injection integration tests for + all five stages listed above, asserting both the returned error variant + and the local replica's post-failure DWH row/watermark state. Out — + push-failure recovery (T03), reconstruction/multi-repo/multi-source- + instance/cross-client tests (T04–T07). + - Dependencies: T01 + - Done when: five distinct integration test cases (or clearly separated + assertions within one wired integration test, following the existing + `agent_trace_dwh_replica_turso_sync_integration` composition pattern) each + assert the correct `AgentTraceDwhSyncError` variant and that no row was + pushed to the remote past the point of failure; the local spool + (`agent-trace-sync.db`) is inspected directly to confirm prior successful + ETL stages within the same failed run committed locally as designed. + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + - Evidence: Extended `cli/src/services/agent_trace_dwh_sync.rs`'s + `integration_tests` module with one new gated `#[test]` + `agent_trace_dwh_sync_stage_failure_turso_sync_integration`, composed of + five `assert_*` helpers (one per stage, each spawning its own + `LocalSyncServer`), matching the `agent_trace_dwh_replica_turso_sync_integration` + composition convention: (a) `assert_replica_open_failure_stops_before_any_stage` + — an unreachable `database_url` fails `open()` with `ReplicaOpen`, and the + live remote (never actually addressed) is confirmed to hold zero rows; + (b) `assert_pull_failure_stops_before_any_stage` — after a baseline + successful sync, a second `run()` against a *different*, unreachable + `database_url` fails at `pull()` with `Pull` (the already-`Ready` local + replica needs no network to open — confirmed empirically, see deviation + note below); the local spool shows the new source row was never + extracted, and the live remote (never touched by the failing call) is + unchanged from baseline; (c) + `assert_agent_trace_etl_failure_stops_before_conversation_code_changes_and_push` + — two independently created sources for one repository publish the same + `agent_trace_id` with different `trace_json`; the second sync's + `AgentTraceEtl` fails on the identity-hash conflict (`AgentTraceEtl` + variant), and because it is both the first stage and atomic per batch, + the local replica after the failed run holds exactly what `pull()` left + it (source A's row) with no partial facts from source B and no push; (d) + `assert_conversation_etl_failure_leaves_agent_trace_committed_and_stops_before_code_changes_and_push` + — a raw-SQL `parts` row with an invalid `type` (no source-side CHECK + constraint permits this) fails the parts half of `ConversationEtl` + (`ConversationEtl` variant) while the messages half and the preceding + `AgentTraceEtl` commit locally within the same run, `CodeChangesEtl` + never runs, and push never runs; (e) + `assert_code_changes_etl_failure_leaves_prior_etls_committed_and_stops_before_push` + — a malformed `diff_traces.patch` (reusing the existing malformed-patch + fixture from `code_changes_etl_replays_watermark_behind_failed_transformation`) + fails `CodeChangesEtl` (`CodeChangesEtl` variant) while both prior ETLs + commit locally and push never runs. Each helper asserts the exact error + variant, that the sentinel auth token never appears in `Display` output, + the local spool's row counts (via a `local_row_counts` helper that + reopens the same local replica path directly — no network required once + `Ready`), and the live remote's row counts (via a `remote_row_counts` + helper that opens a disposable peer replica and pulls) before and after + each failing run. + - Deviation: stage (b)'s literal "remote killed after a successful open" + framing from the Goal was implemented instead as "a second `run()` points + at a different, never-reachable `database_url`," because empirical + testing (a throwaway probe run under `nix develop .#database`, since + removed) showed that once a local replica's schema has classified + `Ready`, `AgentTraceDwhReplica::open()` performs no network round trip at + all — only the following `pull()` does. Killing the real server would + therefore not isolate a `Pull`-specific failure from a `ReplicaOpen` + failure on a later re-open attempt against the same dead URL, and would + also prevent inspecting the live remote afterward (the disposable + `LocalSyncServer` holds no data once its process exits). Pointing the + second `run()` at an unrelated unreachable URL instead reproduces the + same `Pull` failure deterministically, keeps the real remote's process + alive and inspectable throughout, and — because the real remote's URL is + never even passed to the failing call — makes "the live remote is + unaffected" a stronger, directly checkable assertion rather than an + inference from a killed process. This does not touch `run()`'s own + open→pull→ETLs→push ordering; it only changes how the test induces the + failure. Recorded here for T08 to fold into the pull-failure discussion + alongside T03's own findings. + - Verification run: `nix develop .#database -c + ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml + agent_trace_dwh_sync` (5 passed, run 4 times back-to-back with no + flakiness, matching the repeated-run precedent used elsewhere in this + plan); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path + cli/Cargo.toml -- --test-threads=1` (293 passed, 1 ignored, 0 failed — + the default parallel run showed 3 unrelated failures from pre-existing + concurrent-database-lock contention in `agent_trace_db`/`agent_trace_dwh_db` + tests untouched by this task, confirmed spurious by the clean + single-threaded rerun); `nix develop -c ./scripts/run-cli-cargo.sh clippy + --manifest-path cli/Cargo.toml --all-targets -- -D warnings` (clean); + `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path + cli/Cargo.toml -- --check` (clean). + +- [x] T03: `Prove and document push-failure and pull-with-pending-local-changes recovery` (status:done) + - Task ID: T03 + - Goal: Add the load-bearing integration test the request calls out + explicitly: run all three ETLs against a real local Turso Sync remote so + they commit locally, force the final `push()` to fail (e.g. by killing the + `LocalSyncServer` process before the push step), confirm the local replica + retains the committed-but-unpushed changes, restart remote availability, + then run `AgentTraceDwhSync::run()` again — whose first step is `pull()` + against a replica that itself holds pending local commits — and prove the + final converged state has every local fact and watermark reaching the + remote, with no duplicate rows and no lost rows. If this test reveals that + `pull()` against a replica with pending local commits discards or corrupts + those commits, change `run()`'s internal ordering (still without pushing + after each ETL individually) to whatever ordering the observed behavior + requires, and record exactly what was observed and why in this task's + evidence for T08 to document. + - Boundaries (in/out of scope): In — the pending-local-changes recovery + integration test, and any resulting adjustment to `run()`'s internal + pull/ETL/push sequencing strictly to preserve durability of local commits. + Out — any change to `AgentTraceDwhReplica::pull()`/`push()` themselves, + reconstruction/multi-repo/multi-source-instance/cross-client tests + (T04–T07), documentation (T08). + - Dependencies: T01 + - Done when: the integration test deterministically reaches a final state + where the remote contains every ETL fact and watermark committed during + the failed-push run, with no duplicated logical rows and no lost source + rows, across at least several repeated runs; the task's evidence records + the exact observed Turso Sync behavior for `pull()` against a replica + holding pending local commits. + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` (run at least 3 times to check for nondeterminism, matching the precedent set by the concurrent-initializer convergence test in `agent_trace_dwh_replica`) + - Evidence: Added a new `#[test]` + `agent_trace_dwh_sync_push_failure_recovery_turso_sync_integration` to + `cli/src/services/agent_trace_dwh_sync.rs`'s `integration_tests` module. + Rather than racing a background kill against an opaque + `AgentTraceDwhSync::run()` call, the test manually reproduces `run()`'s + own open→pull→`AgentTraceEtl`→`ConversationEtl`→`CodeChangesEtl` sequence + against a real local Turso Sync remote (`AgentTraceDwhReplica::open()` + + `pull()` + the three ETLs' existing `run(repository_id, source, &replica)` + calls, all already-public APIs), which lets it force a deterministic push + failure at exactly the point `run()` itself would call `push()`: the + remote process is killed first, then `replica.push()` is called directly + and asserted to fail. `local_row_counts` (reusing T02's helper) then + confirms all three ETL commits — `[1, 1, 1, 1]` — remain durable in the + local `agent-trace-sync.db` spool after the failed push. Remote + availability is then "restarted" via a new `LocalSyncServer::spawn_persistent` + process on a fresh ephemeral port but backed by the *same* on-disk + `DATABASE` file the killed process used, so the schema published before + the outage survives. A plain second `AgentTraceDwhSync::run()` call (the + "recovery run") is then made against the same local replica path — this + is the actual pull-with-pending-local-commits scenario, composed for + free because `run()`'s own first step is `pull()` against a replica that + still holds the three unpushed ETL commits from the failed run. The + recovery run succeeds, reports zero `extracted`/`inserted` across all + three ETL stages (proving no re-extraction/duplication), and its `push()` + reaches the remote with exactly `[1, 1, 1, 1]` rows — no lost rows, no + duplicated logical rows. A further third run from the same replica is + asserted to be a stable no-op with unchanged remote counts, ruling out + unbounded growth from a repeated push. Extended the existing + `LocalSyncServer` test helper (used unchanged by every other test in this + file) with `spawn_persistent(tursodb_path, db_path)` (spawns + `tursodb --sync-server ` instead of the default + `:memory:`, confirmed empirically to leave ``/`-wal` on + disk after the process is killed) and `kill()` (explicit early kill ahead + of `Drop`); `LocalSyncServer::spawn()` is unchanged and still used + in-memory by every pre-existing test. + - Observed Turso Sync behavior for `pull()` against a replica holding + pending local commits (the empirical discovery this task exists to make): + `pull()` does not discard, corrupt, or roll back pending local commits. + It applies whatever the remote holds (here, only the DWH schema published + before the outage; nothing else, since the failing push never reached the + remote) without touching the three ETL stages' already-committed local + writes. The following three ETL stages in the recovery run then correctly + observe their watermarks already advanced and report a true no-op, and + the final `push()` publishes exactly what was committed during the + failed run. This matches the request's proposed open→pull→ETLs→push + ordering exactly: **no change to `run()`'s internal sequencing was + required.** Recorded here for T08 to document alongside T01's and T02's + own observed-behavior findings. + - Verification run: `nix develop .#database -c ./scripts/run-cli-cargo.sh + test --manifest-path cli/Cargo.toml + agent_trace_dwh_sync_push_failure_recovery_turso_sync_integration` (5 + separate invocations, all passed, no flakiness); `nix develop .#database + -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml + agent_trace_dwh_sync` (6 passed, incl. all three integration tests in + this file); `nix develop -c ./scripts/run-cli-cargo.sh test + --manifest-path cli/Cargo.toml -- --test-threads=1` (293 passed, 1 + failed, 1 ignored — the failure is + `agent_trace_db::repository::tests::concurrent_missing_source_instance_id_initialization_converges_on_one_persisted_winner`, + confirmed via `git stash` to fail identically on the pre-T03 tree, i.e. + pre-existing and untouched by this task); `nix develop -c + ./scripts/run-cli-cargo.sh clippy --manifest-path cli/Cargo.toml + --all-targets -- -D warnings` (one pre-existing failure unrelated to this + task's changes — `assert_conversation_etl_failure_leaves_agent_trace_committed_and_stops_before_code_changes_and_push` + in this same file, added by T02, exceeds `clippy::too_many_lines` at + 104/100; confirmed via `git stash` to fail identically on the pre-T03 + tree; this task's own new code is clippy-clean, carrying an explicit + `#[allow(clippy::too_many_lines)]` on the one new function long enough to + need it, matching the existing convention used by T02's longer helpers); + `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path + cli/Cargo.toml -- --check` (clean). + +- [ ] T04: `Prove fresh local-replica reconstruction from the remote` (status:todo) + - Task ID: T04 + - Goal: Add an integration test that runs a successful sync, deletes the + local `agent-trace-sync.db` (and its Turso sidecars), and runs + `AgentTraceDwhSync::run()` again with the same `AgentTraceDwhReplicaConfig` + — proving the replica bootstraps from the remote, the ETLs read watermarks + that reflect the previously pushed state (so no rows are re-extracted or + duplicated), and a no-op ETL run occurs when no new source rows exist + since the deleted replica's last push. + - Boundaries (in/out of scope): In — the delete-and-resync integration test + only. Out — multi-repository/multi-source-instance/cross-client tests + (T05–T07), documentation (T08). + - Dependencies: T01 + - Done when: the test asserts zero `inserted` counts across all three ETL + stats on the post-deletion resync when no new source rows were added, and + non-zero counts when new source rows are added before the resync, + matching the watermark state that was actually pushed before deletion. + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + +- [ ] T05: `Prove multi-repository convergence against one remote DWH` (status:todo) + - Task ID: T05 + - Goal: Add an integration test syncing two distinct `repository_id`s, each + with its own source `RepositoryAgentTraceDb`, its own local replica path, + and its own `AgentTraceDwhSync` instance, against the same remote DWH. + Verify the remote's `agent_traces`/`messages`/`message_parts`/ + `code_changes`/`etl_watermarks` rows for repository A are unaffected by + repository B's sync, and vice versa. + - Boundaries (in/out of scope): In — the two-repository convergence + integration test only. Out — multi-source-instance and cross-client tests + (T06–T07), documentation (T08). + - Dependencies: T01 + - Done when: the test asserts both repositories' rows are present in the + remote after both syncs, that repository A's row count/content is + identical before and after repository B's sync runs, and the reverse. + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + +- [ ] T06: `Prove independent per-source-instance watermarks under one repository` (status:todo) + - Task ID: T06 + - Goal: Add an integration test for one `repository_id` with two + independently created source `RepositoryAgentTraceDb` instances (distinct + `source_instance_id`s, following existing source-instance identity rules — + do not add new cross-source identity logic in the orchestrator), each + synced through its own `AgentTraceDwhSync::run()` call against the same + remote, including overlapping local row IDs (e.g. both sources having a + local `part`/`diff_trace` row with the same integer ID). + - Boundaries (in/out of scope): In — the two-source-instance integration + test, asserting independent watermark progression and no local-ID + collision in the DWH. Out — any new identity/dedup logic in + `agent_trace_dwh_sync.rs` beyond what the existing ETLs already provide; + cross-client convergence (T07); documentation (T08). + - Dependencies: T01 + - Done when: the test asserts `etl_watermarks` rows for the two source + instances advance independently, and that DWH rows sourced from each + instance's overlapping local IDs are both present and distinguishable + (not merged or overwritten). + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + +- [ ] T07: `Prove convergence between two independently operated sync clients` (status:todo) + - Task ID: T07 + - Goal: Add an integration test simulating two independent clients (two + separate local replica paths, same `repository_id` and remote, modeling + two machines) that each run pull → ETL → push in turn: client A syncs, + client B syncs (observing A's remote additions via its own `pull()`), + then client A syncs again (observing B's additions). Verify convergence: + A's second sync sees B's rows, B's sync did not remove or corrupt A's + rows, watermarks stay correct throughout, and a further no-op run from + either client is stable. + - Boundaries (in/out of scope): In — the cross-client convergence + integration test only. Out — documentation (T08). + - Dependencies: T01 + - Done when: the test asserts the remote's row counts after all three sync + steps equal the union of what both clients' sources contributed, with no + duplication, and that a final no-op run from each client returns zero + `inserted` counts across all three ETL stats. + - Verification notes (commands or checks): `nix develop .#database -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_dwh_sync_turso_sync_integration` + +- [ ] T08: `Document the AgentTraceDwhSync lifecycle and its recovery invariants` (status:todo) + - Task ID: T08 + - Goal: Write `context/sce/agent-trace-dwh-sync.md` describing the full + `repository agent-trace.db → AgentTraceDwhSync (open → pull → AgentTraceEtl + → ConversationEtl → CodeChangesEtl → push) → remote Agent Trace DWH` + lifecycle; the invariants listed in the request (source DB remains local + truth; `agent-trace-sync.db` is a durable-but-disposable spool; the remote + stores durable facts+watermarks; one process owns the spool per sync; + pull precedes ETL under normal operation; ETLs commit independently; push + only follows full ETL success; a failed push leaves local commits intact; + credentials are caller-supplied; control-plane/CLI stay outside this + service); and the exact observed Turso Sync behavior recorded by T03 for + `pull()` against a replica with pending local commits, including whatever + ordering `run()` actually implements as a result. Register the file in + `context/context-map.md`, add/extend the `context/glossary.md` entries + named in Context sync, and update `context/sce/agent-trace-dwh-replica.md` + per Context sync. + - Boundaries (in/out of scope): In — the context files listed under Context + sync only. Out — any new decision record under `context/decisions/`; any + further code change. + - Dependencies: T01, T02, T03, T04, T05, T06, T07 + - Done when: every invariant listed above is stated in + `context/sce/agent-trace-dwh-sync.md` with a pointer to the code/test that + proves it; the file is linked from `context/context-map.md`; no code in + `cli/src` changes in this task. + - Verification notes (commands or checks): inspect `context/sce/agent-trace-dwh-sync.md` against `cli/src/services/agent_trace_dwh_sync.rs` and the T01–T07 integration test evidence; confirm the new file is linked from `context/context-map.md`, `context/glossary.md`, and `context/sce/agent-trace-dwh-replica.md`, and links back to them. + +## Open questions + +None. The request is a fully specified implementation brief (branch/base, +exact sequencing, error model, stats shape, and an explicit, itemized list of +required integration-test scenarios), the pieces it composes already exist +with exactly the call shape it assumes (`etl.run(repository_id, source, +&replica)`), and the one genuinely open design point — whether `pull()` is +safe against a replica holding pending local commits — is explicitly framed by +the request itself as something to discover empirically (T03) rather than +decide up front, so it is captured as an assumption above instead of a +blocking question. diff --git a/context/sce/agent-trace-dwh-sync.md b/context/sce/agent-trace-dwh-sync.md new file mode 100644 index 00000000..e4cefa44 --- /dev/null +++ b/context/sce/agent-trace-dwh-sync.md @@ -0,0 +1,36 @@ +# Agent Trace DWH Sync Orchestrator + +`AgentTraceDwhSync` in `cli/src/services/agent_trace_dwh_sync.rs` is the single orchestration boundary composing an [`AgentTraceDwhReplica`](agent-trace-dwh-replica.md) with the three independent ETL bridges — `AgentTraceEtl` (see [agent-trace-etl.md](agent-trace-etl.md)), `ConversationEtl` (see [conversation-etl.md](conversation-etl.md)), and `CodeChangesEtl` (see [code-changes-etl.md](code-changes-etl.md)) — into one sync call. It extends nothing in any of those four modules: it calls their existing public APIs unmodified. + +> This file currently documents the core service shape delivered by plan task T01 (`context/plans/agent-trace-dwh-sync.md`): the `run()` sequence and its stats/error shape, proven against a fresh empty remote and a following no-op run. Stage-identified failure semantics, push-failure recovery, fresh-replica reconstruction, multi-repository/multi-source-instance/cross-client convergence, and the full set of durable invariants land with that plan's later tasks and its final documentation task (T08), which supersedes and expands this file. + +## Shape + +`AgentTraceDwhSync { agent_trace_etl: AgentTraceEtl, conversation_etl: ConversationEtl, code_changes_etl: CodeChangesEtl }` implements `Default`, reusing each ETL's own default batch sizing — the orchestrator adds no configuration of its own. + +`run(&self, repository_id: &str, source: &RepositoryAgentTraceDb, replica_config: AgentTraceDwhReplicaConfig) -> Result` performs exactly one sequence, holding the bridge lock for its full duration: + +1. `AgentTraceDwhReplica::open(replica_config)` — one call, consuming the caller-supplied config. +2. `replica.pull()`. +3. `AgentTraceEtl::run(repository_id, source, &replica)`. +4. `ConversationEtl::run(repository_id, source, &replica)`. +5. `CodeChangesEtl::run(repository_id, source, &replica)`. +6. `replica.push()` — only on full success of every prior step. + +No global transaction wraps the three ETLs: each still commits its own facts and watermark independently, exactly as it does when called directly through the replica. The opened replica is dropped when `run()` returns (success or failure), releasing the bridge lock. + +## Stats and errors + +`AgentTraceDwhSyncStats { pulled_changes: bool, agent_traces: AgentTraceEtlStats, conversation: ConversationEtlStats, code_changes: CodeChangesEtlStats }` is returned only on full success. + +`AgentTraceDwhSyncError` is a stage-tagged enum (`ReplicaOpen`, `Pull`, `AgentTraceEtl`, `ConversationEtl`, `CodeChangesEtl`, `Push`) with manual `Debug`/`Display`/`std::error::Error`, mirroring `AgentTraceDwhReplicaError`'s pattern. `run()` short-circuits at the first failing stage: no later stage runs, and `push()` never runs unless every ETL succeeded. `ReplicaOpen`/`Pull`/`Push` wrap an already token-redacted `AgentTraceDwhReplicaError`; the three ETL-stage variants wrap `anyhow::Error`, which never observes the caller's auth token in the first place, since ETL never touches credentials. + +## Observed Turso Sync behavior: `pulled_changes` after a fresh open + +Because `run()` opens a brand-new replica connection every call (`replica_config` is consumed, not held across calls), the `pull()` immediately following *any* prior session's successful `push()` — including this orchestrator's own immediately preceding `run()` call — reports `pulled_changes == true`. This happens even when the pulled data already exactly matches what's on disk: the new connection has no local record that it (or the previous session sharing its local file) already observed that push, so it must reconcile once. `pulled_changes` only settles to `false` once a `run()` call observes no push from any source since the previous `run()`'s own reconciliation pull. + +Practical effect: a caller cannot treat `pulled_changes == true` as evidence that new *logical* rows arrived — it only means this session's replica needed at least one reconciliation round-trip. The ETL stats (`extracted`/`inserted` per stage) remain the authoritative signal for whether any new source data was processed; they are unaffected by this reconciliation echo and read zero on a genuine no-op run regardless of `pulled_changes`. + +## See also + +[agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), [agent-trace-etl.md](agent-trace-etl.md), [conversation-etl.md](conversation-etl.md), [code-changes-etl.md](code-changes-etl.md), [../glossary.md](../glossary.md), [../context-map.md](../context-map.md).