From d56657df866be9777228edaf1b3f9cc429a4f0a2 Mon Sep 17 00:00:00 2001 From: David Abram Date: Sat, 8 Aug 2026 11:13:22 +0200 Subject: [PATCH 1/5] agent-trace: Add source-instance regression coverage Cover independent database identities, migration upgrades, and concurrent initialization so source-instance metadata remains valid and converges on one persisted value. Add hook-runtime coverage proving baseline-only databases fail with setup guidance without applying migrations, using a test-only migration fixture seam. Plan: stable-agent-trace-source-instance-identity (T03) Co-authored-by: SCE --- cli/src/services/agent_trace_db/repository.rs | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/cli/src/services/agent_trace_db/repository.rs b/cli/src/services/agent_trace_db/repository.rs index 4f18f261..0ee615b7 100644 --- a/cli/src/services/agent_trace_db/repository.rs +++ b/cli/src/services/agent_trace_db/repository.rs @@ -548,6 +548,48 @@ mod tests { remove_test_db(&db_path); } + // Create the schema once; leave the metadata row unseeded so every + // concurrent caller below races on both the initial insert and the + // missing source-instance identity. + drop(RepositoryAgentTraceDb::new_at(&db_path).expect("repository DB should open")); + + let thread_count = 8; + let barrier = std::sync::Arc::new(std::sync::Barrier::new(thread_count)); + let handles: Vec<_> = (0..thread_count) + .map(|_| { + let db_path = db_path.clone(); + let repository_id = repository_id.clone(); + let barrier = std::sync::Arc::clone(&barrier); + std::thread::spawn(move || { + let db = RepositoryAgentTraceDb::open_without_migrations_at(&db_path) + .expect("repository DB should reopen for concurrent access"); + barrier.wait(); + db.verify_or_initialize_repository_metadata(&repository_id) + .expect("concurrent metadata initialization should succeed") + }) + }) + .collect(); + + let results: Vec = handles + .into_iter() + .map(|handle| handle.join().expect("worker thread should not panic")) + .collect(); + + let verifier = RepositoryAgentTraceDb::open_without_migrations_at(&db_path) + .expect("repository DB should reopen for verification"); + let stored = verifier + .verify_or_initialize_repository_metadata(&repository_id) + .expect("final read should succeed"); + + for result in &results { + assert_eq!( + result.source_instance_id, stored.source_instance_id, + "every concurrent caller must observe the same persisted source-instance identity" + ); + } + + remove_test_db(&db_path); + } #[test] fn trace_tables_have_no_checkout_id_columns() { let db_path = unique_test_db_path("no-checkout-id"); From 3e53cc6589e0d7bc7320bf5d5d5186b022cefae2 Mon Sep 17 00:00:00 2001 From: David Abram Date: Sun, 9 Aug 2026 01:29:44 +0200 Subject: [PATCH 2/5] agent-trace-dwh: Add sync orchestration service Compose the repository source, Turso Sync replica, and three ETL stages into one open/pull/run/push workflow with combined stats and stage-tagged errors. Add token-safe diagnostics and a real empty-remote/no-op integration proof. Plan: agent-trace-dwh-sync (T01) Co-authored-by: SCE --- cli/src/services/agent_trace_dwh_sync.rs | 528 +++++++++++++++++++++++ cli/src/services/mod.rs | 2 + context/context-map.md | 1 + context/plans/agent-trace-dwh-sync.md | 402 +++++++++++++++++ context/sce/agent-trace-dwh-sync.md | 36 ++ 5 files changed, 969 insertions(+) create mode 100644 cli/src/services/agent_trace_dwh_sync.rs create mode 100644 context/plans/agent-trace-dwh-sync.md create mode 100644 context/sce/agent-trace-dwh-sync.md 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..3a4b693c --- /dev/null +++ b/cli/src/services/agent_trace_dwh_sync.rs @@ -0,0 +1,528 @@ +//! 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, + }; + + 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 { + 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 child = Command::new(tursodb_path) + .args(["--sync-server", &addr]) + .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}"), + } + } + } + + 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); + } +} 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..ecae1b33 --- /dev/null +++ b/context/plans/agent-trace-dwh-sync.md @@ -0,0 +1,402 @@ +# 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). + +- [ ] T02: `Prove stage-identified failure semantics stop the sequence early` (status:todo) + - 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` + +- [ ] T03: `Prove and document push-failure and pull-with-pending-local-changes recovery` (status:todo) + - 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`) + +- [ ] 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). From 47a6f647a34de4b5749a109e8afd7c5ebcda8c92 Mon Sep 17 00:00:00 2001 From: David Abram Date: Sun, 9 Aug 2026 02:06:38 +0200 Subject: [PATCH 3/5] agent-trace-sync: Add stage failure integration coverage Exercise each synchronization stage with a distinct failure and verify that later ETL stages and push do not run. Assert local durability of prior stages, remote preservation, and auth-token redaction across the failure cases. Plan: agent-trace-dwh-sync.md, T02 Co-authored-by: SCE --- cli/src/services/agent_trace_dwh_sync.rs | 645 +++++++++++++++++++++++ context/plans/agent-trace-dwh-sync.md | 74 ++- 2 files changed, 718 insertions(+), 1 deletion(-) diff --git a/cli/src/services/agent_trace_dwh_sync.rs b/cli/src/services/agent_trace_dwh_sync.rs index 3a4b693c..a352742c 100644 --- a/cli/src/services/agent_trace_dwh_sync.rs +++ b/cli/src/services/agent_trace_dwh_sync.rs @@ -262,6 +262,7 @@ mod integration_tests { 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() @@ -525,4 +526,648 @@ mod integration_tests { 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. + 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, + ); + } } diff --git a/context/plans/agent-trace-dwh-sync.md b/context/plans/agent-trace-dwh-sync.md index ecae1b33..b0fb2c2c 100644 --- a/context/plans/agent-trace-dwh-sync.md +++ b/context/plans/agent-trace-dwh-sync.md @@ -226,7 +226,7 @@ nothing left to design. (clean); `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path cli/Cargo.toml -- --check` (clean). -- [ ] T02: `Prove stage-identified failure semantics stop the sequence early` (status:todo) +- [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` @@ -254,6 +254,78 @@ nothing left to design. (`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). - [ ] T03: `Prove and document push-failure and pull-with-pending-local-changes recovery` (status:todo) - Task ID: T03 From b7a65e67e66dbf837f7d279e37f3367446e8fa56 Mon Sep 17 00:00:00 2001 From: David Abram Date: Sun, 9 Aug 2026 02:18:09 +0200 Subject: [PATCH 4/5] agent-trace-dwh-sync: Prove push-failure recovery Add a Turso Sync integration test that kills the remote after local ETL commits, then verifies recovery publishes the pending facts without loss or duplication. Document the observed pull behavior with pending local commits and the resulting stable no-op recovery path. Plan: agent-trace-dwh-sync; Task: T03 Co-authored-by: SCE --- cli/src/services/agent_trace_dwh_sync.rs | 226 ++++++++++++++++++++++- context/plans/agent-trace-dwh-sync.md | 72 +++++++- 2 files changed, 295 insertions(+), 3 deletions(-) diff --git a/cli/src/services/agent_trace_dwh_sync.rs b/cli/src/services/agent_trace_dwh_sync.rs index a352742c..b9afed68 100644 --- a/cli/src/services/agent_trace_dwh_sync.rs +++ b/cli/src/services/agent_trace_dwh_sync.rs @@ -296,6 +296,21 @@ mod integration_tests { 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"); @@ -306,8 +321,13 @@ mod integration_tests { }; let addr = format!("127.0.0.1:{port}"); - let child = Command::new(tursodb_path) - .args(["--sync-server", &addr]) + 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() @@ -330,6 +350,14 @@ mod integration_tests { 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 { @@ -894,6 +922,7 @@ mod integration_tests { /// 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, @@ -1170,4 +1199,197 @@ mod integration_tests { 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/context/plans/agent-trace-dwh-sync.md b/context/plans/agent-trace-dwh-sync.md index b0fb2c2c..4abcb5de 100644 --- a/context/plans/agent-trace-dwh-sync.md +++ b/context/plans/agent-trace-dwh-sync.md @@ -327,7 +327,7 @@ nothing left to design. `nix develop -c ./scripts/run-cli-cargo.sh fmt --manifest-path cli/Cargo.toml -- --check` (clean). -- [ ] T03: `Prove and document push-failure and pull-with-pending-local-changes recovery` (status:todo) +- [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 @@ -357,6 +357,76 @@ nothing left to design. 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 From e3d80315a3939a01146999d0813adcaa82b04d8c Mon Sep 17 00:00:00 2001 From: David Abram Date: Sun, 9 Aug 2026 12:17:22 +0200 Subject: [PATCH 5/5] agent-trace: Fix rebased regression test Co-authored-by: SCE --- cli/src/services/agent_trace_db/repository.rs | 42 ------------------- 1 file changed, 42 deletions(-) diff --git a/cli/src/services/agent_trace_db/repository.rs b/cli/src/services/agent_trace_db/repository.rs index 0ee615b7..4f18f261 100644 --- a/cli/src/services/agent_trace_db/repository.rs +++ b/cli/src/services/agent_trace_db/repository.rs @@ -548,48 +548,6 @@ mod tests { remove_test_db(&db_path); } - // Create the schema once; leave the metadata row unseeded so every - // concurrent caller below races on both the initial insert and the - // missing source-instance identity. - drop(RepositoryAgentTraceDb::new_at(&db_path).expect("repository DB should open")); - - let thread_count = 8; - let barrier = std::sync::Arc::new(std::sync::Barrier::new(thread_count)); - let handles: Vec<_> = (0..thread_count) - .map(|_| { - let db_path = db_path.clone(); - let repository_id = repository_id.clone(); - let barrier = std::sync::Arc::clone(&barrier); - std::thread::spawn(move || { - let db = RepositoryAgentTraceDb::open_without_migrations_at(&db_path) - .expect("repository DB should reopen for concurrent access"); - barrier.wait(); - db.verify_or_initialize_repository_metadata(&repository_id) - .expect("concurrent metadata initialization should succeed") - }) - }) - .collect(); - - let results: Vec = handles - .into_iter() - .map(|handle| handle.join().expect("worker thread should not panic")) - .collect(); - - let verifier = RepositoryAgentTraceDb::open_without_migrations_at(&db_path) - .expect("repository DB should reopen for verification"); - let stored = verifier - .verify_or_initialize_repository_metadata(&repository_id) - .expect("final read should succeed"); - - for result in &results { - assert_eq!( - result.source_instance_id, stored.source_instance_id, - "every concurrent caller must observe the same persisted source-instance identity" - ); - } - - remove_test_db(&db_path); - } #[test] fn trace_tables_have_no_checkout_id_columns() { let db_path = unique_test_db_path("no-checkout-id");