diff --git a/cli/src/services/agent_trace_etl/mod.rs b/cli/src/services/agent_trace_etl/mod.rs index 44b8e5fa..8e62374d 100644 --- a/cli/src/services/agent_trace_etl/mod.rs +++ b/cli/src/services/agent_trace_etl/mod.rs @@ -6,7 +6,7 @@ //! fact/watermark loading. Source snapshots end before hashing or destination //! work, and destination facts plus progress share one transaction. -use std::{fmt::Write, thread}; +use std::fmt::Write; use anyhow::{Context, Result}; use sha2::{Digest, Sha256}; @@ -15,7 +15,11 @@ use crate::services::agent_trace_dwh_db::AgentTraceDwhDb; use crate::services::agent_trace_dwh_replica::AgentTraceDwhReplica; use crate::services::{ - agent_trace_db::repository::RepositoryAgentTraceDb, resilience::RetryPolicy, + agent_trace_db::repository::RepositoryAgentTraceDb, + etl::{ + read_watermark, run_with_source_contention_retry, upsert_watermark, validate_batch_size, + TableBatchStats, + }, }; /// One immutable Agent Trace row copied out of the repository source @@ -39,17 +43,6 @@ WHERE id > ?1 ORDER BY id ASC LIMIT ?2"; -/// Bounded backoff for source extraction contention retries. Contention on a -/// short, non-blocking read transaction is expected to be rare and -/// self-clearing, so the budget stays small relative to the connection-open -/// retry policy in `crate::services::db`. -const SOURCE_CONTENTION_RETRY_POLICY: RetryPolicy = RetryPolicy { - max_attempts: 5, - timeout_ms: 1_000, - initial_backoff_ms: 25, - max_backoff_ms: 200, -}; - /// Extract one bounded, ordered batch of `agent_traces` rows with `id` /// greater than `watermark`, up to `batch_size` rows, from a short consistent /// read transaction. @@ -69,10 +62,7 @@ pub fn extract_agent_trace_batch( watermark: i64, batch_size: u32, ) -> Result> { - anyhow::ensure!( - batch_size > 0, - "agent trace extraction batch_size must be greater than zero" - ); + validate_batch_size(batch_size, "agent trace extraction")?; run_with_source_contention_retry( |_attempt| { @@ -88,45 +78,6 @@ pub fn extract_agent_trace_batch( ) } -/// Run `operation` with bounded retry limited to transient source contention. -/// -/// `before_retry` runs before every retried attempt (not before the first), -/// so callers can issue a best-effort rollback ahead of the next `BEGIN`. -/// Extracted as its own function so the retry/classification behavior is -/// unit-testable without a real database. -fn run_with_source_contention_retry( - mut operation: impl FnMut(u32) -> Result, - mut before_retry: impl FnMut(), -) -> Result { - let mut attempt = 1; - - loop { - match operation(attempt) { - Ok(value) => return Ok(value), - Err(error) => { - let attempts_remain = attempt < SOURCE_CONTENTION_RETRY_POLICY.max_attempts; - if !attempts_remain || !is_transient_source_contention(&error) { - return Err(error); - } - - before_retry(); - thread::sleep(SOURCE_CONTENTION_RETRY_POLICY.backoff_for_attempt(attempt + 1)); - attempt += 1; - } - } - } -} - -/// Recognize transient source contention worth retrying: Turso's typed `Busy` -/// error (whose message is the `SQLite` "database is locked" text) and the -/// narrow "table is locked" textual form used when typed classification is -/// unavailable. Every other error, including genuine extraction/mapping -/// failures, is not retried. -fn is_transient_source_contention(error: &anyhow::Error) -> bool { - let message = error.to_string().to_lowercase(); - message.contains("database is locked") || message.contains("table is locked") -} - fn source_agent_trace_from_row(row: &turso::Row) -> Result { Ok(SourceAgentTrace { id: row.get(0).context("failed to read agent_traces.id")?, @@ -240,7 +191,10 @@ fn load_transformed_agent_trace_batch_with_failure( txn.execute("INSERT INTO repositories (repository_id) VALUES (?1) ON CONFLICT (repository_id) DO NOTHING", (repository_id,))?; txn.execute("INSERT INTO source_instances (repository_id, source_instance_id) VALUES (?1, ?2) ON CONFLICT (repository_id, source_instance_id) DO NOTHING", (repository_id, source_instance_id))?; - let mut stats = AgentTraceBatchStats { watermark: last_row.source_row_id, ..Default::default() }; + let mut stats = TableBatchStats { + watermark: last_row.source_row_id, + ..Default::default() + }; for (index, row) in rows.iter().enumerate() { let existing = txn.query_map( "SELECT trace_json_sha256 FROM agent_traces WHERE repository_id = ?1 AND agent_trace_id = ?2", @@ -261,8 +215,18 @@ fn load_transformed_agent_trace_batch_with_failure( } } - txn.execute("INSERT INTO etl_watermarks (repository_id, source_instance_id, source_table, last_extracted_source_row_id) VALUES (?1, ?2, ?3, ?4) ON CONFLICT (repository_id, source_instance_id, source_table) DO UPDATE SET last_extracted_source_row_id = excluded.last_extracted_source_row_id, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", (repository_id, source_instance_id, AGENT_TRACES_SOURCE_TABLE, last_row.source_row_id))?; - Ok(stats) + upsert_watermark( + txn, + repository_id, + source_instance_id, + AGENT_TRACES_SOURCE_TABLE, + last_row.source_row_id, + )?; + Ok(AgentTraceBatchStats { + inserted: stats.inserted, + already_present: stats.already_present, + watermark: stats.watermark, + }) }) } @@ -286,10 +250,7 @@ impl Default for AgentTraceEtl { impl AgentTraceEtl { /// Create an ETL runner with the requested bounded source batch size. pub fn with_batch_size(batch_size: u32) -> Result { - anyhow::ensure!( - batch_size > 0, - "agent trace ETL batch_size must be greater than zero" - ); + validate_batch_size(batch_size, "agent trace ETL")?; Ok(Self { batch_size }) } @@ -383,7 +344,12 @@ pub fn read_agent_trace_watermark( repository_id: &str, source_instance_id: &str, ) -> Result { - db.query_map("SELECT COALESCE(last_extracted_source_row_id, 0) FROM etl_watermarks WHERE repository_id = ?1 AND source_instance_id = ?2 AND source_table = ?3", (repository_id, source_instance_id, AGENT_TRACES_SOURCE_TABLE), |row| row.get::(0).map_err(Into::into))?.into_iter().next().map_or(Ok(0), Ok) + read_watermark( + db, + repository_id, + source_instance_id, + AGENT_TRACES_SOURCE_TABLE, + ) } #[cfg(test)] @@ -399,7 +365,10 @@ mod agent_trace_etl_source_tests { use anyhow::anyhow; use super::*; - use crate::services::agent_trace_db::AgentTraceInsert; + use crate::services::{ + agent_trace_db::AgentTraceInsert, + etl::{is_transient_source_contention, SOURCE_CONTENTION_RETRY_POLICY}, + }; fn unique_test_db_path(label: &str) -> PathBuf { let nonce = SystemTime::now() diff --git a/cli/src/services/conversation_etl.rs b/cli/src/services/conversation_etl.rs new file mode 100644 index 00000000..866220f3 --- /dev/null +++ b/cli/src/services/conversation_etl.rs @@ -0,0 +1,425 @@ +//! Conversation-level composition for the independently watermarked messages +//! and message-parts ETL runners. +//! +//! This service owns only local table-runner composition. Source extraction, +//! destination transactions, replica ownership, and table-specific identity +//! rules remain in the sibling modules; pull/push and credential handling are +//! deliberately outside this API. + +use anyhow::{Context, Result}; + +use crate::services::{ + agent_trace_db::repository::RepositoryAgentTraceDb, + agent_trace_dwh_db::AgentTraceDwhDb, + agent_trace_dwh_replica::AgentTraceDwhReplica, + conversation_messages_etl::{MessagesEtl, MessagesEtlStats}, + conversation_parts_etl::{PartsEtl, PartsEtlStats}, +}; + +/// Configuration for both conversation table runners. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ConversationEtl { + messages: MessagesEtl, + parts: PartsEtl, +} + +/// Results for one complete conversation ETL run. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ConversationEtlStats { + pub messages: MessagesEtlStats, + pub parts: PartsEtlStats, +} + +impl ConversationEtl { + /// Create a runner with independently validated message and part batch + /// sizes. + pub fn with_batch_sizes(message_batch_size: u32, part_batch_size: u32) -> Result { + Ok(Self { + messages: MessagesEtl::with_batch_size(message_batch_size)?, + parts: PartsEtl::with_batch_size(part_batch_size)?, + }) + } + + /// Create a runner using one validated batch size for both tables. + pub fn with_batch_size(batch_size: u32) -> Result { + Self::with_batch_sizes(batch_size, batch_size) + } + + /// Return a copy configured with a new message batch size. + pub fn with_message_batch_size(self, batch_size: u32) -> Result { + Ok(Self { + messages: MessagesEtl::with_batch_size(batch_size)?, + ..self + }) + } + + /// Return a copy configured with a new part batch size. + pub fn with_part_batch_size(self, batch_size: u32) -> Result { + Ok(Self { + parts: PartsEtl::with_batch_size(batch_size)?, + ..self + }) + } + + /// Return the configured message batch size. + pub fn message_batch_size(self) -> u32 { + self.messages.batch_size() + } + + /// Return the configured part batch size. + pub fn part_batch_size(self) -> u32 { + self.parts.batch_size() + } + + /// Run messages and parts through the lock-owning DWH replica. + /// + /// Metadata is verified once before the two table runners execute. Each + /// table still reads and commits its own watermark independently: a + /// successful messages run is not part of a transaction with the parts + /// run, and a parts failure cannot roll it back. + pub fn run( + self, + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, + ) -> Result { + let metadata = source + .verify_or_initialize_repository_metadata(repository_id) + .context("failed to verify Agent Trace source metadata")?; + run_with_destination( + self, + repository_id, + &metadata.source_instance_id, + source, + replica.db(), + ) + } +} + +pub(crate) fn run_with_destination( + config: ConversationEtl, + repository_id: &str, + source_instance_id: &str, + source: &RepositoryAgentTraceDb, + destination: &AgentTraceDwhDb, +) -> Result { + let messages = crate::services::conversation_messages_etl::run_with_destination( + config.messages, + repository_id, + source_instance_id, + source, + destination, + )?; + let parts = crate::services::conversation_parts_etl::run_with_destination( + config.parts, + repository_id, + source_instance_id, + source, + destination, + )?; + + Ok(ConversationEtlStats { messages, parts }) +} + +#[cfg(test)] +mod tests { + use std::{ + fs, + path::{Path, PathBuf}, + sync::mpsc, + thread, + time::{SystemTime, UNIX_EPOCH}, + }; + + use super::*; + use crate::services::agent_trace_db::{ + repository::RepositoryAgentTraceDb, InsertMessageInsert, InsertPartInsert, MessageRole, + PartType, + }; + + fn unique_path(label: &str, file: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after Unix epoch") + .as_nanos(); + std::env::temp_dir() + .join(format!( + "sce-conversation-etl-{label}-{}-{nonce}", + std::process::id() + )) + .join(file) + } + + fn clean(path: &Path) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } + } + + fn insert_message(db: &RepositoryAgentTraceDb, message_id: &str, timestamp: i64) { + db.insert_message(InsertMessageInsert { + session_id: String::from("session-1"), + message_id: message_id.to_string(), + role: MessageRole::User, + generated_at_unix_ms: timestamp, + }) + .expect("source message insert should succeed"); + } + + fn insert_part(db: &RepositoryAgentTraceDb, message_id: &str, text: &str, timestamp: i64) { + db.insert_part(InsertPartInsert { + part_type: PartType::Text, + text: text.to_string(), + session_id: String::from("session-1"), + message_id: message_id.to_string(), + generated_at_unix_ms: timestamp, + }) + .expect("source part insert should succeed"); + } + + fn open_source_and_destination( + label: &str, + ) -> ( + PathBuf, + RepositoryAgentTraceDb, + PathBuf, + AgentTraceDwhDb, + String, + ) { + let source_path = unique_path(label, "agent-trace.db"); + let dwh_path = unique_path(label, "agent-trace-dwh.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + let destination = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let metadata = source + .verify_or_initialize_repository_metadata("repo-a") + .unwrap(); + ( + source_path, + source, + dwh_path, + destination, + metadata.source_instance_id, + ) + } + + #[test] + fn conversation_etl_defaults_to_independent_500_row_batches() { + let config = ConversationEtl::default(); + assert_eq!(config.message_batch_size(), 500); + assert_eq!(config.part_batch_size(), 500); + + let config = ConversationEtl::with_batch_sizes(2, 3).unwrap(); + assert_eq!(config.message_batch_size(), 2); + assert_eq!(config.part_batch_size(), 3); + } + + #[test] + fn conversation_etl_rejects_invalid_batch_sizes() { + assert!(ConversationEtl::with_batch_sizes(0, 1).is_err()); + assert!(ConversationEtl::with_batch_sizes(1, 0).is_err()); + assert!(ConversationEtl::default() + .with_message_batch_size(0) + .is_err()); + assert!(ConversationEtl::default().with_part_batch_size(0).is_err()); + } + + #[test] + fn conversation_etl_runs_initial_incremental_and_noop_batches() { + let (source_path, source, dwh_path, destination, source_instance_id) = + open_source_and_destination("growth"); + insert_message(&source, "message-1", 1_000); + insert_part(&source, "message-1", "one", 1_001); + + let config = ConversationEtl::with_batch_sizes(1, 1).unwrap(); + let first = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(first.messages.extracted, 1); + assert_eq!(first.messages.inserted, 1); + assert_eq!(first.parts.extracted, 1); + assert_eq!(first.parts.inserted, 1); + assert_eq!(first.messages.after_watermark, 1); + assert_eq!(first.parts.after_watermark, 1); + + insert_message(&source, "message-2", 2_000); + insert_part(&source, "message-2", "two", 2_001); + let second = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(second.messages.before_watermark, 1); + assert_eq!(second.messages.extracted, 1); + assert_eq!(second.messages.after_watermark, 2); + assert_eq!(second.parts.before_watermark, 1); + assert_eq!(second.parts.extracted, 1); + assert_eq!(second.parts.after_watermark, 2); + + let noop = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(noop.messages.extracted, 0); + assert_eq!(noop.messages.batches, 0); + assert_eq!(noop.messages.before_watermark, 2); + assert_eq!(noop.messages.after_watermark, 2); + assert_eq!(noop.parts.extracted, 0); + assert_eq!(noop.parts.batches, 0); + assert_eq!(noop.parts.before_watermark, 2); + assert_eq!(noop.parts.after_watermark, 2); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn conversation_etl_advances_message_and_part_watermarks_independently() { + let (source_path, source, dwh_path, destination, source_instance_id) = + open_source_and_destination("independent-progress"); + insert_message(&source, "message-1", 1_000); + insert_part(&source, "message-1", "one", 1_000); + + let config = ConversationEtl::with_batch_size(10).unwrap(); + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination).unwrap(); + + insert_part(&source, "message-1", "two", 1_001); + let parts_only = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(parts_only.messages.extracted, 0); + assert_eq!(parts_only.messages.before_watermark, 1); + assert_eq!(parts_only.parts.extracted, 1); + assert_eq!(parts_only.parts.before_watermark, 1); + assert_eq!(parts_only.parts.after_watermark, 2); + + insert_message(&source, "message-2", 2_000); + let messages_only = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(messages_only.messages.extracted, 1); + assert_eq!(messages_only.messages.before_watermark, 1); + assert_eq!(messages_only.messages.after_watermark, 2); + assert_eq!(messages_only.parts.extracted, 0); + assert_eq!(messages_only.parts.before_watermark, 2); + assert_eq!(messages_only.parts.after_watermark, 2); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn conversation_etl_allows_parts_before_messages_and_reconstructs_equal_timestamps() { + let (source_path, source, dwh_path, destination, source_instance_id) = + open_source_and_destination("out-of-order"); + insert_part(&source, "message-later", "first", 1_000); + insert_part(&source, "message-later", "second", 1_000); + + let config = ConversationEtl::with_batch_size(10).unwrap(); + let parts_first = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(parts_first.messages.extracted, 0); + assert_eq!(parts_first.parts.inserted, 2); + + let before_parent = destination + .query_map("SELECT COUNT(*) FROM messages", (), |row| { + row.get::(0).map_err(Into::into) + }) + .unwrap(); + assert_eq!(before_parent, vec![0]); + + let ordered = destination + .query_map( + "SELECT text FROM message_parts + WHERE repository_id = ?1 AND session_id = ?2 AND message_id = ?3 + ORDER BY generated_at_unix_ms ASC, source_part_id ASC", + ("repo-a", "session-1", "message-later"), + |row| row.get::(0).map_err(Into::into), + ) + .unwrap(); + assert_eq!(ordered, vec![String::from("first"), String::from("second")]); + + insert_message(&source, "message-later", 1_001); + let message_after_parts = + run_with_destination(config, "repo-a", &source_instance_id, &source, &destination) + .unwrap(); + assert_eq!(message_after_parts.messages.inserted, 1); + assert_eq!(message_after_parts.parts.extracted, 0); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn conversation_etl_source_read_transactions_do_not_block_message_writers() { + let source_path = unique_path("messages-writer", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + insert_message(&source, "message-1", 1_000); + let (reader_ready_tx, reader_ready_rx) = mpsc::channel::<()>(); + let (release_reader_tx, release_reader_rx) = mpsc::channel::<()>(); + let reader_path = source_path.clone(); + let reader = thread::spawn(move || { + let reader_db = RepositoryAgentTraceDb::open_without_migrations_at(&reader_path) + .expect("reader connection should reopen"); + reader_db.read_transaction(|txn| { + txn.query_map("SELECT id FROM messages ORDER BY id", (), |row| { + row.get::(0).map_err(Into::into) + })?; + reader_ready_tx + .send(()) + .expect("reader should signal readiness"); + release_reader_rx + .recv() + .expect("reader should wait while holding the snapshot"); + Ok(()) + }) + }); + + reader_ready_rx + .recv() + .expect("test should observe the open message snapshot"); + let writer = RepositoryAgentTraceDb::open_without_migrations_at(&source_path).unwrap(); + insert_message(&writer, "message-2", 1_001); + release_reader_tx.send(()).unwrap(); + reader + .join() + .expect("reader thread should not panic") + .expect("reader transaction should commit"); + clean(&source_path); + } + + #[test] + fn conversation_etl_source_read_transactions_do_not_block_part_writers() { + let source_path = unique_path("parts-writer", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + insert_part(&source, "message-1", "one", 1_000); + let (reader_ready_tx, reader_ready_rx) = mpsc::channel::<()>(); + let (release_reader_tx, release_reader_rx) = mpsc::channel::<()>(); + let reader_path = source_path.clone(); + let reader = thread::spawn(move || { + let reader_db = RepositoryAgentTraceDb::open_without_migrations_at(&reader_path) + .expect("reader connection should reopen"); + reader_db.read_transaction(|txn| { + txn.query_map("SELECT id FROM parts ORDER BY id", (), |row| { + row.get::(0).map_err(Into::into) + })?; + reader_ready_tx + .send(()) + .expect("reader should signal readiness"); + release_reader_rx + .recv() + .expect("reader should wait while holding the snapshot"); + Ok(()) + }) + }); + + reader_ready_rx + .recv() + .expect("test should observe the open parts snapshot"); + let writer = RepositoryAgentTraceDb::open_without_migrations_at(&source_path).unwrap(); + insert_part(&writer, "message-1", "two", 1_001); + release_reader_tx.send(()).unwrap(); + reader + .join() + .expect("reader thread should not panic") + .expect("reader transaction should commit"); + clean(&source_path); + } +} diff --git a/cli/src/services/conversation_messages_etl.rs b/cli/src/services/conversation_messages_etl.rs new file mode 100644 index 00000000..7ae9c6c2 --- /dev/null +++ b/cli/src/services/conversation_messages_etl.rs @@ -0,0 +1,611 @@ +//! Incremental repository-source to DWH ETL for logical conversation messages. +//! +//! Source extraction is intentionally short and read-only. Transformation and +//! destination loading happen after the source snapshot ends, and each batch's +//! facts, dimensions, and `messages` watermark commit in one destination +//! transaction. + +use anyhow::{bail, Context, Result}; + +use crate::services::{ + agent_trace_db::{repository::RepositoryAgentTraceDb, MessageRole}, + agent_trace_dwh_db::AgentTraceDwhDb, + agent_trace_dwh_replica::AgentTraceDwhReplica, + db::TursoTransaction, + etl::{ + read_watermark, run_with_source_contention_retry, upsert_watermark, validate_batch_size, + TableBatchStats, + }, +}; + +/// The source table represented by this pipeline. +pub const MESSAGES_SOURCE_TABLE: &str = "messages"; + +/// Default number of source messages processed by one batch. +pub const DEFAULT_MESSAGES_ETL_BATCH_SIZE: u32 = 500; + +/// One source message copied from a short repository database snapshot. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SourceMessage { + pub id: i64, + pub session_id: String, + pub message_id: String, + pub role: String, + pub generated_at_unix_ms: i64, +} + +/// A source message after role validation and destination-independent +/// transformation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TransformedMessage { + pub source_row_id: i64, + pub session_id: String, + pub message_id: String, + pub role: MessageRole, + pub generated_at_unix_ms: i64, +} + +/// Counts returned by one atomically loaded messages batch. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct MessagesBatchStats { + pub inserted: u64, + pub already_present: u64, + pub watermark: i64, +} + +/// Summary of one complete incremental messages ETL run. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct MessagesEtlStats { + pub extracted: u64, + pub inserted: u64, + pub already_present: u64, + pub batches: u64, + pub before_watermark: i64, + pub after_watermark: i64, +} + +/// Configuration for the independently watermarked messages pipeline. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct MessagesEtl { + batch_size: u32, +} + +/// Descriptive alias for callers composing conversation table pipelines. +pub type ConversationMessagesEtl = MessagesEtl; + +const SELECT_MESSAGES_BATCH_SQL: &str = + "SELECT id, session_id, message_id, role, generated_at_unix_ms +FROM messages +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +/// Extract one bounded, ascending source batch in a short read transaction. +/// +/// Only transient database/table lock contention is retried. The source read +/// transaction is committed before this function returns, so transformation +/// and destination work never hold a source snapshot open. +pub fn extract_message_batch( + db: &RepositoryAgentTraceDb, + watermark: i64, + batch_size: u32, +) -> Result> { + validate_batch_size(batch_size, "messages extraction")?; + + run_with_source_contention_retry( + |_attempt| { + db.read_transaction(|txn| { + txn.query_map( + SELECT_MESSAGES_BATCH_SQL, + (watermark, i64::from(batch_size)), + source_message_from_row, + ) + }) + }, + || db.rollback_best_effort(), + ) +} + +fn source_message_from_row(row: &turso::Row) -> Result { + Ok(SourceMessage { + id: row.get(0).context("failed to read messages.id")?, + session_id: row.get(1).context("failed to read messages.session_id")?, + message_id: row.get(2).context("failed to read messages.message_id")?, + role: row.get(3).context("failed to read messages.role")?, + generated_at_unix_ms: row + .get(4) + .context("failed to read messages.generated_at_unix_ms")?, + }) +} + +/// Validate the source role and prepare a message for destination loading. +pub fn transform_message(source: &SourceMessage) -> Result { + let role = match source.role.as_str() { + "user" => MessageRole::User, + "assistant" => MessageRole::Assistant, + other => bail!( + "unsupported source messages.role '{other}' for message {}", + source.message_id + ), + }; + + Ok(TransformedMessage { + source_row_id: source.id, + session_id: source.session_id.clone(), + message_id: source.message_id.clone(), + role, + generated_at_unix_ms: source.generated_at_unix_ms, + }) +} + +/// Load one source batch atomically into the DWH messages table. +pub fn load_message_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + source_rows: &[SourceMessage], +) -> Result { + let transformed = source_rows + .iter() + .map(transform_message) + .collect::>>()?; + load_transformed_message_batch(db, repository_id, source_instance_id, &transformed) +} + +fn load_transformed_message_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedMessage], +) -> Result { + load_transformed_message_batch_with_failure(db, repository_id, source_instance_id, rows, None) +} + +fn load_transformed_message_batch_with_failure( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedMessage], + fail_after_row: Option, +) -> Result { + let Some(last_row) = rows.last() else { + return Ok(MessagesBatchStats::default()); + }; + + db.transaction(|txn| { + ensure_lineage(txn, repository_id, source_instance_id)?; + + let mut stats = TableBatchStats { + watermark: last_row.source_row_id, + ..Default::default() + }; + + for (index, row) in rows.iter().enumerate() { + let existing = txn.query_map( + "SELECT role, generated_at_unix_ms FROM messages + WHERE repository_id = ?1 AND session_id = ?2 AND message_id = ?3", + (repository_id, row.session_id.as_str(), row.message_id.as_str()), + |db_row| { + Ok(( + db_row.get::(0)?, + db_row.get::(1)?, + )) + }, + )?; + + if let Some((existing_role, existing_timestamp)) = existing.into_iter().next() { + let incoming_role = row.role.to_string(); + if existing_role != incoming_role || existing_timestamp != row.generated_at_unix_ms + { + bail!( + "message integrity conflict for repository {repository_id}, session {}, message {}: existing role/timestamp {existing_role}/{existing_timestamp}, incoming {incoming_role}/{}", + row.session_id, + row.message_id, + row.generated_at_unix_ms + ); + } + stats.already_present += 1; + } else { + txn.execute( + "INSERT INTO messages (repository_id, source_instance_id, session_id, message_id, role, generated_at_unix_ms) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + ( + repository_id, + source_instance_id, + row.session_id.as_str(), + row.message_id.as_str(), + row.role.to_string(), + row.generated_at_unix_ms, + ), + )?; + stats.inserted += 1; + } + + if fail_after_row == Some(index + 1) { + bail!("injected messages destination failure after row {}", index + 1); + } + } + + upsert_watermark( + txn, + repository_id, + source_instance_id, + MESSAGES_SOURCE_TABLE, + last_row.source_row_id, + )?; + + Ok(MessagesBatchStats { + inserted: stats.inserted, + already_present: stats.already_present, + watermark: stats.watermark, + }) + }) +} + +fn ensure_lineage( + txn: &TursoTransaction<'_, crate::services::agent_trace_dwh_db::AgentTraceDwhDbSpec>, + repository_id: &str, + source_instance_id: &str, +) -> Result<()> { + txn.execute( + "INSERT INTO repositories (repository_id) VALUES (?1) + ON CONFLICT (repository_id) DO NOTHING", + (repository_id,), + )?; + txn.execute( + "INSERT INTO source_instances (repository_id, source_instance_id) VALUES (?1, ?2) + ON CONFLICT (repository_id, source_instance_id) DO NOTHING", + (repository_id, source_instance_id), + )?; + Ok(()) +} + +impl Default for MessagesEtl { + fn default() -> Self { + Self { + batch_size: DEFAULT_MESSAGES_ETL_BATCH_SIZE, + } + } +} + +impl MessagesEtl { + /// Create a messages runner with a positive bounded source batch size. + pub fn with_batch_size(batch_size: u32) -> Result { + validate_batch_size(batch_size, "messages ETL")?; + Ok(Self { batch_size }) + } + + /// Return the configured source batch size. + pub fn batch_size(self) -> u32 { + self.batch_size + } + + /// Run the independently watermarked messages ETL through an open replica. + pub fn run( + self, + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, + ) -> Result { + let metadata = source + .verify_or_initialize_repository_metadata(repository_id) + .context("failed to verify Agent Trace source metadata")?; + run_with_destination( + self, + repository_id, + &metadata.source_instance_id, + source, + replica.db(), + ) + } +} + +pub(crate) fn run_with_destination( + config: MessagesEtl, + repository_id: &str, + source_instance_id: &str, + source: &RepositoryAgentTraceDb, + destination: &AgentTraceDwhDb, +) -> Result { + let before_watermark = read_messages_watermark(destination, repository_id, source_instance_id)?; + let mut watermark = before_watermark; + let mut stats = MessagesEtlStats { + before_watermark, + after_watermark: before_watermark, + ..Default::default() + }; + + loop { + let rows = extract_message_batch(source, watermark, config.batch_size)?; + if rows.is_empty() { + break; + } + + let batch = load_message_batch(destination, repository_id, source_instance_id, &rows)?; + watermark = batch.watermark; + stats.extracted += rows.len() as u64; + stats.inserted += batch.inserted; + stats.already_present += batch.already_present; + stats.batches += 1; + stats.after_watermark = watermark; + } + + Ok(stats) +} + +/// Run the default-sized messages ETL through an open DWH replica. +pub fn run_messages_etl( + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, +) -> Result { + MessagesEtl::default().run(repository_id, source, replica) +} + +/// Read the messages watermark, treating an absent row as zero. +pub fn read_messages_watermark( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, +) -> Result { + read_watermark(db, repository_id, source_instance_id, MESSAGES_SOURCE_TABLE) +} + +#[cfg(test)] +mod tests { + use std::{ + fs, + path::PathBuf, + time::{SystemTime, UNIX_EPOCH}, + }; + + use super::*; + use crate::services::agent_trace_db::{InsertMessageInsert, MessageRole}; + + fn unique_path(label: &str, file: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after Unix epoch") + .as_nanos(); + std::env::temp_dir() + .join(format!( + "sce-conversation-messages-{label}-{}-{nonce}", + std::process::id() + )) + .join(file) + } + + fn clean(path: &std::path::Path) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } + } + + fn insert_message(db: &RepositoryAgentTraceDb, id: &str, role: MessageRole, timestamp: i64) { + db.insert_message(InsertMessageInsert { + session_id: String::from("session-1"), + message_id: id.to_string(), + role, + generated_at_unix_ms: timestamp, + }) + .expect("source message insert should succeed"); + } + + fn source_row( + id: i64, + session_id: &str, + message_id: &str, + role: &str, + timestamp: i64, + ) -> SourceMessage { + SourceMessage { + id, + session_id: session_id.to_string(), + message_id: message_id.to_string(), + role: role.to_string(), + generated_at_unix_ms: timestamp, + } + } + + fn transformed_row( + id: i64, + session_id: &str, + message_id: &str, + role: MessageRole, + timestamp: i64, + ) -> TransformedMessage { + TransformedMessage { + source_row_id: id, + session_id: session_id.to_string(), + message_id: message_id.to_string(), + role, + generated_at_unix_ms: timestamp, + } + } + + #[test] + fn conversation_messages_etl_extracts_ordered_bounded_batches_from_zero() { + let source_path = unique_path("extract", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + insert_message(&source, "message-1", MessageRole::User, 1_000); + insert_message(&source, "message-2", MessageRole::Assistant, 1_001); + insert_message(&source, "message-3", MessageRole::User, 1_002); + + let rows = extract_message_batch(&source, 0, 2).unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!( + rows.iter().map(|row| row.id).collect::>(), + vec![1, 2] + ); + assert_eq!(rows[1].role, "assistant"); + assert!(extract_message_batch(&source, 3, 10).unwrap().is_empty()); + + clean(&source_path); + } + + #[test] + fn conversation_messages_etl_rejects_unknown_roles() { + let error = transform_message(&source_row(1, "session-1", "message-1", "system", 1_000)) + .expect_err("unsupported source roles must fail"); + assert!(error + .to_string() + .contains("unsupported source messages.role 'system'")); + } + + #[test] + fn conversation_messages_etl_inserts_lineage_content_and_watermark() { + let dwh_path = unique_path("insert", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let rows = vec![ + source_row(1, "session-1", "message-1", "user", 1_000), + source_row(2, "session-1", "message-2", "assistant", 1_001), + ]; + + let stats = load_message_batch(&dwh, "repo-a", "instance-a", &rows).unwrap(); + assert_eq!(stats.inserted, 2); + assert_eq!(stats.watermark, 2); + assert_eq!( + read_messages_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 2 + ); + let values = dwh + .query_map( + "SELECT source_instance_id, session_id, message_id, role, generated_at_unix_ms + FROM messages ORDER BY id", + (), + |row| { + Ok(( + row.get::(0)?, + row.get::(1)?, + row.get::(2)?, + row.get::(3)?, + row.get::(4)?, + )) + }, + ) + .unwrap(); + assert_eq!( + values[0], + ( + "instance-a".into(), + "session-1".into(), + "message-1".into(), + "user".into(), + 1_000 + ) + ); + assert_eq!(values[1].3, "assistant"); + + clean(&dwh_path); + } + + #[test] + fn conversation_messages_identity_matching_replay_counts_already_present() { + let dwh_path = unique_path("replay", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let row = source_row(7, "session-1", "message-1", "assistant", 1_000); + + assert_eq!( + load_message_batch(&dwh, "repo-a", "instance-a", std::slice::from_ref(&row)) + .unwrap() + .inserted, + 1 + ); + let replay = load_message_batch(&dwh, "repo-a", "instance-a", &[row]).unwrap(); + assert_eq!(replay.already_present, 1); + assert_eq!( + dwh.query_map("SELECT COUNT(*) FROM messages", (), |db_row| db_row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![1] + ); + + clean(&dwh_path); + } + + #[test] + fn conversation_messages_identity_conflict_contains_logical_identity_and_rolls_back() { + let dwh_path = unique_path("conflict", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let first = source_row(1, "session-1", "message-1", "user", 1_000); + load_message_batch(&dwh, "repo-a", "instance-a", &[first]).unwrap(); + + let error = load_message_batch( + &dwh, + "repo-a", + "instance-a", + &[source_row(2, "session-1", "message-1", "assistant", 1_001)], + ) + .expect_err("a changed logical message must fail"); + let message = error.to_string(); + assert!(message.contains("repo-a")); + assert!(message.contains("session-1")); + assert!(message.contains("message-1")); + assert_eq!( + read_messages_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 1 + ); + + clean(&dwh_path); + } + + #[test] + fn conversation_messages_identity_rolls_back_facts_dimensions_and_watermark() { + let dwh_path = unique_path("rollback", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let rows = vec![ + transformed_row(1, "session-1", "message-1", MessageRole::User, 1_000), + transformed_row(2, "session-1", "message-2", MessageRole::Assistant, 1_001), + ]; + + load_transformed_message_batch_with_failure(&dwh, "repo-a", "instance-a", &rows, Some(1)) + .expect_err("injected failure should roll back the complete batch"); + + for table in [ + "repositories", + "source_instances", + "messages", + "etl_watermarks", + ] { + assert_eq!( + dwh.query_map(&format!("SELECT COUNT(*) FROM {table}"), (), |row| row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![0], + "{table} should be rolled back" + ); + } + + let replay = load_message_batch( + &dwh, + "repo-a", + "instance-a", + &[ + source_row(1, "session-1", "message-1", "user", 1_000), + source_row(2, "session-1", "message-2", "assistant", 1_001), + ], + ) + .unwrap(); + assert_eq!(replay.inserted, 2); + assert_eq!(replay.watermark, 2); + + clean(&dwh_path); + } + + #[test] + fn conversation_messages_etl_rejects_zero_batch_size() { + let source_path = unique_path("zero-batch", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + let error = extract_message_batch(&source, 0, 0).expect_err("zero batch must fail"); + assert!(error.to_string().contains("batch_size")); + clean(&source_path); + } + + #[test] + fn conversation_messages_etl_config_rejects_zero_batch_size() { + let error = MessagesEtl::with_batch_size(0).expect_err("zero batch must fail"); + assert!(error.to_string().contains("batch_size")); + } +} diff --git a/cli/src/services/conversation_parts_etl.rs b/cli/src/services/conversation_parts_etl.rs new file mode 100644 index 00000000..5f17e067 --- /dev/null +++ b/cli/src/services/conversation_parts_etl.rs @@ -0,0 +1,818 @@ +//! Incremental repository-source to DWH ETL for message parts. +//! +//! Parts retain source lineage and the exact source text. Source extraction is +//! short and read-only; transformation and destination loading happen after +//! the source snapshot ends, and each batch's facts, dimensions, and `parts` +//! watermark commit in one destination transaction. + +use std::fmt::Write; + +use anyhow::{bail, Context, Result}; +use sha2::{Digest, Sha256}; + +use crate::services::{ + agent_trace_db::{repository::RepositoryAgentTraceDb, PartType}, + agent_trace_dwh_db::AgentTraceDwhDb, + agent_trace_dwh_replica::AgentTraceDwhReplica, + db::TursoTransaction, + etl::{ + read_watermark, run_with_source_contention_retry, upsert_watermark, validate_batch_size, + TableBatchStats, + }, +}; + +/// The source table represented by this pipeline. +pub const PARTS_SOURCE_TABLE: &str = "parts"; + +/// Default number of source parts processed by one batch. +pub const DEFAULT_PARTS_ETL_BATCH_SIZE: u32 = 500; + +/// One source message part copied from a short repository database snapshot. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SourceMessagePart { + pub id: i64, + pub part_type: String, + pub text: String, + pub message_id: String, + pub session_id: String, + pub generated_at_unix_ms: i64, +} + +/// A source part after type validation and exact-text hashing. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TransformedMessagePart { + pub source_part_id: i64, + pub part_type: PartType, + pub text: String, + pub text_sha256: String, + pub message_id: String, + pub session_id: String, + pub generated_at_unix_ms: i64, +} + +/// Counts returned by one atomically loaded parts batch. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct PartsBatchStats { + pub inserted: u64, + pub already_present: u64, + pub watermark: i64, +} + +/// Summary of one complete incremental parts ETL run. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct PartsEtlStats { + pub extracted: u64, + pub inserted: u64, + pub already_present: u64, + pub batches: u64, + pub before_watermark: i64, + pub after_watermark: i64, +} + +/// Configuration for the independently watermarked parts pipeline. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PartsEtl { + batch_size: u32, +} + +const SELECT_PARTS_BATCH_SQL: &str = + "SELECT id, type, text, message_id, session_id, generated_at_unix_ms +FROM parts +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +/// Extract one bounded, ascending source batch in a short read transaction. +/// +/// Only transient database/table lock contention is retried. The source read +/// transaction is committed before this function returns, so transformation +/// and destination work never hold a source snapshot open. +pub fn extract_part_batch( + db: &RepositoryAgentTraceDb, + watermark: i64, + batch_size: u32, +) -> Result> { + validate_batch_size(batch_size, "parts extraction")?; + + run_with_source_contention_retry( + |_attempt| { + db.read_transaction(|txn| { + txn.query_map( + SELECT_PARTS_BATCH_SQL, + (watermark, i64::from(batch_size)), + source_message_part_from_row, + ) + }) + }, + || db.rollback_best_effort(), + ) +} + +fn source_message_part_from_row(row: &turso::Row) -> Result { + Ok(SourceMessagePart { + id: row.get(0).context("failed to read parts.id")?, + part_type: row.get(1).context("failed to read parts.type")?, + text: row.get(2).context("failed to read parts.text")?, + message_id: row.get(3).context("failed to read parts.message_id")?, + session_id: row.get(4).context("failed to read parts.session_id")?, + generated_at_unix_ms: row + .get(5) + .context("failed to read parts.generated_at_unix_ms")?, + }) +} + +/// Validate the source part type and hash the exact UTF-8 text bytes. +pub fn transform_message_part(source: &SourceMessagePart) -> Result { + let part_type = match source.part_type.as_str() { + "text" => PartType::Text, + "reasoning" => PartType::Reasoning, + "patch" => PartType::Patch, + "question" => PartType::Question, + other => bail!( + "unsupported source parts.type '{other}' for part {}", + source.id + ), + }; + + Ok(TransformedMessagePart { + source_part_id: source.id, + part_type, + text: source.text.clone(), + text_sha256: sha256_hex(source.text.as_bytes()), + message_id: source.message_id.clone(), + session_id: source.session_id.clone(), + generated_at_unix_ms: source.generated_at_unix_ms, + }) +} + +fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + digest.iter().fold( + String::with_capacity(digest.len() * 2), + |mut output, byte| { + write!(&mut output, "{byte:02x}").expect("writing to String cannot fail"); + output + }, + ) +} + +/// Load one source batch atomically into the DWH `message_parts` table. +pub fn load_part_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + source_rows: &[SourceMessagePart], +) -> Result { + let transformed = source_rows + .iter() + .map(transform_message_part) + .collect::>>()?; + load_transformed_part_batch(db, repository_id, source_instance_id, &transformed) +} + +fn load_transformed_part_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedMessagePart], +) -> Result { + load_transformed_part_batch_with_failure(db, repository_id, source_instance_id, rows, None) +} + +fn load_transformed_part_batch_with_failure( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedMessagePart], + fail_after_row: Option, +) -> Result { + let Some(last_row) = rows.last() else { + return Ok(PartsBatchStats::default()); + }; + + db.transaction(|txn| { + ensure_lineage(txn, repository_id, source_instance_id)?; + + let mut stats = TableBatchStats { + watermark: last_row.source_part_id, + ..Default::default() + }; + + for (index, row) in rows.iter().enumerate() { + let existing = txn.query_map( + "SELECT session_id, message_id, part_type, text, text_sha256, generated_at_unix_ms + FROM message_parts + WHERE repository_id = ?1 AND source_instance_id = ?2 AND source_part_id = ?3", + (repository_id, source_instance_id, row.source_part_id), + |db_row| { + Ok(( + db_row.get::(0)?, + db_row.get::(1)?, + db_row.get::(2)?, + db_row.get::(3)?, + db_row.get::(4)?, + db_row.get::(5)?, + )) + }, + )?; + + if let Some(( + existing_session_id, + existing_message_id, + existing_part_type, + existing_text, + existing_hash, + existing_timestamp, + )) = existing.into_iter().next() + { + let incoming_part_type = row.part_type.to_string(); + if existing_session_id != row.session_id + || existing_message_id != row.message_id + || existing_part_type != incoming_part_type + || existing_text != row.text + || existing_hash != row.text_sha256 + || existing_timestamp != row.generated_at_unix_ms + { + bail!( + "message part integrity conflict for repository {repository_id}, source instance {source_instance_id}, source part {}", + row.source_part_id + ); + } + stats.already_present += 1; + } else { + txn.execute( + "INSERT INTO message_parts (repository_id, source_instance_id, session_id, message_id, source_part_id, part_type, text, text_sha256, generated_at_unix_ms) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", + ( + repository_id, + source_instance_id, + row.session_id.as_str(), + row.message_id.as_str(), + row.source_part_id, + row.part_type.to_string(), + row.text.as_str(), + row.text_sha256.as_str(), + row.generated_at_unix_ms, + ), + )?; + stats.inserted += 1; + } + + if fail_after_row == Some(index + 1) { + bail!( + "injected message parts destination failure after row {}", + index + 1 + ); + } + } + + upsert_watermark( + txn, + repository_id, + source_instance_id, + PARTS_SOURCE_TABLE, + last_row.source_part_id, + )?; + + Ok(PartsBatchStats { + inserted: stats.inserted, + already_present: stats.already_present, + watermark: stats.watermark, + }) + }) +} + +fn ensure_lineage( + txn: &TursoTransaction<'_, crate::services::agent_trace_dwh_db::AgentTraceDwhDbSpec>, + repository_id: &str, + source_instance_id: &str, +) -> Result<()> { + txn.execute( + "INSERT INTO repositories (repository_id) VALUES (?1) + ON CONFLICT (repository_id) DO NOTHING", + (repository_id,), + )?; + txn.execute( + "INSERT INTO source_instances (repository_id, source_instance_id) VALUES (?1, ?2) + ON CONFLICT (repository_id, source_instance_id) DO NOTHING", + (repository_id, source_instance_id), + )?; + Ok(()) +} + +impl Default for PartsEtl { + fn default() -> Self { + Self { + batch_size: DEFAULT_PARTS_ETL_BATCH_SIZE, + } + } +} + +impl PartsEtl { + /// Create a parts runner with a positive bounded source batch size. + pub fn with_batch_size(batch_size: u32) -> Result { + validate_batch_size(batch_size, "parts ETL")?; + Ok(Self { batch_size }) + } + + /// Return the configured source batch size. + pub fn batch_size(self) -> u32 { + self.batch_size + } + + /// Run the independently watermarked parts ETL through an open replica. + pub fn run( + self, + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, + ) -> Result { + let metadata = source + .verify_or_initialize_repository_metadata(repository_id) + .context("failed to verify Agent Trace source metadata")?; + run_with_destination( + self, + repository_id, + &metadata.source_instance_id, + source, + replica.db(), + ) + } +} + +pub(crate) fn run_with_destination( + config: PartsEtl, + repository_id: &str, + source_instance_id: &str, + source: &RepositoryAgentTraceDb, + destination: &AgentTraceDwhDb, +) -> Result { + let before_watermark = read_parts_watermark(destination, repository_id, source_instance_id)?; + let mut watermark = before_watermark; + let mut stats = PartsEtlStats { + before_watermark, + after_watermark: before_watermark, + ..Default::default() + }; + + loop { + let rows = extract_part_batch(source, watermark, config.batch_size)?; + if rows.is_empty() { + break; + } + + let batch = load_part_batch(destination, repository_id, source_instance_id, &rows)?; + watermark = batch.watermark; + stats.extracted += rows.len() as u64; + stats.inserted += batch.inserted; + stats.already_present += batch.already_present; + stats.batches += 1; + stats.after_watermark = watermark; + } + + Ok(stats) +} + +/// Run the default-sized parts ETL through an open DWH replica. +pub fn run_parts_etl( + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, +) -> Result { + PartsEtl::default().run(repository_id, source, replica) +} + +/// Read the parts watermark, treating an absent row as zero. +pub fn read_parts_watermark( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, +) -> Result { + read_watermark(db, repository_id, source_instance_id, PARTS_SOURCE_TABLE) +} + +#[cfg(test)] +mod tests { + use std::{ + fs, + path::{Path, PathBuf}, + sync::mpsc, + thread, + time::{SystemTime, UNIX_EPOCH}, + }; + + use super::*; + use crate::services::agent_trace_db::{InsertPartInsert, PartType}; + + fn unique_path(label: &str, file: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after Unix epoch") + .as_nanos(); + std::env::temp_dir() + .join(format!( + "sce-conversation-parts-{label}-{}-{nonce}", + std::process::id() + )) + .join(file) + } + + fn clean(path: &Path) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } + } + + fn source_row( + id: i64, + part_type: &str, + text: &str, + session_id: &str, + message_id: &str, + timestamp: i64, + ) -> SourceMessagePart { + SourceMessagePart { + id, + part_type: part_type.to_string(), + text: text.to_string(), + message_id: message_id.to_string(), + session_id: session_id.to_string(), + generated_at_unix_ms: timestamp, + } + } + + fn transformed_row( + id: i64, + part_type: PartType, + text: &str, + session_id: &str, + message_id: &str, + timestamp: i64, + ) -> TransformedMessagePart { + TransformedMessagePart { + source_part_id: id, + part_type, + text: text.to_string(), + text_sha256: sha256_hex(text.as_bytes()), + message_id: message_id.to_string(), + session_id: session_id.to_string(), + generated_at_unix_ms: timestamp, + } + } + + fn insert_part( + db: &RepositoryAgentTraceDb, + part_type: PartType, + text: &str, + session_id: &str, + message_id: &str, + timestamp: i64, + ) { + db.insert_part(InsertPartInsert { + part_type, + text: text.to_string(), + session_id: session_id.to_string(), + message_id: message_id.to_string(), + generated_at_unix_ms: timestamp, + }) + .expect("source part insert should succeed"); + } + + #[test] + fn conversation_parts_etl_extracts_ordered_bounded_batches_from_zero() { + let source_path = unique_path("extract", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + insert_part( + &source, + PartType::Text, + "one", + "session-1", + "message-1", + 1_000, + ); + insert_part( + &source, + PartType::Reasoning, + "two", + "session-1", + "message-1", + 1_001, + ); + insert_part( + &source, + PartType::Patch, + "three", + "session-1", + "message-1", + 1_002, + ); + + let rows = extract_part_batch(&source, 0, 2).unwrap(); + assert_eq!( + rows.iter().map(|row| row.id).collect::>(), + vec![1, 2] + ); + assert_eq!(rows[0].part_type, "text"); + assert_eq!(rows[1].part_type, "reasoning"); + assert!(extract_part_batch(&source, 3, 10).unwrap().is_empty()); + + clean(&source_path); + } + + #[test] + fn conversation_parts_etl_accepts_only_supported_types() { + for part_type in ["text", "reasoning", "patch", "question"] { + let transformed = transform_message_part(&source_row( + 1, + part_type, + "content", + "session-1", + "message-1", + 1_000, + )) + .unwrap(); + assert_eq!(transformed.part_type.to_string(), part_type); + } + + let error = transform_message_part(&source_row( + 1, + "tool", + "content", + "session-1", + "message-1", + 1_000, + )) + .expect_err("unknown part types must fail"); + assert!(error + .to_string() + .contains("unsupported source parts.type 'tool'")); + } + + #[test] + fn conversation_parts_etl_preserves_text_and_uses_lowercase_sha256() { + let text = "line 1\r\n\u{0000}unicode: café"; + let transformed = transform_message_part(&source_row( + 7, + "text", + text, + "session-1", + "message-1", + 1_000, + )) + .unwrap(); + assert_eq!(transformed.text, text); + assert_eq!(transformed.text_sha256, sha256_hex(text.as_bytes())); + assert_eq!( + transformed.text_sha256, + "5d6fb926a3bfcd8394b33e5a5aecaa23a5feb92d5d13a596818f929b26ee3221" + ); + } + + #[test] + fn conversation_parts_etl_inserts_lineage_content_and_watermark_without_parent() { + let dwh_path = unique_path("insert", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let rows = vec![ + source_row(1, "text", "hello", "session-1", "missing-message", 1_000), + source_row(2, "patch", "diff", "session-1", "missing-message", 1_001), + ]; + + let stats = load_part_batch(&dwh, "repo-a", "instance-a", &rows).unwrap(); + assert_eq!(stats.inserted, 2); + assert_eq!(stats.watermark, 2); + assert_eq!( + read_parts_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 2 + ); + let values = dwh + .query_map( + "SELECT source_instance_id, session_id, message_id, source_part_id, part_type, text, text_sha256 + FROM message_parts ORDER BY source_part_id", + (), + |row| { + Ok(( + row.get::(0)?, + row.get::(1)?, + row.get::(2)?, + row.get::(3)?, + row.get::(4)?, + row.get::(5)?, + row.get::(6)?, + )) + }, + ) + .unwrap(); + assert_eq!(values[0].0, "instance-a"); + assert_eq!(values[0].2, "missing-message"); + assert_eq!(values[0].4, "text"); + assert_eq!(values[0].5, "hello"); + assert_eq!(values[0].6, sha256_hex(b"hello")); + + clean(&dwh_path); + } + + #[test] + fn conversation_parts_identity_matching_replay_counts_already_present() { + let dwh_path = unique_path("replay", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let row = source_row(7, "text", "hello", "session-1", "message-1", 1_000); + + assert_eq!( + load_part_batch(&dwh, "repo-a", "instance-a", std::slice::from_ref(&row)) + .unwrap() + .inserted, + 1 + ); + let replay = load_part_batch(&dwh, "repo-a", "instance-a", &[row]).unwrap(); + assert_eq!(replay.already_present, 1); + assert_eq!( + dwh.query_map("SELECT COUNT(*) FROM message_parts", (), |db_row| db_row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![1] + ); + + clean(&dwh_path); + } + + #[test] + fn conversation_parts_identity_conflict_fails_without_overwrite() { + let dwh_path = unique_path("conflict", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let first = source_row(1, "text", "hello", "session-1", "message-1", 1_000); + load_part_batch(&dwh, "repo-a", "instance-a", &[first]).unwrap(); + + let error = load_part_batch( + &dwh, + "repo-a", + "instance-a", + &[source_row( + 1, + "text", + "changed", + "session-1", + "message-1", + 1_000, + )], + ) + .expect_err("changed source-lineage content must fail"); + assert!(error + .to_string() + .contains("message part integrity conflict")); + assert_eq!( + read_parts_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 1 + ); + assert_eq!( + dwh.query_map("SELECT text FROM message_parts", (), |row| row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![String::from("hello")] + ); + + clean(&dwh_path); + } + + #[test] + fn conversation_parts_identity_rolls_back_facts_dimensions_and_watermark() { + let dwh_path = unique_path("rollback", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let rows = vec![ + transformed_row(1, PartType::Text, "one", "session-1", "message-1", 1_000), + transformed_row(2, PartType::Patch, "two", "session-1", "message-1", 1_001), + ]; + + load_transformed_part_batch_with_failure(&dwh, "repo-a", "instance-a", &rows, Some(1)) + .expect_err("injected failure should roll back the complete batch"); + + for table in [ + "repositories", + "source_instances", + "message_parts", + "etl_watermarks", + ] { + assert_eq!( + dwh.query_map(&format!("SELECT COUNT(*) FROM {table}"), (), |row| row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![0], + "{table} should be rolled back" + ); + } + + let replay = load_part_batch( + &dwh, + "repo-a", + "instance-a", + &[ + source_row(1, "text", "one", "session-1", "message-1", 1_000), + source_row(2, "patch", "two", "session-1", "message-1", 1_001), + ], + ) + .unwrap(); + assert_eq!(replay.inserted, 2); + assert_eq!(replay.watermark, 2); + + clean(&dwh_path); + } + + #[test] + fn conversation_parts_same_local_ids_from_different_source_instances_coexist() { + let dwh_path = unique_path("lineage", "agent-trace-dwh.db"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).unwrap(); + let row = source_row(1, "text", "hello", "session-1", "message-1", 1_000); + + load_part_batch(&dwh, "repo-a", "instance-a", std::slice::from_ref(&row)).unwrap(); + load_part_batch(&dwh, "repo-a", "instance-b", &[row]).unwrap(); + + assert_eq!( + dwh.query_map( + "SELECT COUNT(*) FROM message_parts WHERE repository_id = 'repo-a'", + (), + |db_row| db_row.get::(0).map_err(Into::into) + ) + .unwrap(), + vec![2] + ); + clean(&dwh_path); + } + + #[test] + fn conversation_parts_etl_rejects_zero_batch_size() { + let source_path = unique_path("zero-batch", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + let error = extract_part_batch(&source, 0, 0).expect_err("zero batch must fail"); + assert!(error.to_string().contains("batch_size")); + clean(&source_path); + } + + #[test] + fn conversation_parts_etl_config_rejects_zero_batch_size() { + let error = PartsEtl::with_batch_size(0).expect_err("zero batch must fail"); + assert!(error.to_string().contains("batch_size")); + } + + #[test] + fn concurrent_source_writer_is_not_blocked_by_parts_read_transaction() { + let source_path = unique_path("concurrent-writer", "agent-trace.db"); + let source = RepositoryAgentTraceDb::new_at(&source_path).unwrap(); + insert_part( + &source, + PartType::Text, + "first", + "session-1", + "message-1", + 1_000, + ); + + let (reader_ready_tx, reader_ready_rx) = mpsc::channel::<()>(); + let (release_reader_tx, release_reader_rx) = mpsc::channel::<()>(); + let reader_db_path = source_path.clone(); + let reader_handle = thread::spawn(move || { + let reader_db = RepositoryAgentTraceDb::open_without_migrations_at(&reader_db_path) + .expect("reader connection should reopen"); + reader_db.read_transaction(|txn| { + let rows = txn.query_map( + SELECT_PARTS_BATCH_SQL, + (0i64, 10i64), + source_message_part_from_row, + )?; + reader_ready_tx + .send(()) + .expect("reader should signal it holds an open read transaction"); + release_reader_rx + .recv() + .expect("reader should wait to be released while holding the transaction open"); + Ok(rows) + }) + }); + + reader_ready_rx + .recv() + .expect("test should observe the parts reader transaction"); + let writer_db = RepositoryAgentTraceDb::open_without_migrations_at(&source_path) + .expect("writer connection should reopen"); + insert_part( + &writer_db, + PartType::Question, + "second", + "session-1", + "message-1", + 1_001, + ); + + release_reader_tx + .send(()) + .expect("reader should be released"); + let rows = reader_handle + .join() + .expect("reader thread should not panic") + .expect("reader transaction should commit"); + assert_eq!(rows.len(), 1); + + clean(&source_path); + } +} diff --git a/cli/src/services/etl.rs b/cli/src/services/etl.rs new file mode 100644 index 00000000..4f918759 --- /dev/null +++ b/cli/src/services/etl.rs @@ -0,0 +1,105 @@ +//! Small mechanics shared by incremental source-to-DWH table ETLs. +//! +//! Row extraction, transformation, identity validation, and loading remain +//! table-specific. This module only owns the bounded retry, configuration, +//! watermark, and common batch-accounting seams. + +use std::thread; + +use anyhow::{ensure, Result}; + +use crate::services::{ + agent_trace_dwh_db::{AgentTraceDwhDb, AgentTraceDwhDbSpec}, + db::TursoTransaction, + resilience::RetryPolicy, +}; + +/// Bounded backoff for short source read transactions. Only transient source +/// contention is retried; extraction and mapping errors fail immediately. +pub(crate) const SOURCE_CONTENTION_RETRY_POLICY: RetryPolicy = RetryPolicy { + max_attempts: 5, + timeout_ms: 1_000, + initial_backoff_ms: 25, + max_backoff_ms: 200, +}; + +/// Statistics for one atomic destination batch, shared by table runners. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) struct TableBatchStats { + pub inserted: u64, + pub already_present: u64, + pub watermark: i64, +} + +/// Validate a positive source batch size while preserving the caller's +/// table-specific diagnostic prefix. +pub(crate) fn validate_batch_size(batch_size: u32, operation: &str) -> Result<()> { + ensure!( + batch_size > 0, + "{operation} batch_size must be greater than zero" + ); + Ok(()) +} + +/// Run a short source read with bounded retry for transient lock contention. +/// `before_retry` clears a failed read transaction before the next `BEGIN`. +pub(crate) fn run_with_source_contention_retry( + mut operation: impl FnMut(u32) -> Result, + mut before_retry: impl FnMut(), +) -> Result { + let mut attempt = 1; + + loop { + match operation(attempt) { + Ok(value) => return Ok(value), + Err(error) => { + let attempts_remain = attempt < SOURCE_CONTENTION_RETRY_POLICY.max_attempts; + if !attempts_remain || !is_transient_source_contention(&error) { + return Err(error); + } + + before_retry(); + thread::sleep(SOURCE_CONTENTION_RETRY_POLICY.backoff_for_attempt(attempt + 1)); + attempt += 1; + } + } + } +} + +/// Recognize only the transient lock errors emitted by source reads. +pub(crate) fn is_transient_source_contention(error: &anyhow::Error) -> bool { + let message = error.to_string().to_lowercase(); + message.contains("database is locked") || message.contains("table is locked") +} + +/// Read a table watermark, treating an absent row as zero. +pub(crate) fn read_watermark( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + source_table: &str, +) -> Result { + db.query_map( + "SELECT COALESCE(last_extracted_source_row_id, 0) FROM etl_watermarks WHERE repository_id = ?1 AND source_instance_id = ?2 AND source_table = ?3", + (repository_id, source_instance_id, source_table), + |row| row.get::(0).map_err(Into::into), + )? + .into_iter() + .next() + .map_or(Ok(0), Ok) +} + +/// Atomically upsert a table watermark in the caller's destination +/// transaction. +pub(crate) fn upsert_watermark( + txn: &TursoTransaction<'_, AgentTraceDwhDbSpec>, + repository_id: &str, + source_instance_id: &str, + source_table: &str, + watermark: i64, +) -> Result { + txn.execute( + "INSERT INTO etl_watermarks (repository_id, source_instance_id, source_table, last_extracted_source_row_id) VALUES (?1, ?2, ?3, ?4) ON CONFLICT (repository_id, source_instance_id, source_table) DO UPDATE SET last_extracted_source_row_id = excluded.last_extracted_source_row_id, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + (repository_id, source_instance_id, source_table, watermark), + ) +} diff --git a/cli/src/services/mod.rs b/cli/src/services/mod.rs index 9cb883a5..80d8c20a 100644 --- a/cli/src/services/mod.rs +++ b/cli/src/services/mod.rs @@ -19,10 +19,17 @@ pub mod checkout; pub mod command_registry; pub mod completion; pub mod config; +#[allow(dead_code)] +pub mod conversation_etl; +#[allow(dead_code)] +pub mod conversation_messages_etl; +#[allow(dead_code)] +pub mod conversation_parts_etl; pub mod db; pub mod default_paths; pub mod doctor; pub mod error; +pub(crate) mod etl; pub mod help; pub mod hooks; pub mod lifecycle; diff --git a/context/architecture.md b/context/architecture.md index e7453c64..01804300 100644 --- a/context/architecture.md +++ b/context/architecture.md @@ -125,7 +125,7 @@ The repository includes a new placeholder Rust binary crate at `cli/`. - `cli/src/services/agent_trace_db/mod.rs` owns the shared Agent Trace insert payloads, SQL constants, and typed row helpers (diff-trace/intersection/Agent Trace/message/part) plus `ensure_schema_ready_for_hooks()` consumed by the repository adapter. `cli/src/services/agent_trace_db/repository.rs` defines the sole `RepositoryAgentTraceDb` adapter over `TursoDb` with a fresh `agent-trace-repository/001_repository_schema.sql` baseline for `diff_traces` (including `payload_type`), `post_commit_patch_intersections`, `agent_traces`, `messages`, `parts`, indexes, and triggers, plus the additive `002_repository_source_instance_id.sql` migration adding `repository_metadata.source_instance_id`, `repository_metadata` validation via typed `RepositoryMetadata { repository_id, source_instance_id }`, no trace-table `checkout_id` columns, `agent_traces.agent_trace_id NOT NULL UNIQUE`, and `recent_diff_trace_patches(cutoff_time_ms, end_time_ms)` using the inclusive chronological parser without checkout filtering. Active hook runtime, setup/lifecycle storage, and trace status/list/shell flows resolve through `agent_trace_storage` and use `RepositoryAgentTraceDb`. The checkout-scoped `AgentTraceDb`/`AgentTraceDbSpec` adapter, its `agent_trace_db_path()`/`agent_trace_db_path_for_checkout()` helpers, the 15-file `cli/migrations/agent-trace/` chain, and the `sce trace --legacy` surface were removed by the `retire-legacy-agent-trace-db` plan. - `cli/src/services/agent_trace_dwh_db/mod.rs` defines `AgentTraceDwhDb = TursoDb`, a separate append-oriented destination-schema adapter for the CLI-independent Agent Trace ETL consumer, backed by a fresh `agent-trace-dwh/001_dwh_schema.sql` baseline (`repositories`, `source_instances`, `etl_watermarks`, `messages`, `message_parts`, `agent_traces`, `code_changes`, no foreign keys). Explicit-path only, reuses the `agent_trace_db` retry config key, and is not wired into any lifecycle provider, doctor/setup flow, or CLI command yet. See `context/sce/agent-trace-dwh-db.md` and the accepted decision at `context/decisions/2026-08-08-agent-trace-dwh-schema-identity-contract.md`. - `cli/src/services/agent_trace_dwh_replica/mod.rs` defines `AgentTraceDwhReplica`, the sole owner of a Turso Sync connection to the repository-scoped, single-owner `agent-trace-sync.db` replica at `/sce/repos//agent-trace-sync.db` (distinct from both the multiprocess-WAL source `agent-trace.db` and the DWH's own explicit-path `AgentTraceDwhDbSpec`). `open()` acquires a non-blocking `BridgeLock` (`lock.rs`, over the sibling `.bridge-lock` file) before any Turso access, opens the local file through `turso::sync::Builder` without ever enabling `experimental_multiprocess_wal`, and wraps the result as an `AgentTraceDwhDb` via a narrow `TursoDb::from_connection`/`block_on` seam in `cli/src/services/db/mod.rs` to classify DWH schema state (`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 the open loudly without repair. It exposes lock-lifetime-bound SQL access plus `run_agent_trace_etl()`, `pull()`, and `push()`; the ETL method delegates bounded source extraction and atomic fact/watermark loading while preserving lock ownership, and pull/push remain separate explicit operations. It also redacts the caller-supplied auth token from every error. Callers provide `local_path`/`database_url`/`auth_token` explicitly; no credential discovery/persistence or CLI/lifecycle wiring exists yet. See `context/sce/agent-trace-dwh-replica.md` and the accepted decision at `context/decisions/2026-08-08-agent-trace-dwh-empty-remote-auto-initialization.md` (superseding `context/decisions/2026-08-08-agent-trace-dwh-turso-sync-replica-ownership.md`). -- `cli/src/services/agent_trace_etl/mod.rs` defines `AgentTraceEtl`, the CLI-independent incremental bridge from the repository source of truth to the DWH replica. It validates source metadata, uses short non-blocking `agent_traces` snapshots and bounded source-contention retry, then commits every fact/dimension batch and its `(repository_id, source_instance_id, agent_traces)` watermark in one local destination transaction. The replica's `run_agent_trace_etl()` preserves destination lock ownership; remote pull/push is separate, and the local replica is reconstructible through replay. +- `cli/src/services/agent_trace_etl/mod.rs` defines `AgentTraceEtl`, the CLI-independent incremental bridge from the repository source of truth to the DWH replica. It validates source metadata, uses short non-blocking `agent_traces` snapshots and bounded source-contention retry, then commits every fact/dimension batch and its `(repository_id, source_instance_id, agent_traces)` watermark in one local destination transaction. The sibling `conversation_messages_etl.rs`, `conversation_parts_etl.rs`, and `conversation_etl.rs` reuse those mechanics for logical messages and source-lineage-scoped parts, with separate `messages`/`parts` watermarks, exact text hashing, conflict checks, and no parent-message foreign key. The replica's `run_agent_trace_etl()` and conversation runner preserve destination lock ownership; remote pull/push is separate, and the local replica is reconstructible through replay. The source audit found only schema-maintenance `updated_at` trigger bodies issuing SQL `UPDATE messages`/`UPDATE parts`; active production capture writes those tables through append-oriented insert helpers, so synchronized fields are treated as immutable after insertion and have no update CDC. - `cli/src/services/setup/mod.rs` defines the setup command contract (`SetupMode`, `SetupTarget`, `SetupRequest`, CLI flag parser/validator), an `inquire`-backed interactive target prompter (`InquireSetupTargetPrompter`), setup dispatch outcomes (proceed/cancelled), additive durable-context bootstrap (`bootstrap_context_baseline` for standalone `--bootstrap-context` and every normal successful setup path), and compile-time embedded asset access (`EmbeddedAsset`, target-scoped iterators, required-hook asset iterators/lookups). It also owns the install-time optional-workflow seam: the `OptionalWorkflow` type plus the build-generated `OPTIONAL_WORKFLOWS` catalog, a per-target `WorkflowAssetLayout` built from the existing `default_paths` command/skill directory constants (`command`/`commands`/`prompts` plus `skills`), and `iter_embedded_assets_for_setup_target_with_selection(target, selection)`, which yields every embedded asset except the `{command_dir}/{command_slug}.md` file and `{skills_dir}/{skill_slug}/` subtree of each optional workflow the selection omits. Membership is derived from the catalog's slugs rather than an enumerated file list, so a new optional workflow needs no Rust change. This filtered iterator is the only way embedded assets are enumerated; setup installs through it and doctor inspects through it, so there is no unfiltered enumeration path that could reintroduce an unselected workflow. The non-interactive selection flows through the repeatable `--workflow ` flag into `SetupRequest.optional_workflows: Option>` (`None` meaning the flag was absent); `validate_optional_workflow_slugs` checks each slug against `OPTIONAL_WORKFLOWS` during request resolution, before any write, and `run_setup_for_mode` resolves `None` to the persisted `integrations.optional_workflows` (exported as `persisted_optional_workflows`) before installing through the filtered iterator and persisting the resolved selection. The interactive selection flows through the same seam: `SetupTargetPrompter` carries `prompt_target` plus `prompt_optional_workflows(defaults)` (returning `None` for a cancelled prompt), `SetupDispatch::Proceed { mode, optional_workflows }` carries a prompted selection alongside the resolved mode, and `resolve_setup_dispatch(mode, prompter, defaults)` runs the workflow prompt only after an interactive target prompt, mapping either cancellation to `SetupDispatch::Cancelled`. The prompt module builds its `inquire::MultiSelect` from `optional_workflow_prompt_inputs(catalog, defaults)`, which returns `None` for an empty catalog (skipping the prompt) and otherwise catalog-ordered rows plus the indices to pre-check, ignoring ids absent from the catalog. `setup/command.rs` therefore resolves the repository root before dispatch, so a non-git directory fails before any prompt. For repository builds, `cli/build.rs` validates the `SCE_CLI_GENERATED_INPUT_DIR` payload and canonical-input inventories, copies the payload into Cargo `OUT_DIR/pkl-generated`, stages `cli/assets/hooks/**` under `OUT_DIR/static`, requires the staged `config/optional-workflows.json`, and generates both the setup manifest and the optional-workflow catalog (`optional_workflows.rs`, rejecting a manifest whose `schemaVersion` is not 1 or whose entries lack a non-empty `id`/`title`/`description`/`commandSlug`/`skillSlug`) in `OUT_DIR`; focused internal seams separate install-flow from prompt-flow logic; `cli/src/services/setup/command.rs` owns the `SetupCommand` payload used by the static `RuntimeCommand` enum and executes against any context implementing repo-root scoping. Its install engine/orchestrator stages embedded files and uses a unified remove-and-replace policy (removing existing targets before swapping staged content, with deterministic recovery guidance on swap failure and no backup artifact creation), and formats deterministic completion messaging; required-hook install orchestration (`install_required_git_hooks`) follows the same remove-and-replace policy (removing existing hooks before swapping staged content, with deterministic recovery guidance on swap failure). After the Git gate, setup always ensures the context baseline; context-only requests return there, while normal modes derive a repo-root-scoped context before aggregating static lifecycle provider `setup` dispatch across providers (config → local_db → auth_db → agent_trace_db → hooks when requested), so setup providers consume only repo-root access from the scoped context. - `cli/src/services/setup/mod.rs` keeps those responsibilities inside one file for now, but the current ownership split is explicit: the inline `install` module owns repository-path normalization, staging/swap install behavior, required-hook installation, and filesystem safety guards, while the inline `prompt` module owns interactive target selection and prompt styling. - `cli/src/services/security.rs` provides shared security utilities for deterministic secret redaction (`redact_sensitive_text`) and directory write-permission probes (`ensure_directory_is_writable`) used by app/setup/observability surfaces. diff --git a/context/context-map.md b/context/context-map.md index ffc41adf..05dab237 100644 --- a/context/context-map.md +++ b/context/context-map.md @@ -64,7 +64,10 @@ Feature/domain context: - `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()` 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-db.md` (Agent Trace DWH: a separate append-oriented destination schema for a future 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` (Agent Trace ETL bridge between the `agent-trace.db` source and the DWH replica destination; covers the bounded `AgentTraceEtl` run loop, exact-JSON lowercase SHA-256 transformation, idempotent logical-fact loading, atomic per-lineage watermark advancement, source contention handling, stats, and replica-owned orchestration through `cli/src/services/agent_trace_etl/mod.rs`) +- `context/sce/agent-trace-etl.md` (shared Agent Trace ETL mechanics and the `agent_traces`, `messages`, and `parts` 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, and replica-owned orchestration) +- `context/sce/conversation-messages-etl.md` (independently watermarked logical-message pipeline: short ordered source snapshots, validated roles, repository/source lineage, logical replay/conflict handling, atomic DWH facts plus watermark, and CLI-independent replica usage) +- `context/sce/conversation-parts-etl.md` (independently watermarked source-lineage-scoped parts pipeline: short ordered snapshots, supported `PartType` values, verbatim text and lowercase SHA-256, replay/conflict validation, no-parent loading, deterministic ordering, and atomic DWH watermarking) +- `context/sce/conversation-etl.md` (conversation-level `ConversationEtl` composition API: replica-owned sequential table runs, table-level stats, independently configurable batches, separate progress, parts-before-messages validity, deterministic part reconstruction, and no transport or control-plane ownership) - `context/sce/agent-trace-core-schema-migrations.md` (historical reference for removed local DB schema bootstrap behavior; T03 now implements the actual local DB with migrations) - `context/sce/agent-trace-retry-queue-observability.md` (inactive local-hook retry path plus historical retry/metrics reference) - `context/sce/agent-trace-local-hooks-mvp-contract-gap-matrix.md` (T01 Local Hooks MVP production contract freeze and deterministic gap matrix for `agent-trace-local-hooks-production-mvp`) @@ -115,3 +118,4 @@ Recent decision records: - `context/decisions/2026-08-08-agent-trace-dwh-schema-identity-contract.md` (introduces the separate append-oriented Agent Trace DWH destination schema and explicit-path-only adapter, distinct from the repository-scoped source schema, with the two-scope logical-vs-local-ID identity contract for idempotent re-ingestion from independently created source databases) - `context/decisions/2026-08-08-agent-trace-dwh-turso-sync-replica-ownership.md` (single-owner, lock-guarded, disposable Turso Sync replica boundary for the Agent Trace DWH: `AgentTraceDwhReplica` is the sole Turso Sync builder owner, never enables multiprocess WAL, never discovers/persists credentials, and is always reconstructible from the remote; superseded by `2026-08-08-agent-trace-dwh-empty-remote-auto-initialization.md`'s empty-remote auto-initialization guardrail) - `context/decisions/2026-08-08-agent-trace-dwh-empty-remote-auto-initialization.md` (supersedes the prior verify-only guardrail: `AgentTraceDwhReplica::open()` may auto-initialize a genuinely `Empty` remote DWH schema via `AgentTraceDwhDb::classify_schema_state()` and `run_migrations()` + `push()`, with a narrow one-`pull()`-and-re-verify push-conflict recovery; still never repairs an `Incompatible` schema; SCE remains sole owner of the DWH schema/migrations, control-plane owns only remote provisioning and credentials) +- `context/decisions/2026-08-08-conversation-etl-append-only-watermarks.md` (conversation messages/parts use independent integer-ID watermarks and treat synchronized source fields as append-only for ETL purposes; update CDC requires an explicit future design) diff --git a/context/decisions/2026-08-08-conversation-etl-append-only-watermarks.md b/context/decisions/2026-08-08-conversation-etl-append-only-watermarks.md new file mode 100644 index 00000000..b7ee9b8a --- /dev/null +++ b/context/decisions/2026-08-08-conversation-etl-append-only-watermarks.md @@ -0,0 +1,85 @@ +# Decision: Treat conversation source fields as append-only under table-specific ID watermarks + +Date: 2026-08-08 +Status: Accepted +Plan: `context/plans/incremental-conversation-etl.md` +Task: T02, T03, T04, T05 + +## Context + +The repository-scoped Agent Trace source stores conversation messages and parts +that are incrementally copied into the Agent Trace DWH. The source uses local +integer IDs, while the DWH must preserve source lineage, tolerate parts arriving +before messages, and allow messages and parts to advance independently. The +completed ETL implementation has no safe way to discover historical edits after +an integer-ID watermark, and source inspection found only schema-maintenance +`updated_at` trigger bodies issuing `UPDATE messages` or `UPDATE parts`; active +capture uses insert helpers. + +## Decision + +Treat the synchronized message role/timestamp and part session/message/type/text/ +hash/timestamp fields as append-only and immutable for ETL purposes, using +independent integer-ID watermarks for the `messages` and `parts` source tables; +update CDC is deliberately out of scope. + +## Rationale + +Table-specific watermarks let either pipeline progress without coupling its +transaction or progress state to the other table, while append-only semantics +make the cursor complete and deterministic. The source audit supports this +assumption, and an intentional future source update must be addressed by an +explicit CDC design or architecture decision rather than silently being missed. + +## Alternatives considered + +- **Add update CDC or rescan rows at or below the watermark** — not selected + because no update requirement exists in the current capture path and it would + add a separate change-tracking contract to this append-oriented bridge. +- **Use one conversation-level watermark or transaction** — not selected + because messages and parts have different source tables, local IDs, and valid + out-of-order ingestion requirements. +- **Use timestamp cursors** — not selected because timestamps are not unique and + cannot provide deterministic, lossless progress. + +## Compatibility and risks + +- Existing source writers remain compatible because they append messages and + parts; the DWH contract does not alter the live-capture schema or writers. +- A future intentional update to synchronized fields would be ignored after its + row ID is watermarked unless update CDC or an explicit replay strategy is + designed; this risk is mitigated by documenting the assumption and requiring + an architectural decision before changing it. + +## Guardrails + +- Keep `messages` and `parts` watermarks separate and keyed by repository, + source instance, and source table. +- Preserve source row ordering by integer ID, and do not replace it with a + timestamp-only cursor. +- Do not add a parent-message foreign key or couple parts ingestion to message + existence. +- Do not introduce update CDC implicitly; future source updates require an + explicit design and validation contract. + +## Consequences + +- Message and part ETL runs can progress independently and replay complete + failed batches atomically. +- Parts may be loaded before their parent message, and source-lineage local IDs + remain safe across independently created source databases. +- Historical updates to rows at or below a committed watermark are outside the + current ETL guarantee. + +## Follow-up + +None. + +## References + +- Plan: [`incremental-conversation-etl`](../plans/incremental-conversation-etl.md) +- Task: `T02, T03, T04, T05` +- Current-state context: [`agent-trace-etl.md`](../sce/agent-trace-etl.md), [`conversation-messages-etl.md`](../sce/conversation-messages-etl.md), [`conversation-parts-etl.md`](../sce/conversation-parts-etl.md), [`conversation-etl.md`](../sce/conversation-etl.md) +- Evidence: [`incremental-conversation-etl.md`](../plans/incremental-conversation-etl.md), [`conversation_messages_etl.rs`](../../cli/src/services/conversation_messages_etl.rs), [`conversation_parts_etl.rs`](../../cli/src/services/conversation_parts_etl.rs) +- Related decision: [`Separate Agent Trace DWH destination schema with a source-instance-scoped identity contract`](2026-08-08-agent-trace-dwh-schema-identity-contract.md) +- Related decision: [`Single-owner, disposable Turso Sync replica boundary for the Agent Trace DWH`](2026-08-08-agent-trace-dwh-turso-sync-replica-ownership.md) diff --git a/context/glossary.md b/context/glossary.md index edcb391f..bbad2dbd 100644 --- a/context/glossary.md +++ b/context/glossary.md @@ -97,6 +97,11 @@ - `__sce_migrations`: Per-database migration metadata table created by the shared `TursoConnectionCore` migration path behind public adapter `run_migrations()` methods; records applied migration IDs after successful execution so later setup/lifecycle initialization applies only migrations not yet recorded, while existing metadata-less DBs are brought forward by re-applying the current idempotent migration set and recording each ID. - `CLI generated migration manifest`: Build-time Rust source at `OUT_DIR/generated_migrations.rs` written by `cli/build.rs` from immediate `cli/migrations//*.sql` directories after staging SQL under `OUT_DIR/static/migrations`; constants are named from the database directory (for example `AGENT_TRACE_REPOSITORY_MIGRATIONS`, `AUTH_MIGRATIONS`), sorted by the numeric filename prefix before `_`, and embed staged SQL via `include_str!`. - `AgentTraceEtl`: CLI-independent incremental `agent_traces` bridge that verifies repository metadata, extracts bounded source-ID batches from short plain-`BEGIN` snapshots, transforms exact JSON with lowercase SHA-256, and atomically loads facts plus the per-lineage DWH watermark. It reports batch accounting and before/after watermarks, retries only transient source contention, and never owns credentials or remote pull/push. +- `ConversationEtl`: CLI-independent composition API in `cli/src/services/conversation_etl.rs` that runs the existing messages and parts table bridges through an already-open DWH replica and returns table-level stats. It verifies source metadata once, keeps table transactions and watermarks independent, supports separate batch sizes, and owns no credentials, pull/push, scheduling, or CLI orchestration. +- `MessagesEtl`: CLI-independent incremental logical-message bridge in `cli/src/services/conversation_messages_etl.rs`. It validates source metadata and `MessageRole`, reads bounded `messages` rows from short plain-`BEGIN` snapshots, uses a separate `(repository_id, source_instance_id, "messages")` watermark, treats matching logical replays as already present, rejects role/timestamp conflicts, and atomically commits message facts plus lineage dimensions and watermark without owning pull/push or CLI orchestration. +- `PartsEtl`: CLI-independent incremental message-part bridge in `cli/src/services/conversation_parts_etl.rs`. It validates the four supported `PartType` values, reads bounded `parts` rows from short plain-`BEGIN` snapshots, uses a separate `(repository_id, source_instance_id, "parts")` watermark, preserves exact UTF-8 text with lowercase SHA-256, validates source-lineage replays/conflicts, and atomically commits `message_parts` facts plus lineage dimensions and watermark without requiring a parent message or owning pull/push. +- `source-part identity`: DWH message-part identity `(repository_id, source_instance_id, source_part_id)`, where `source_part_id` is the source repository database's local `parts.id`. The source instance scopes local IDs so independently created source databases can coexist, while replay validation prevents a changed session, message, type, text, hash, or timestamp from overwriting an existing part. +- `conversation ETL source immutability`: The messages and parts ETL cursors are append-only integer-ID watermarks. After a source row is inserted, its synchronized role/timestamp or part session/message/type/text/hash/timestamp is treated as immutable for ETL purposes; repository source code currently writes these rows through insert helpers, and the only `UPDATE messages`/`UPDATE parts` matches are schema-maintenance `updated_at` triggers. Update CDC is therefore intentionally out of scope. - `sync command deferral`: Current plan/state note that a user-invocable `sce sync` command is not wired yet and is deferred to `0.4.0`; local DB bootstrap and setup-time repository-scoped Agent Trace DB initialization flow through lifecycle providers aggregated by the setup command, hook runtime instead requires `sce setup` to have already prepared the repository Agent Trace DB (it never creates or migrates one), and DB health/repair flows through the doctor surface. - `CLI bounded resilience wrapper`: Shared policy in `cli/src/services/resilience.rs` (`RetryPolicy`, async `run_with_retry`, sync `run_with_retry_sync`) that applies deterministic retries/timeouts/capped backoff to transient operations, emits retry observability events, and returns actionable terminal failure guidance. The sync helper is currently wired into shared database constructors for local open/connect retry and into `TursoDb`/`EncryptedTursoDb` operation retry for `execute()`/`query()`/`query_map()`. - `setup service orchestration`: Setup execution logic in `cli/src/services/setup/command.rs` that resolves the repository root, always ensures the durable-context baseline, then for normal modes derives a repo-root-scoped `AppContext` from the runtime command context, dispatches `setup` through the static lifecycle provider catalog (config → local_db → auth_db → agent_trace_db → hooks when requested), handles interactive target selection for config asset installation, and emits deterministic success messaging per target. @@ -241,5 +246,4 @@ - `agent-trace plugin secondary diff persistence ownership`: Current runtime contract where `buildTrace` no longer writes diff-trace artifacts or database rows directly; extracted diff payloads are forwarded to CLI `diff-trace` intake and the Rust hook runtime owns AgentTraceDb insertion without any `context/tmp` artifact fallback. - `messages table (Agent Trace DB)`: Agent Trace DB table created by migration `008_create_messages.sql`; stores session-scoped parent messages with columns `session_id`, `message_id`, `role` (`user`/`assistant` via CHECK constraint), `generated_at_unix_ms`, `created_at`, and `updated_at`. Message body text belongs to `parts.text`, not the parent `messages` row. Has a unique index on `(session_id, message_id)` for duplicate-ignore parent message inserts and a compound index on `(session_id, generated_at_unix_ms, id)` for chronological session message retrieval. No foreign keys to any other table. - `musl static Linux release`: The Linux binary release targets (`x86_64-unknown-linux-musl` and `aarch64-unknown-linux-musl`) compile against musl libc and link fully statically. The resulting binary has no runtime libc dependency and zero `/nix/store/` references in ELF metadata, strings, or dynamic-linker fields, satisfying the native portability audit. The musl targets replace the previous glibc-linked `*-unknown-linux-gnu` targets; macOS (`aarch64-apple-darwin`) is unchanged. Introduced in the `musl-static-linux-release` plan. - - `parts table (Agent Trace DB)`: Agent Trace DB table created by migration `009_create_parts.sql`; stores append-only message parts with columns `type` (typed by Rust as `text`/`reasoning`/`patch`/`question` and stored as unconstrained `TEXT NOT NULL`), `text`, `message_id`, `session_id`, `generated_at_unix_ms`, `created_at`, `updated_at`. Uses only the internal `id` for row identity (no upsert/dedup). Multiple parts can exist for the same `(session_id, message_id)`. A compound index on `(session_id, message_id, generated_at_unix_ms, id)` enables ordered joins. No foreign keys to `messages` or any other table, so parts may be inserted before their parent message exists. diff --git a/context/overview.md b/context/overview.md index a8ab2ff1..04ef7566 100644 --- a/context/overview.md +++ b/context/overview.md @@ -134,4 +134,4 @@ Lightweight post-task verification baseline (required after each completed task) - Use `context/sce/setup-githooks-hook-asset-packaging.md` for the implemented `sce-setup-githooks-any-repo` T02 compile-time hook-template packaging contract and setup-service required-hook embedded accessor surface. - Use `context/sce/setup-githooks-install-flow.md` for the implemented `sce-setup-githooks-any-repo` T03 required-hook install orchestration contract (git-truth hooks-path resolution, per-hook installed/updated/skipped outcomes, and remove-and-replace behavior). - Use `context/sce/setup-githooks-cli-ux.md` for the implemented `sce-setup-githooks-any-repo` T04 setup command-surface contract (`--hooks`, optional `--repo`), compatibility validation rules, and deterministic hook setup messaging. -- Use `context/sce/agent-trace-etl.md` for the production `AgentTraceEtl` bridge: repository source metadata and short non-blocking extraction snapshots feed exact-JSON, hash-verified Agent Trace facts and atomically advanced lineage watermarks in the lock-owned DWH replica; remote pull/push remains separate. +- Use `context/sce/agent-trace-etl.md` for the production `AgentTraceEtl` bridge and its conversation siblings: repository source metadata and short non-blocking extraction snapshots feed exact-content Agent Trace, message, and part facts with atomically advanced lineage watermarks in the lock-owned DWH replica; remote pull/push remains separate. Messages and parts use independent integer-ID cursors and are treated as append-only/immutable after source insertion for ETL purposes; update CDC is intentionally absent. diff --git a/context/plans/incremental-conversation-etl.md b/context/plans/incremental-conversation-etl.md new file mode 100644 index 00000000..00145018 --- /dev/null +++ b/context/plans/incremental-conversation-etl.md @@ -0,0 +1,153 @@ +# Plan: incremental-conversation-etl + +## Change summary + +Extend the existing CLI-independent Agent Trace ETL bridge with two independently watermarked conversation pipelines: repository-scoped Agent Trace `messages` rows into DWH `messages`, and source `parts` rows into DWH `message_parts`. Both pipelines will reuse the PR4 short-source-snapshot, bounded contention retry, destination transaction, watermark, and stats mechanics while preserving source lineage, logical message identity, supported part types, exact part text, SHA-256 hashes, and deterministic ordering. + +Expose a `ConversationEtl` runner and table-level stats without coupling the ETLs to pull/push, credentials, CLI orchestration, control-plane behavior, or the deferred code-change pipeline. The existing DWH baseline migration remains unchanged because it already contains the required destination columns and identity indexes. The source fields synchronized by these high-watermark pipelines are treated as append-only/immutable for ETL purposes; update CDC is deliberately not added. + +## Acceptance criteria + +How this plan is proven complete. Each criterion is observable and names the check that proves it. `/validate` runs these checks; no task in the stack performs final validation. + +- [x] AC1: Shared ETL mechanics used by `agent_traces`, `messages`, and `parts` include bounded source-contention retry, absent-watermark-as-zero reads, validated batch sizes, atomic watermark upserts, and common batch accounting without introducing a broad generic ETL trait hierarchy; existing Agent Trace ETL behavior remains unchanged. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_etl` +- [x] AC2: The messages ETL extracts `id, session_id, message_id, role, generated_at_unix_ms` with `id > watermark ORDER BY id ASC LIMIT batch_size` in a short read transaction, obtains and validates `source_instance_id` through repository metadata, and atomically loads each batch with the `messages` watermark. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_etl` +- [x] AC3: Message destination identity is `(repository_id, session_id, message_id)`; same role and timestamp replay is counted as `already_present`, while a differing role or `generated_at_unix_ms` fails with a deterministic integrity conflict and rolls back the complete batch and watermark. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_identity` +- [x] AC4: The parts ETL extracts `id, type, text, message_id, session_id, generated_at_unix_ms` incrementally in a short source read transaction, accepts exactly `text`, `reasoning`, `patch`, and `question` through the existing `PartType` representation, rejects other values explicitly, preserves text verbatim, and stores lowercase hexadecimal SHA-256 of the exact UTF-8 bytes. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_etl` +- [x] AC5: Part destination identity is `(repository_id, source_instance_id, source_part_id)`; matching `session_id`, `message_id`, `part_type`, `text_sha256`, and timestamp is an idempotent replay, any mismatch fails loudly and rolls back the batch, and parts can load without a parent message row. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_identity` +- [x] AC6: A conversation-level API exposes independently configurable/default-batched message and part runs with stats matching existing ETL conventions; messages and parts have separate `(repository_id, source_instance_id, source_table)` watermarks, may progress independently, and no conversation-level transaction or shared watermark couples them. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl` +- [x] AC7: End-to-end tests prove initial and incremental sync, no-op reruns, logical message replay/conflict, exact text/hash and supported part types, equal-timestamp ordering by `source_part_id`, source-lineage ID collisions across source instances, part-before-message ingestion, independent watermarks, source contention with concurrent writers, and injected mid-batch rollback followed by successful replay for both tables. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl` +- [x] AC8: Durable context documents the messages/parts DWH pipeline, separate watermarks, logical message identity, source-scoped part identity, verbatim text and hashing, deterministic ordering, absence of a message-to-part foreign-key requirement, append-only source assumption, and pull/push as orchestration concerns; source inspection confirms no existing production code intentionally updates synchronized message/part fields beyond schema-maintenance triggers. + - Validate: inspect the updated conversation ETL, DWH, replica, shared ETL, root architecture/context-map/glossary files and the source `messages`/`parts` writers against the implementation; confirm only `updated_at` triggers issue SQL `UPDATE` for these tables. + +### Full validation + +Repository-wide checks `/validate` runs after the last task, regardless of which criterion they map to. + +- `nix run .#pkl-check-generated` +- `nix flake check` + +### Context sync + +- Update `context/overview.md`, `context/architecture.md`, and `context/glossary.md` with the conversation ETL boundary and independent watermark/identity terminology. +- Extend `context/sce/agent-trace-etl.md`, `context/sce/agent-trace-dwh-db.md`, `context/sce/agent-trace-dwh-replica.md`, `context/sce/shared-turso-db.md`, and `context/sce/agent-trace-db.md` to describe the implemented messages/parts behavior. +- Add or update a focused conversation ETL domain context file and index it from `context/context-map.md`. + +## Constraints and non-goals + +- **In scope:** shared PR4 ETL helper extraction; source row models and short snapshot extraction for `messages` and `parts`; deterministic transforms; DWH fact/dimension/watermark transactions; message/part conflict validation; conversation runner/stats; focused filesystem/in-memory-DWH and source-contention tests; and durable documentation. +- **Out of scope:** `diff_traces` to `code_changes`; post-commit intersection ETL; commits, session or model materialization; control-plane calls; DWH provisioning; credential retrieval; OAuth; CLI or background sync orchestration; archive/search; message update CDC; and remote-to-source synchronization. +- **Constraints:** use the existing `RepositoryAgentTraceDb` metadata API; use the existing `AgentTraceDwhReplica` ownership boundary; never call `pull()` or `push()` from ETL; end source read transactions before transformation or destination work; retry only bounded transient source lock contention; do not add a parent-message foreign key; do not modify `001_dwh_schema.sql`; do not use timestamp-only cursors or ordering; and do not add a broad generic ETL trait framework. +- **Non-goal:** detect or reconcile historical updates to rows at or below a committed integer-ID watermark. Fields synchronized from `messages` and `parts` are append-only/immutable after insertion for ETL purposes, even though the source schema's updated-at triggers technically permit SQL updates. + +## Assumptions + +- The existing DWH baseline columns and indexes are sufficient: message rows store source lineage plus role/timestamp, and message parts store source lineage, source ID, type, text, text hash, and timestamp without a migration. +- `MessageRole` remains the authoritative supported representation for `user` and `assistant`; invalid source message roles fail explicitly rather than bypassing the destination constraint. +- `PartType` remains the authoritative supported representation for `text`, `reasoning`, `patch`, and `question`; unknown source values fail explicitly. +- The default batch size remains `500`, with one validated configuration seam shared by the table runners or an equivalent small local seam consistent with `AgentTraceEtl`. +- `ConversationEtl::run` may execute messages then parts sequentially, but each table commits facts and its own watermark independently; a failure in one table cannot roll back a previously committed batch in the other table. +- Test-only destination failure injection is acceptable for proving atomic rollback and does not become a production error path. + +## Task stack + +- [x] T01: `Extract shared ETL mechanics without changing Agent Trace behavior` (status:complete) + - Task ID: T01 + - Goal: Move genuinely reusable PR4 mechanics behind small internal helpers that can serve three table ETLs. + - Boundaries (in/out of scope): In — shared source-contention classification/retry and rollback hook, watermark read/upsert helpers keyed by repository/source/table, batch-size validation, common table batch-stat shape where it fits existing conventions, and updates to Agent Trace ETL tests/call sites. Out — messages/parts behavior, generic ETL traits, schema changes, API orchestration, and documentation beyond implementation comments. + - Dependencies: none + - Done when: `AgentTraceEtl` uses the shared helpers, its existing source snapshot and atomic load tests still pass, and the helper surface is small enough that row extraction/transformation/loading remain table-specific. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_etl`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Implementation evidence: Added the internal `cli/src/services/etl.rs` helper module for bounded source-contention retry/classification, positive batch-size validation, table watermark reads/upserts, and shared batch accounting. Registered it as `pub(crate) mod etl`; `AgentTraceEtl` now uses these helpers while retaining table-specific extraction, transformation, and loading behavior. + - Verification evidence: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_etl` passed all 15 focused tests; `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T02: `Implement independently watermarked messages ETL` (status:complete) + - Task ID: T02 + - Goal: Add incremental source extraction and transactional DWH loading for logical messages. + - Boundaries (in/out of scope): In — `SourceMessage`, exact source projection/query, supported role validation, source metadata lookup, logical identity lookup/insert/verification, `messages` watermark handling, stats, and message-focused initial/incremental/replay/conflict/rollback tests. Out — parts, conversation orchestration, code-change ETL, pull/push, and source update tracking. + - Dependencies: T01 + - Done when: message batches use `id > watermark ORDER BY id ASC LIMIT ?`, source reads are short and contention-safe, missing rows insert all required lineage/content fields, equal role/timestamp logical replays increment `already_present`, differing role or timestamp returns an integrity error containing repository/session/message identity, and every batch failure leaves rows and the `messages` watermark unchanged. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_etl`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_identity`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Implementation evidence: Added `cli/src/services/conversation_messages_etl.rs` with `SourceMessage`, bounded short-snapshot extraction using the shared contention retry, validated `MessageRole` transformation, independently keyed messages watermark reads/upserts, logical identity replay/conflict handling, atomic lineage/fact/watermark loading, configurable batching, and replica-bound runner/stats. Registered the module in `cli/src/services/mod.rs`. + - Verification evidence: `conversation_messages_etl` passed 8 focused tests; `conversation_messages_identity` passed 3 focused identity/rollback tests; `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T03: `Implement source-lineage-scoped parts ETL` (status:complete) + - Task ID: T03 + - Goal: Add incremental, verbatim-preserving message-part extraction and transactional loading. + - Boundaries (in/out of scope): In — `SourceMessagePart`, exact source projection/query, `PartType` conversion, UTF-8 SHA-256 transform, source-part identity verification/insertion, no-parent requirement, `parts` watermark handling, ordering and part-focused tests, and source-writer contention coverage. Out — messages ETL changes except shared test fixtures, conversation runner, code-change ETL, pull/push, and CDC. + - Dependencies: T01 + - Done when: parts use `id > watermark ORDER BY id ASC LIMIT ?`, valid types are preserved exactly, unknown types fail, text bytes round-trip exactly, hash values are lowercase SHA-256, identical source-lineage replays are counted without duplication, conflicts fail without overwrite, same local IDs from different source instances coexist, and a part batch succeeds before its parent message exists. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_etl`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_identity`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Implementation evidence: Added `cli/src/services/conversation_parts_etl.rs` with bounded short-snapshot extraction, supported `PartType` validation, exact UTF-8 SHA-256 hashing, source-lineage identity replay/conflict handling, atomic `message_parts` loading, independent `parts` watermarks, configurable batching, and source-writer contention coverage. Registered the module in `cli/src/services/mod.rs`. + - Verification evidence: `conversation_parts_etl` passed all 11 focused tests; `conversation_parts_identity` passed all 3 identity/rollback tests; `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T04: `Expose ConversationEtl and prove independent table progress` (status:complete) + - Task ID: T04 + - Goal: Provide the conversation-level API and end-to-end proof that message and part ETLs share mechanics but not progress state. + - Boundaries (in/out of scope): In — `ConversationEtl`, `ConversationEtlStats`, table-runner composition, replica-owned execution, independent message/part watermark tests, initial/incremental/no-op orchestration tests, out-of-order reconstruction/order checks, source contention integration tests for both source tables, and batch rollback/replay coverage through the public table runners. Out — pull/push calls, credentials, CLI command wiring, background scheduling, code changes, and remote orchestration. + - Dependencies: T02, T03 + - Done when: callers can run `conversation_etl.run(repository_id, source, replica)`, stats expose both table results, messages and parts can advance independently, parts-before-messages remains valid, equal timestamps reconstruct by `generated_at_unix_ms, source_part_id`, and the API contains no transport or control-plane behavior. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl`; inspect the runner for absence of `pull()`/`push()` and credential access; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Implementation evidence: Added `cli/src/services/conversation_etl.rs` with independently configurable/default-batched `ConversationEtl`, table-level `ConversationEtlStats`, replica-bound execution, and composition over the existing messages and parts runners. Registered the service module and exposed crate-internal destination composition seams so the runner remains the sole conversation-level transaction coordinator while table watermarks commit independently. Added end-to-end coverage for initial/incremental/no-op runs, independent progress, parts-before-messages ingestion, deterministic equal-timestamp ordering, and non-blocking source readers for both source tables. The runner contains no transport, credential, CLI, or control-plane behavior. + - Verification evidence: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl` passed all 7 focused tests; `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T05: `Record the conversation ETL and append-only architecture contract` (status:complete) + - Task ID: T05 + - Goal: Make the implemented conversation pipeline and its source immutability assumption durable in repository context. + - Boundaries (in/out of scope): In — focused conversation ETL context, context-map index entry, updates to root architecture/overview/glossary and related Agent Trace ETL/DWH/replica/source documents, and an inspection of production message/part writers for append-only violations. Out — code changes, migration edits, source update CDC, CLI/control-plane docs, and unrelated context cleanup. + - Dependencies: T04 + - Done when: durable context describes the two source-to-DWH flows, independent watermarks, exact identity/conflict rules, verbatim text/hash, deterministic ordering, no parent FK, pull/push separation, and append-only assumption; the source audit records that current production writes are inserts and only schema-maintenance `updated_at` triggers issue updates, or identifies a concrete architectural conflict if that is no longer true. + - Verification notes (commands or checks): inspect the documented paths against `cli/src/services/agent_trace_etl/`, the source repository adapter, DWH schema, replica API, and `grep -RInE 'UPDATE[[:space:]]+(messages|parts)' cli/src cli/migrations` output; run `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl`. + - Implementation evidence: Updated root overview, architecture, and glossary contracts plus the shared ETL, DWH replica, and repository-source context to document ConversationEtl, independent message/part watermarks, identity/conflict and preservation rules, pull/push separation, and the append-only ETL assumption. The existing focused conversation context files and context-map entries were verified as canonical descriptions. The source audit found only `trg_messages_updated_at` and `trg_parts_updated_at` trigger bodies issuing `UPDATE` statements; active capture uses insert helpers. + - Verification evidence: `grep -RInE 'UPDATE[[:space:]]+(messages|parts)' cli/src cli/migrations` returned only the two schema-maintenance trigger bodies; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl` passed all 7 focused tests; `git diff --check` passed; changed context files remain at or below 250 lines. + +## Open questions + +None. The request fixes the identity, transaction, ordering, source-safety, API, and non-goal boundaries, and the existing DWH baseline already supplies the required destination schema without a migration. + +## Validation Report + +**Status:** validated +**Date:** 2026-08-08 + +### Commands run + +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_etl` -> exit 0 (15 tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_etl` -> exit 0 (8 tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_messages_identity` -> exit 0 (3 tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_etl` -> exit 0 (11 tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_parts_identity` -> exit 0 (3 tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml conversation_etl` -> exit 0 (7 tests passed) +- `grep -RInE 'UPDATE[[:space:]]+(messages|parts)' cli/src cli/migrations` -> exit 0 (only the two schema-maintenance trigger bodies matched) +- `nix run .#pkl-check-generated` -> exit 0 (101 generated files passed) +- `nix flake check` -> exit 0 (all checks passed) + +### Scaffolding removed + +- None. + +### Success-criteria verification + +- [x] AC1: Shared ETL mechanics and unchanged Agent Trace behavior -> focused Agent Trace ETL tests passed. +- [x] AC2: Incremental messages extraction and atomic loading -> focused messages ETL tests passed. +- [x] AC3: Logical message identity and conflict rollback -> focused messages identity tests passed. +- [x] AC4: Supported parts, verbatim text, and SHA-256 hashing -> focused parts ETL tests passed. +- [x] AC5: Source-lineage part identity, conflict handling, and parentless loading -> focused parts identity tests passed. +- [x] AC6: Conversation API and independent progress -> conversation ETL tests passed. +- [x] AC7: End-to-end conversation behavior and rollback/contention coverage -> conversation ETL tests passed. +- [x] AC8: Durable context and append-only source audit -> required context files were inspected; source audit matched only `trg_messages_updated_at` and `trg_parts_updated_at` trigger bodies. + +### Failed checks and follow-ups + +- None. + +### Residual risks + +- None identified. diff --git a/context/sce/agent-trace-db.md b/context/sce/agent-trace-db.md index 092f61de..ea51ed37 100644 --- a/context/sce/agent-trace-db.md +++ b/context/sce/agent-trace-db.md @@ -2,7 +2,7 @@ `cli/src/services/agent_trace_db/mod.rs` defines the shared Agent Trace insert payloads and helpers consumed by the repository-scoped adapter. `RepositoryAgentTraceDb` (see [Repository-scoped adapter seam](#repository-scoped-adapter-seam)) is the sole Agent Trace DB adapter; the former checkout-scoped `AgentTraceDb` / `AgentTraceDbSpec` type, its `open_at` / `open_for_hooks_without_migrations` constructors, and its 15-file migration chain were removed by the `retire-legacy-agent-trace-db` plan (see [context/decisions/2026-07-17-retire-legacy-agent-trace-db.md](../decisions/2026-07-17-retire-legacy-agent-trace-db.md)). -This is the live-capture **source** schema written by hooks and `sce trace` — one row family per repository, mutated on every capture event. It is a distinct database boundary from the append-oriented Agent Trace **DWH destination** schema, which is never written by hooks or any live capture path and exists only for a future ETL consumer to ingest into. See [agent-trace-dwh-db.md](agent-trace-dwh-db.md). +This is the live-capture **source** schema written by hooks and `sce trace` — one row family per repository, mutated on every capture event. It is a distinct database boundary from the append-oriented Agent Trace **DWH destination** schema, which is never written by hooks or any live capture path and is populated by the CLI-independent ETL consumers. See [agent-trace-dwh-db.md](agent-trace-dwh-db.md). ## Shared insert/query payloads @@ -42,7 +42,7 @@ This adapter has no canonical `DbSpec::db_path()`; callers must resolve ` Result` inserts the singleton metadata row on first initialization, errors if an existing DB stores a different repository ID, and returns the typed metadata: an existing valid `source_instance_id` is returned unchanged, while a missing one (the migration's empty placeholder) is atomically claimed through `UPDATE repository_metadata SET source_instance_id = ?1 WHERE id = 1 AND source_instance_id = ''` and re-read — the conditional `WHERE` means only one concurrent caller's `UPDATE` changes a row, so every racing caller reads back the same stored winner and no caller ever replaces an already-valid identity. -`source_instance_id` identifies a database-lineage (one independently created database file), not a checkout: every clone/worktree that resolves the same physical repository-scoped DB file shares one `source_instance_id`, while two independently created DB files for the same `repository_id` (for example after a manual copy or an out-of-band re-initialization) receive different ones. This is distinct from `checkout_id`, which is per clone/worktree, diagnostic-only, and never persisted on Agent Trace rows or `repository_metadata`. A future ETL/DWH consumer is expected to key row provenance on the tuple `(repository_id, source_instance_id, source_table, source_row_id)` — the logical repository, the database lineage that produced the row, the source table name, and the row's local primary key — so rows from independently created database files for the same repository never collide during ingestion; no such consumer exists yet in this repository. +`source_instance_id` identifies a database-lineage (one independently created database file), not a checkout: every clone/worktree that resolves the same physical repository-scoped DB file shares one `source_instance_id`, while two independently created DB files for the same `repository_id` (for example after a manual copy or an out-of-band re-initialization) receive different ones. This is distinct from `checkout_id`, which is per clone/worktree, diagnostic-only, and never persisted on Agent Trace rows or `repository_metadata`. The CLI-independent ETL consumers key row provenance on the tuple `(repository_id, source_instance_id, source_table, source_row_id)` — the logical repository, the database lineage that produced the row, the source table name, and the row's local primary key — so rows from independently created database files for the same repository never collide during ingestion. The messages pipeline is implemented in `cli/src/services/conversation_messages_etl.rs`, and the source-lineage-scoped message-part pipeline is implemented in `cli/src/services/conversation_parts_etl.rs`; code-change consumers remain future work. `RepositoryAgentTraceDb` exposes repository-level write helpers for the current row families by delegating to the same typed insert payloads and parameterized SQL used by the checkout-scoped adapter: `insert_diff_trace`, `insert_post_commit_patch_intersection`, `insert_agent_trace`, `insert_message`, `insert_messages`, `insert_part`, and `insert_parts`. It also exposes `recent_diff_trace_patches(cutoff_time_ms, end_time_ms)` by delegating to the shared recent diff-trace query/parser helper, so repository-scoped attribution reads use the same chronological inclusive window semantics without a checkout filter. These methods preserve the existing row shapes and do not add checkout provenance columns or checkout-scoped write/query APIs. @@ -206,6 +206,10 @@ Post-commit intersection rows are written by the active `post-commit` hook flow - No `context/tmp` artifact is written for conversation traces. - The generated OpenCode agent-trace plugin sends mixed-batch envelopes for conversation traces: regular `message` and `message.part` events each carry one per-item `type`, while diff-backed `message` events send one envelope containing the synthetic parent message item plus patch part items. +## Source writer audit and ETL immutability + +The messages/parts ETL uses integer-ID watermarks and deliberately does not provide update CDC. An audit of `cli/src` and `cli/migrations` found that the only matches for `UPDATE messages` and `UPDATE parts` are the baseline schema's `trg_messages_updated_at` and `trg_parts_updated_at` trigger bodies. Active conversation capture writes through `RepositoryAgentTraceDb::insert_messages()` and `insert_parts()`; those helpers append rows (message duplicates are ignored by natural identity, while parts remain append-only). Therefore ETL treats synchronized role/timestamp and part session/message/type/text/hash/timestamp fields as immutable after insertion. If a future production writer intentionally changes those fields, update CDC or an explicit architectural decision is required rather than silently advancing the existing watermark. + `sce hooks session-model` is no longer a supported command route, generated Claude settings no longer produce `SessionStart` model-attribution events, and the Agent Trace DB adapter no longer exposes a `session_models` API or fresh-schema table. See [agent-trace-hooks-command-routing.md](agent-trace-hooks-command-routing.md). ## Recent patch reads diff --git a/context/sce/agent-trace-dwh-db.md b/context/sce/agent-trace-dwh-db.md index 05662be6..2d11bd02 100644 --- a/context/sce/agent-trace-dwh-db.md +++ b/context/sce/agent-trace-dwh-db.md @@ -1,6 +1,6 @@ # Agent Trace DWH Database (Destination Schema) -The Agent Trace DWH is a separate, append-oriented destination schema for the Agent Trace ETL consumer of repository-scoped Agent Trace data. It is a distinct database boundary from the repository-scoped `agent-trace.db` source schema (see [agent-trace-db.md](agent-trace-db.md)): the DWH is never written by hooks, `sce trace`, or any live capture path. The current ETL slice transforms and atomically loads only `agent_traces` through the multi-batch `AgentTraceEtl` run loop; transport synchronization remains a separate caller concern. +The Agent Trace DWH is a separate, append-oriented destination schema for the Agent Trace ETL consumer of repository-scoped Agent Trace data. It is a distinct database boundary from the repository-scoped `agent-trace.db` source schema (see [agent-trace-db.md](agent-trace-db.md)): the DWH is never written by hooks, `sce trace`, or any live capture path. The current ETL slices transform and atomically load `agent_traces`, logical `messages`, and source-lineage-scoped `message_parts` through independent table runners; transport synchronization remains a separate caller concern. ## Adapter @@ -32,6 +32,6 @@ Two different uniqueness scopes are used, chosen by whether the source identity ## Data preservation and hashing -`message_parts.text` and `agent_traces.trace_json` store complete source text/JSON verbatim, with no truncation or normalization columns. Source event timestamps (`generated_at_unix_ms`, `commit_time_ms`, `time_ms`) are preserved as integer milliseconds, matching the source schema; only DWH-local metadata timestamps (`first_seen_at`, `updated_at`, `ingested_at`) use the shared UTC text default `strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`. Integrity hash columns (`message_parts.text_sha256`, `agent_traces.trace_json_sha256`, `code_changes.patch_sha256`) are schema storage; the `agent_traces` ETL slice computes and populates the lowercase SHA-256 of exact source JSON bytes, while other table hashes remain future work. +`message_parts.text` and `agent_traces.trace_json` store complete source text/JSON verbatim, with no truncation or normalization columns. Source event timestamps (`generated_at_unix_ms`, `commit_time_ms`, `time_ms`) are preserved as integer milliseconds, matching the source schema; only DWH-local metadata timestamps (`first_seen_at`, `updated_at`, `ingested_at`) use the shared UTC text default `strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`. Integrity hash columns (`message_parts.text_sha256`, `agent_traces.trace_json_sha256`, `code_changes.patch_sha256`) are schema storage; the `agent_traces` ETL slice computes and populates the lowercase SHA-256 of exact source JSON bytes, and the `PartsEtl` slice computes and populates the lowercase SHA-256 of exact source part text bytes. Message rows require no content hash and code-change hashing remains future work. -See also: [agent-trace-db.md](agent-trace-db.md), [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), [shared-turso-db.md](shared-turso-db.md), [../context-map.md](../context-map.md) +See also: [agent-trace-db.md](agent-trace-db.md), [agent-trace-etl.md](agent-trace-etl.md), [conversation-parts-etl.md](conversation-parts-etl.md), [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), [shared-turso-db.md](shared-turso-db.md), [../context-map.md](../context-map.md) diff --git a/context/sce/agent-trace-dwh-replica.md b/context/sce/agent-trace-dwh-replica.md index 6f23e5ff..7eac06f5 100644 --- a/context/sce/agent-trace-dwh-replica.md +++ b/context/sce/agent-trace-dwh-replica.md @@ -1,6 +1,6 @@ # Agent Trace DWH Turso Sync Replica -`AgentTraceDwhReplica` is the sole owner of a Turso Sync connection to a repository's `agent-trace-sync.db` — a disposable, single-owner local database distinct from both the multiprocess-WAL source `agent-trace.db` (see [agent-trace-db.md](agent-trace-db.md)) and the `Agent Trace DWH`'s own explicit-path adapter (see [agent-trace-dwh-db.md](agent-trace-dwh-db.md)). It is the boundary used by the CLI-independent `AgentTraceEtl` bridge for local fact/watermark loading; ETL never performs pull/push, credential discovery/persistence, or background sync. +`AgentTraceDwhReplica` is the sole owner of a Turso Sync connection to a repository's `agent-trace-sync.db` — a disposable, single-owner local database distinct from both the multiprocess-WAL source `agent-trace.db` (see [agent-trace-db.md](agent-trace-db.md)) and the `Agent Trace DWH`'s own explicit-path adapter (see [agent-trace-dwh-db.md](agent-trace-dwh-db.md)). It is the boundary used by the CLI-independent `AgentTraceEtl` and `ConversationEtl` bridges for local fact/watermark loading; ETL never performs pull/push, credential discovery/persistence, or background sync. ## Ownership and lock-before-open diff --git a/context/sce/agent-trace-etl.md b/context/sce/agent-trace-etl.md index 812203be..a544a537 100644 --- a/context/sce/agent-trace-etl.md +++ b/context/sce/agent-trace-etl.md @@ -1,6 +1,6 @@ # Agent Trace ETL -`cli/src/services/agent_trace_etl/mod.rs` owns the first production ETL slice between the repository-scoped multiprocess-WAL `agent-trace.db` source and the lock-owned `agent-trace-sync.db` DWH replica. `AgentTraceEtl` accepts an open `RepositoryAgentTraceDb`, a repository ID, and an `AgentTraceDwhReplica`; it verifies source metadata, obtains the stored `source_instance_id`, and runs only the `agent_traces` table. It does not acquire credentials, invoke `pull()`/`push()`, or depend on CLI orchestration. +`cli/src/services/agent_trace_etl/mod.rs` owns the shared mechanics and the table ETL slices between the repository-scoped multiprocess-WAL `agent-trace.db` source and the lock-owned `agent-trace-sync.db` DWH replica. `AgentTraceEtl` accepts an open `RepositoryAgentTraceDb`, a repository ID, and an `AgentTraceDwhReplica`; it verifies source metadata, obtains the stored `source_instance_id`, and runs the `agent_traces` table. The sibling `conversation_messages_etl` and `conversation_parts_etl` modules apply the same mechanics to logical `messages` and source-lineage-scoped `parts` rows, each with its own table watermark. None of these pipelines acquires credentials, invokes `pull()`/`push()`, or depends on CLI orchestration. ## Incremental run contract @@ -16,6 +16,19 @@ Each destination batch uses one `BEGIN IMMEDIATE` transaction for repository/sou ## Replica and reconstruction boundary -`AgentTraceDwhReplica` remains the sole owner of the sync connection and bridge lock. Its `run_agent_trace_etl()` method delegates to `AgentTraceEtl` while preserving that ownership; ETL never pulls or pushes remote state. The local sync database is a durable transaction/replay boundary and is reconstructible from the remote DWH: crash or local-file loss is handled by reopening/replaying source rows and, when required, remote reconstruction. Future message, message-part, and code-change ETLs must preserve the same short-source-transaction, exact-cursor, and local fact-plus-watermark invariants. +`AgentTraceDwhReplica` remains the sole owner of the sync connection and bridge lock. Its `run_agent_trace_etl()` method delegates to `AgentTraceEtl` while preserving that ownership; `ConversationEtl` composes the message and part runners through the same lock-owned destination without acquiring credentials or invoking transport. ETL never pulls or pushes remote state. The local sync database is a durable transaction/replay boundary and is reconstructible from the remote DWH: crash or local-file loss is handled by reopening/replaying source rows and, when required, remote reconstruction. The messages and parts runners preserve the same short-source-transaction, exact-cursor, and local fact-plus-watermark invariants; their synchronized source fields are append-only/immutable for ETL purposes. Source audit found no production `UPDATE` of those fields beyond the schema-maintenance `updated_at` triggers, so update CDC remains deliberately out of scope. Code-change ETL remains future work. -See also: [agent-trace-db.md](agent-trace-db.md), [agent-trace-dwh-db.md](agent-trace-dwh-db.md), [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), [shared-turso-db.md](shared-turso-db.md), [../overview.md](../overview.md), [../architecture.md](../architecture.md), [../glossary.md](../glossary.md) +## Source contention retry + +A private `run_with_source_contention_retry` wraps each extraction attempt with a bounded backoff policy (5 attempts, 1s per-attempt timeout, 25ms..200ms backoff — deliberately smaller than the connection-open retry budget in `crate::services::db`, since contention on a non-blocking read is expected to be rare and self-clearing). + +`is_transient_source_contention` classifies only two textual forms as retryable: + +- Turso's typed `Busy` error, whose SDK-mapped message is exactly `"database is locked"`. +- The narrow `"table is locked"` textual form used when the underlying `LimboError::TableLocked` case falls through to a generic error variant (its Display is `"Runtime error: database table is locked"`). + +Every other error — including genuine extraction/mapping failures such as a missing table or column — fails on the first attempt without retry. Because `TursoTransaction::execute`/`query_map` (used inside `read_transaction`) wrap the underlying `turso::Error` into a formatted `anyhow::Error` message rather than preserving it as a typed source, classification matches on the resulting message text; this is why the SDK's textual "database is locked" content, not a `downcast_ref::()`, is the retry signal. + +Before every retried attempt (not the first), the retry loop calls `TursoDb::rollback_best_effort()` (`pub(crate)` in `cli/src/services/db/mod.rs`) to clear a stale failed transaction before issuing the next `BEGIN`. + +See also: [agent-trace-db.md](agent-trace-db.md), [agent-trace-dwh-db.md](agent-trace-dwh-db.md), [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), [shared-turso-db.md](shared-turso-db.md), [../overview.md](../overview.md), [../architecture.md](../architecture.md), [../glossary.md](../glossary.md), [conversation ETL append-only watermark decision](../decisions/2026-08-08-conversation-etl-append-only-watermarks.md) diff --git a/context/sce/conversation-etl.md b/context/sce/conversation-etl.md new file mode 100644 index 00000000..181013ae --- /dev/null +++ b/context/sce/conversation-etl.md @@ -0,0 +1,29 @@ +# Conversation ETL + +`cli/src/services/conversation_etl.rs` exposes `ConversationEtl`, the +CLI-independent composition boundary for the repository-source conversation +pipelines. Callers provide a repository ID, an already-open +`RepositoryAgentTraceDb`, and the lock-owning `AgentTraceDwhReplica`: + +```rust +ConversationEtl::default().run(repository_id, source, replica)?; +``` + +The runner verifies source metadata once, then executes the existing +`MessagesEtl` and `PartsEtl` table runners. It reports +`ConversationEtlStats` with the complete table-level message and part stats. +Batch sizes default to 500 and can be configured independently. + +Messages and parts do not share a transaction or watermark. Each table commits +its own facts, lineage dimensions, replay/conflict checks, and +`(repository_id, source_instance_id, source_table)` watermark. Consequently, +parts may be ingested before their parent message, and a no-op table does not +prevent the other table from advancing. Parts with equal timestamps are +reconstructed by `generated_at_unix_ms, source_part_id`. + +This composition layer owns no credentials, remote transport, pull/push calls, +CLI command wiring, scheduling, or control-plane behavior. See +[conversation-messages-etl.md](conversation-messages-etl.md), +[conversation-parts-etl.md](conversation-parts-etl.md), +[agent-trace-dwh-replica.md](agent-trace-dwh-replica.md), and +[agent-trace-etl.md](agent-trace-etl.md). diff --git a/context/sce/conversation-messages-etl.md b/context/sce/conversation-messages-etl.md new file mode 100644 index 00000000..647553d7 --- /dev/null +++ b/context/sce/conversation-messages-etl.md @@ -0,0 +1,42 @@ +# Conversation Messages ETL + +`cli/src/services/conversation_messages_etl.rs` owns the incremental messages +pipeline from a repository-scoped Agent Trace database into the Agent Trace +DWH. It is a CLI-independent table runner: it does not pull or push the DWH +replica, retrieve credentials, or call CLI orchestration. + +## Source and progress + +`MessagesEtl` verifies the source `RepositoryMetadata` and uses its +`source_instance_id`. Each bounded batch reads exactly +`id > watermark ORDER BY id ASC LIMIT batch_size` in a short plain `BEGIN` +transaction. Only transient database/table lock contention receives the shared +bounded retry; transformation and destination work begin after the source +snapshot ends. The default batch size is 500 and callers can provide a +validated positive size. + +Progress is independently stored under +`(repository_id, source_instance_id, "messages")`. An absent watermark is zero. +The runner repeats batches until extraction is empty and reports extracted, +inserted, already-present, batch, and before/after watermark counts. + +## Destination identity and atomicity + +A DWH message is logically identified by +`(repository_id, session_id, message_id)`, excluding `source_instance_id`. +Missing rows preserve the source session, message, validated `user` or +`assistant` role, and `generated_at_unix_ms`, together with repository and source +lineage. A matching role/timestamp replay increments `already_present`; a role +or timestamp mismatch returns an integrity conflict naming the repository, +session, and message identity and never overwrites the row. + +Each batch commits lineage dimensions, message facts, and the messages +watermark in one `BEGIN IMMEDIATE` destination transaction. A conflict, +transformation failure, or other batch error rolls all of them back so the +source batch can be replayed completely. Parts, conversation orchestration, +source update CDC, and remote synchronization remain separate ETL concerns. + +See also: [agent-trace-etl.md](agent-trace-etl.md), +[agent-trace-dwh-db.md](agent-trace-dwh-db.md), +[agent-trace-db.md](agent-trace-db.md), [shared-turso-db.md](shared-turso-db.md), +and [../context-map.md](../context-map.md). diff --git a/context/sce/conversation-parts-etl.md b/context/sce/conversation-parts-etl.md new file mode 100644 index 00000000..9f8ba3c7 --- /dev/null +++ b/context/sce/conversation-parts-etl.md @@ -0,0 +1,56 @@ +# Conversation Parts ETL + +`cli/src/services/conversation_parts_etl.rs` owns the independently watermarked +parts pipeline from a repository-scoped Agent Trace source database into the +DWH `message_parts` fact table. It is CLI-independent: it does not pull or +push the replica, retrieve credentials, call orchestration, or require a +parent message row. + +## Source and progress + +`PartsEtl` verifies the source repository metadata and uses its +`source_instance_id`. Each bounded batch reads exactly +`id > watermark ORDER BY id ASC LIMIT batch_size` in a short plain `BEGIN` +transaction. Only transient database/table lock contention receives the shared +bounded retry; transformation and destination work start after the source +snapshot ends. The default batch size is 500 and callers can provide a +validated positive size. + +Progress is independently stored under +`(repository_id, source_instance_id, "parts")`. An absent watermark is zero. +The runner reports extracted, inserted, already-present, batch, and +before/after watermark counts. A failed destination batch leaves facts, +lineage dimensions, and the watermark unchanged for complete replay. + +## Transformation and identity + +The source `type` is converted through the existing `PartType` representation; +only `text`, `reasoning`, `patch`, and `question` are accepted. The source text +is copied verbatim and its exact UTF-8 bytes receive a lowercase hexadecimal +SHA-256 hash. + +A destination part is identified by +`(repository_id, source_instance_id, source_part_id)`. A matching replay +verifies session, message, type, exact text, hash, and timestamp, then counts +`already_present`; any mismatch is an integrity conflict and never overwrites +the existing row. Because the local source ID is scoped to the source +instance, independently created source databases can contribute the same +integer ID without collision. + +The DWH schema has no foreign key from `message_parts` to `messages`, so a part +batch may commit before its parent message exists. When parts share a message +timestamp, consumers reconstruct deterministic order using +`generated_at_unix_ms, source_part_id`. + +## Source immutability and boundaries + +The integer-ID watermark treats synchronized source fields as append-only for +ETL purposes. Update CDC is deliberately not part of this pipeline. Remote +synchronization, credentials, pull/push orchestration, conversation-level +composition, and code-change ingestion remain separate concerns. + +See also: [conversation-messages-etl.md](conversation-messages-etl.md), +[agent-trace-etl.md](agent-trace-etl.md), +[agent-trace-dwh-db.md](agent-trace-dwh-db.md), +[agent-trace-db.md](agent-trace-db.md), +[shared-turso-db.md](shared-turso-db.md), and [../context-map.md](../context-map.md). diff --git a/context/sce/shared-turso-db.md b/context/sce/shared-turso-db.md index 2bd60c9a..6fad2518 100644 --- a/context/sce/shared-turso-db.md +++ b/context/sce/shared-turso-db.md @@ -25,7 +25,7 @@ - `pub(crate) fn from_connection(conn: turso::Connection, runtime: tokio::runtime::Runtime) -> Self`: wraps an already-open connection and the runtime that opened it as a `TursoDb`, without opening a new connection, creating `__sce_migrations`, or running migrations. For callers that open a connection through a non-local-path builder — currently only `AgentTraceDwhReplica`'s Turso Sync open (see [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md)) — and want to reuse this adapter's synchronous SQL surface and schema-readiness checks instead of duplicating them. - `pub(crate) fn block_on(&self, future: F) -> F::Output`: runs a future to completion on the runtime backing this connection, so a companion async handle opened alongside it (e.g. a Turso Sync `Database` used for `pull`/`push`) can be driven without owning a second runtime. - `pub fn transaction(&self, f: impl FnOnce(&TursoTransaction) -> Result) -> Result`: runs `f` inside an explicit `BEGIN IMMEDIATE` transaction on the same connection. Commits only when `f` returns `Ok`; any closure error, or a failed `COMMIT`, triggers a best-effort `ROLLBACK` before the original error is returned. `TursoTransaction<'a, M: DbSpec>` is the transaction-scoped handle passed to `f`, exposing non-retried synchronous `execute()`/`query_map()` over the same connection/runtime; no raw `turso::Transaction` is ever exposed outside this module. This is a single-level seam: no nested transactions, savepoints, or transaction-level retry. It works for any `TursoDb`, including `AgentTraceDwhDb` opened through `new_at()` or through `AgentTraceDwhReplica`'s `from_connection` seam. - - `pub fn read_transaction(&self, f: impl FnOnce(&TursoTransaction) -> Result) -> Result`: same commit/rollback shape as `transaction()`, but issues a plain `BEGIN` instead of `BEGIN IMMEDIATE`, so the transaction never reserves the database's write lock and concurrent writers on other connections are not blocked while it is open. Both methods share a private `run_transaction(begin_sql, f)` helper. Intended for read-only callers such as Agent Trace ETL source extraction (see [agent-trace-etl.md](agent-trace-etl.md)), which must not interfere with concurrent multiprocess-WAL hook writers on `agent-trace.db`. + - `pub fn read_transaction(&self, f: impl FnOnce(&TursoTransaction) -> Result) -> Result`: same commit/rollback shape as `transaction()`, but issues a plain `BEGIN` instead of `BEGIN IMMEDIATE`, so the transaction never reserves the database's write lock and concurrent writers on other connections are not blocked while it is open. Both methods share a private `run_transaction(begin_sql, f)` helper. Intended for read-only callers such as the Agent Trace, conversation messages, and conversation parts ETL source extraction (see [agent-trace-etl.md](agent-trace-etl.md), [conversation-messages-etl.md](conversation-messages-etl.md), and [conversation-parts-etl.md](conversation-parts-etl.md)), which must not interfere with concurrent multiprocess-WAL hook writers on `agent-trace.db`. - `pub(crate) fn rollback_best_effort(&self)`: issues a best-effort `ROLLBACK`, ignoring failures. Used internally by `transaction()`/`read_transaction()` after a closure error or failed commit, and available to same-crate callers that retry a whole transaction attempt (e.g. Agent Trace ETL source contention retries) to clear a stale failed transaction before issuing a new `BEGIN`. - `EncryptedTursoDb`: encrypted-adapter seam parallel to `TursoDb` with the same structural shape (connection, runtime bridge, and spec marker). `EncryptedTursoDb::new()` resolves the encryption key via `encryption_key::get_or_create_encryption_key()` (environment variable `SCE_AUTH_DB_ENCRYPTION_KEY` with OS credential-store fallback), enables Turso experimental local encryption, applies strict `aegis256` cipher selection through `turso::EncryptionOpts` during local DB open/connect, wraps that open/connect block in the same connection-open retry policy resolved from `policies.database_retry..connection_open`, and runs embedded migrations after connect. - `EncryptedTursoDb` exposes the same public synchronous `execute()`, `query()`, `query_map()`, and `run_migrations()` methods; operation methods use the same config-driven query retry policy as `TursoDb`. @@ -66,7 +66,7 @@ The shared module is exported from `cli/src/services/mod.rs` and compile-checked - `cli/src/services/auth_db/mod.rs`: `AuthDb = EncryptedTursoDb`, with `AuthDbSpec` resolving `auth_db_path()` and loading ordered auth migrations where baseline SQL creates `auth_credentials` without `user_id`, with `updated_at`, and a trigger that auto-refreshes `updated_at` on row updates. - `cli/src/services/agent_trace_dwh_db/mod.rs`: `AgentTraceDwhDb = TursoDb`, a separate append-oriented destination-schema adapter for the CLI-independent Agent Trace ETL consumer, distinct from the repository-scoped source schema above. Explicit-path only (no canonical `db_path()`), reuses the `"agent_trace_db"` retry config key, and is not wired into any lifecycle provider, doctor/setup flow, or CLI command. See [agent-trace-dwh-db.md](agent-trace-dwh-db.md). - `cli/src/services/agent_trace_dwh_replica/mod.rs`: `AgentTraceDwhReplica`, the sole owner of a Turso Sync connection to the repository-scoped `agent-trace-sync.db` replica. Built by opening a connection through `turso::sync::Builder` (the `sync` Cargo feature) and wrapping it as an `AgentTraceDwhDb` via the `from_connection`/`block_on` seam above, rather than through `TursoDb::new`/`new_at`. See [agent-trace-dwh-replica.md](agent-trace-dwh-replica.md). -- `cli/src/services/agent_trace_etl/mod.rs`: the first `TursoDb::read_transaction` consumer and the incremental `AgentTraceEtl` bridge. It extracts bounded, ordered `agent_traces` batches without reserving the source write lock, then loads facts and the lineage watermark atomically through `AgentTraceDwhReplica`. See [agent-trace-etl.md](agent-trace-etl.md). +- `cli/src/services/agent_trace_etl/mod.rs`: the incremental `AgentTraceEtl` bridge. It extracts bounded, ordered `agent_traces` batches without reserving the source write lock, then loads facts and the lineage watermark atomically through `AgentTraceDwhReplica`. `cli/src/services/conversation_messages_etl.rs` uses the same `read_transaction()` and shared retry seam for logical `messages` batches with an independent watermark. See [agent-trace-etl.md](agent-trace-etl.md) and [conversation-messages-etl.md](conversation-messages-etl.md). All three database areas (local DB, auth DB, Agent Trace DB) have lifecycle providers. `lifecycle_providers(include_hooks)` registers database providers in order `LocalDbLifecycle` → `AuthDbLifecycle` → `AgentTraceDbLifecycle` before optional hooks. Setup initializes local/auth DBs, establishes Agent Trace checkout identity for diagnostics, initializes the repository-scoped Agent Trace DB with migrations/metadata, and reports credential-safe repository identity metadata. Hook runtime (`open_repository_db_for_hook_runtime` in `cli/src/services/agent_trace_storage/mod.rs`) never creates, migrates, or repairs schema/migration metadata: it opens through `open_without_migrations_at` and calls `ensure_schema_ready_for_hooks()`, a non-mutating readiness check that fails with `sce setup` guidance when the schema is missing or migration-incomplete; only source-instance metadata initialization (an atomic, race-safe `UPDATE` of an existing column) still runs on the hook path. Doctor diagnoses/fixes DB parent/path readiness through lifecycle providers.