diff --git a/cli/src/services/agent_trace_db/mod.rs b/cli/src/services/agent_trace_db/mod.rs index c972c0ac..97996afa 100644 --- a/cli/src/services/agent_trace_db/mod.rs +++ b/cli/src/services/agent_trace_db/mod.rs @@ -1,6 +1,6 @@ //! Agent trace Turso database adapter. -use anyhow::{Context, Result}; +use anyhow::{bail, Context, Result}; use turso::Value as TursoValue; use crate::services::{ @@ -21,6 +21,37 @@ pub mod repository; pub const PAYLOAD_TYPE_PATCH: &str = "patch"; pub const PAYLOAD_TYPE_STRUCTURED: &str = "structured"; +/// Strictly normalize one stored diff-trace payload into the canonical patch model. +/// +/// Unified patches and Claude structured payloads share this boundary so ETL callers +/// receive an explicit error for malformed or unsupported source values. The recent +/// diff-trace reader adapts those errors back into its existing best-effort skip +/// accounting. +pub fn normalize_diff_trace_payload( + payload_type: &str, + payload: &str, + session_id: &str, + time_ms: i64, + tool_version: Option<&str>, +) -> Result { + match payload_type { + PAYLOAD_TYPE_PATCH => parse_patch(payload, Some(session_id)).map_err(Into::into), + PAYLOAD_TYPE_STRUCTURED => { + let time = u64::try_from(time_ms) + .context("structured diff-trace time_ms must be non-negative")?; + let payload = serde_json::from_str::(payload) + .context("invalid structured payload JSON")?; + match derive_claude_structured_patch("PostToolUse", &payload, time, tool_version) { + ClaudeStructuredPatchDerivationResult::Derived(derived) => Ok(derived.patch), + ClaudeStructuredPatchDerivationResult::Skipped(reason) => { + bail!("{reason}") + } + } + } + other => bail!("unsupported diff-trace payload_type: {other}"), + } +} + /// Parameterized SQL for inserting a captured diff trace payload. pub const INSERT_DIFF_TRACE_SQL: &str = "INSERT INTO diff_traces (time_ms, session_id, patch, model_id, tool_name, tool_version, payload_type) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"; @@ -376,25 +407,22 @@ fn parse_recent_diff_trace_patch_rows(rows: Vec) -> RecentDif let mut skipped = Vec::new(); for row in rows { - let parse_result = match row.payload_type.as_str() { - PAYLOAD_TYPE_PATCH => parse_patch(&row.patch, Some(row.session_id.as_str())) - .map_err(|error| skipped_diff_trace_patch_reason(&error)), - PAYLOAD_TYPE_STRUCTURED => match serde_json::from_str::(&row.patch) { - Ok(payload) => match derive_claude_structured_patch( - "PostToolUse", - &payload, - u64::try_from(row.time_ms).expect("diff trace time_ms should be non-negative"), - row.tool_version.as_deref(), - ) { - ClaudeStructuredPatchDerivationResult::Derived(derived) => Ok(derived.patch), - ClaudeStructuredPatchDerivationResult::Skipped(reason) => { - Err(reason.to_string()) - } - }, - Err(error) => Err(format!("invalid structured payload JSON: {error}")), - }, - other => Err(format!("unsupported diff-trace payload_type: {other}")), - }; + let parse_result = normalize_diff_trace_payload( + row.payload_type.as_str(), + &row.patch, + row.session_id.as_str(), + row.time_ms, + row.tool_version.as_deref(), + ) + .map_err(|error| { + if row.payload_type == PAYLOAD_TYPE_PATCH { + error + .downcast_ref::() + .map_or_else(|| error.to_string(), skipped_diff_trace_patch_reason) + } else { + error.to_string() + } + }); match parse_result { Ok(mut patch) => { @@ -460,6 +488,91 @@ mod tests { ) } + #[test] + fn patch_payload_normalization_strictly_dispatches_supported_payloads() { + let patch = valid_patch("notes/strict.md", "strict payload"); + let parsed_patch = + normalize_diff_trace_payload(PAYLOAD_TYPE_PATCH, &patch, "oc_strict", 1_000, None) + .expect("unified patch should normalize"); + assert_eq!(parsed_patch.files.len(), 1); + assert_eq!(parsed_patch.files[0].new_path, "notes/strict.md"); + + let fixture_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join( + "src/services/structured_patch/fixtures/edit_single_hunk/claude-post-tool-use.json", + ); + let structured_payload = + fs::read_to_string(fixture_path).expect("structured fixture should be readable"); + let parsed_structured = normalize_diff_trace_payload( + PAYLOAD_TYPE_STRUCTURED, + &structured_payload, + "cc_strict", + 1_000, + Some("claude-test"), + ) + .expect("structured payload should normalize"); + assert_eq!(parsed_structured.files.len(), 1); + assert_eq!(parsed_structured.files[0].new_path, "poem.md"); + } + + #[test] + fn patch_payload_normalization_rejects_malformed_and_unsupported_payloads() { + let malformed_patch = normalize_diff_trace_payload( + PAYLOAD_TYPE_PATCH, + "Index: notes/malformed.md\n===================================================================\n--- notes/malformed.md\n+++ notes/malformed.md\n@@ malformed @@\n+bad\n", + "oc_malformed", + 1_000, + None, + ) + .expect_err("malformed unified patch should fail"); + assert!(malformed_patch.to_string().contains("patch parse error")); + + let malformed_structured = normalize_diff_trace_payload( + PAYLOAD_TYPE_STRUCTURED, + "{not-json}", + "cc_malformed", + 1_000, + None, + ) + .expect_err("malformed structured payload should fail"); + assert!(malformed_structured + .to_string() + .contains("invalid structured payload JSON")); + + let unsupported_structured = normalize_diff_trace_payload( + PAYLOAD_TYPE_STRUCTURED, + r#"{"hook_event_name":"PostToolUse","session_id":"cc_unknown","tool_name":"Unknown"}"#, + "cc_unknown", + 1_000, + None, + ) + .expect_err("unsupported structured tool should fail"); + assert!(unsupported_structured + .to_string() + .contains("unsupported Claude tool")); + + let future_payload = + normalize_diff_trace_payload("future", "anything", "future-session", 1_000, None) + .expect_err("future payload type should fail"); + assert!(future_payload + .to_string() + .contains("unsupported diff-trace payload_type: future")); + } + + #[test] + fn patch_payload_normalization_rejects_negative_structured_time() { + let error = normalize_diff_trace_payload( + PAYLOAD_TYPE_STRUCTURED, + "{}", + "cc_negative-time", + -1, + None, + ) + .expect_err("negative structured time should fail"); + assert!(error + .to_string() + .contains("structured diff-trace time_ms must be non-negative")); + } + fn insert_test_diff_trace( db: &RepositoryAgentTraceDb, time_ms: i64, diff --git a/cli/src/services/agent_trace_dwh_replica/replica.rs b/cli/src/services/agent_trace_dwh_replica/replica.rs index 6ca46cf9..4f6597f2 100644 --- a/cli/src/services/agent_trace_dwh_replica/replica.rs +++ b/cli/src/services/agent_trace_dwh_replica/replica.rs @@ -20,6 +20,7 @@ use crate::services::{ agent_trace_dwh_db::{AgentTraceDwhDb, AgentTraceDwhSchemaState}, agent_trace_dwh_replica::lock::{BridgeLock, BridgeLockError}, agent_trace_etl::{AgentTraceEtl, AgentTraceEtlStats}, + code_changes_etl::{CodeChangesEtl, CodeChangesEtlStats}, }; /// Explicit caller-supplied configuration for opening an @@ -176,6 +177,17 @@ impl AgentTraceDwhReplica { etl.run(repository_id, source, self) } + /// Run the incremental code-change ETL while this replica owns its bridge + /// lock. Pull/push and credential handling remain explicit caller work. + pub fn run_code_changes_etl( + &self, + repository_id: &str, + source: &RepositoryAgentTraceDb, + etl: CodeChangesEtl, + ) -> anyhow::Result { + etl.run(repository_id, source, self) + } + /// Pull remote changes into the local replica. /// /// Returns `true` if any changes were applied. Non-mutating for already diff --git a/cli/src/services/code_changes_etl.rs b/cli/src/services/code_changes_etl.rs new file mode 100644 index 00000000..e4136d72 --- /dev/null +++ b/cli/src/services/code_changes_etl.rs @@ -0,0 +1,1262 @@ +//! Bounded source extraction and deterministic transformation for code changes. +//! +//! This module deliberately stops at destination-independent values. It copies +//! `diff_traces` rows out of a short source snapshot, then strictly normalizes +//! payloads and derives code-change metrics after that snapshot has ended. + +use std::fmt::Write; + +use anyhow::{bail, Context, Result}; +use sha2::{Digest, Sha256}; + +use crate::services::{ + agent_trace_db::{normalize_diff_trace_payload, repository::RepositoryAgentTraceDb}, + 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, + }, + patch::{ParsedPatch, TouchedLineKind}, +}; + +/// The source table represented by this pipeline. +pub const CODE_CHANGES_SOURCE_TABLE: &str = "diff_traces"; + +/// Default number of source diff traces processed by one batch. +pub const DEFAULT_CODE_CHANGES_ETL_BATCH_SIZE: u32 = 500; + +/// One diff-trace row copied out of a short repository source snapshot. +/// +/// Values are owned source copies. No parsing, hashing, or metric derivation +/// occurs while the source read transaction is open. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SourceDiffTrace { + pub id: i64, + pub time_ms: i64, + pub session_id: String, + pub patch: String, + pub model_id: Option, + pub tool_name: Option, + pub tool_version: Option, + pub payload_type: String, +} + +/// Destination-independent metrics derived from a canonical parsed patch. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct CodeChangeMetrics { + pub files_changed: i64, + pub lines_added: i64, + pub lines_removed: i64, +} + +/// One source diff trace after strict normalization and deterministic +/// transformation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TransformedCodeChange { + pub source_row_id: i64, + pub session_id: String, + pub time_ms: i64, + pub model_id: Option, + pub tool_name: Option, + pub tool_version: Option, + pub payload_type: String, + pub files_changed: i64, + pub lines_added: i64, + pub lines_removed: i64, + pub patch_sha256: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct StoredCodeChange { + session_id: String, + time_ms: i64, + model_id: Option, + tool_name: String, + tool_version: Option, + payload_type: String, + files_changed: i64, + lines_added: i64, + lines_removed: i64, + patch_sha256: String, +} + +/// Counts returned by one atomically loaded code-change batch. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct CodeChangesBatchStats { + pub inserted: u64, + pub already_present: u64, + pub watermark: i64, +} + +/// Configuration for the incremental `diff_traces` to `code_changes` ETL. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct CodeChangesEtl { + batch_size: u32, +} + +impl Default for CodeChangesEtl { + fn default() -> Self { + Self { + batch_size: DEFAULT_CODE_CHANGES_ETL_BATCH_SIZE, + } + } +} + +impl CodeChangesEtl { + /// Create a runner with the requested bounded source batch size. + pub fn with_batch_size(batch_size: u32) -> Result { + validate_batch_size(batch_size, "code changes ETL")?; + Ok(Self { batch_size }) + } + + /// Return the configured source batch size. + pub fn batch_size(self) -> u32 { + self.batch_size + } + + /// Run the incremental code-change ETL through an open DWH replica. + /// + /// Source metadata is verified before extraction. The runner owns neither + /// replica transport nor credentials: callers explicitly decide when to + /// pull or push the lock-owning 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(), + ) + } +} + +/// Summary of one complete incremental code-change ETL run. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct CodeChangesEtlStats { + pub extracted: u64, + pub inserted: u64, + pub already_present: u64, + pub batches: u64, + pub before_watermark: i64, + pub after_watermark: i64, +} + +const SELECT_DIFF_TRACES_BATCH_SQL: &str = + "SELECT id, time_ms, session_id, patch, model_id, tool_name, tool_version, payload_type +FROM diff_traces +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +/// Extract one bounded, ascending `diff_traces` batch in a short read +/// transaction. +/// +/// The transaction commits before this function returns, so callers can parse, +/// hash, and derive metrics without retaining a source snapshot. Only the +/// shared transient lock-contention retry is applied. +pub fn extract_diff_trace_batch( + db: &RepositoryAgentTraceDb, + watermark: i64, + batch_size: u32, +) -> Result> { + validate_batch_size(batch_size, "diff trace extraction")?; + + run_with_source_contention_retry( + |_attempt| { + db.read_transaction(|txn| { + txn.query_map( + SELECT_DIFF_TRACES_BATCH_SQL, + (watermark, i64::from(batch_size)), + source_diff_trace_from_row, + ) + }) + }, + || db.rollback_best_effort(), + ) +} + +fn source_diff_trace_from_row(row: &turso::Row) -> Result { + Ok(SourceDiffTrace { + id: row.get(0).context("failed to read diff_traces.id")?, + time_ms: row.get(1).context("failed to read diff_traces.time_ms")?, + session_id: row + .get(2) + .context("failed to read diff_traces.session_id")?, + patch: row.get(3).context("failed to read diff_traces.patch")?, + model_id: row.get(4).context("failed to read diff_traces.model_id")?, + tool_name: row.get(5).context("failed to read diff_traces.tool_name")?, + tool_version: row + .get(6) + .context("failed to read diff_traces.tool_version")?, + payload_type: row + .get(7) + .context("failed to read diff_traces.payload_type")?, + }) +} + +/// Derive checked destination-sized metrics from a canonical parsed patch. +pub fn derive_code_change_metrics(parsed: &ParsedPatch) -> Result { + let files_changed = i64::try_from(parsed.files.len()) + .context("code-change files_changed does not fit in destination integer")?; + let mut lines_added = 0_i64; + let mut lines_removed = 0_i64; + + for file in &parsed.files { + for hunk in &file.hunks { + for line in &hunk.lines { + match line.kind { + TouchedLineKind::Added => { + lines_added = lines_added + .checked_add(1) + .context("code-change lines_added overflowed destination integer")?; + } + TouchedLineKind::Removed => { + lines_removed = lines_removed + .checked_add(1) + .context("code-change lines_removed overflowed destination integer")?; + } + } + } + } + } + + Ok(CodeChangeMetrics { + files_changed, + lines_added, + lines_removed, + }) +} + +/// Transform an extracted source row after the source snapshot has ended. +/// +/// Payload normalization is strict for both supported payload types. The hash +/// is computed from the exact original UTF-8 payload bytes, not normalized +/// patch output. +pub fn transform_code_change(source: &SourceDiffTrace) -> Result { + let parsed = normalize_diff_trace_payload( + source.payload_type.as_str(), + &source.patch, + source.session_id.as_str(), + source.time_ms, + source.tool_version.as_deref(), + )?; + let metrics = derive_code_change_metrics(&parsed)?; + + Ok(TransformedCodeChange { + source_row_id: source.id, + session_id: source.session_id.clone(), + time_ms: source.time_ms, + model_id: source.model_id.clone(), + tool_name: source.tool_name.clone(), + tool_version: source.tool_version.clone(), + payload_type: source.payload_type.clone(), + files_changed: metrics.files_changed, + lines_added: metrics.lines_added, + lines_removed: metrics.lines_removed, + patch_sha256: sha256_hex(source.patch.as_bytes()), + }) +} + +/// Load one transformed source batch atomically into the DWH `code_changes` +/// table. Transformation happens before the destination transaction starts. +pub fn load_code_change_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + source_rows: &[SourceDiffTrace], +) -> Result { + let transformed = source_rows + .iter() + .map(transform_code_change) + .collect::>>()?; + load_transformed_code_change_batch(db, repository_id, source_instance_id, &transformed) +} + +fn load_transformed_code_change_batch( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedCodeChange], +) -> Result { + load_transformed_code_change_batch_with_failure( + db, + repository_id, + source_instance_id, + rows, + None, + ) +} + +fn load_transformed_code_change_batch_with_failure( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, + rows: &[TransformedCodeChange], + fail_after_row: Option, +) -> Result { + let Some(last_row) = rows.last() else { + return Ok(CodeChangesBatchStats::default()); + }; + if rows.iter().any(|row| row.tool_name.is_none()) { + bail!("diff_traces.tool_name cannot be null for code-change loading"); + } + + 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 = existing_code_change(txn, repository_id, source_instance_id, row.source_row_id)?; + + if let Some(existing) = existing.into_iter().next() { + let incoming = StoredCodeChange { + session_id: row.session_id.clone(), + time_ms: row.time_ms, + model_id: row.model_id.clone(), + tool_name: row + .tool_name + .clone() + .expect("tool_name validated before transaction"), + tool_version: row.tool_version.clone(), + payload_type: row.payload_type.clone(), + files_changed: row.files_changed, + lines_added: row.lines_added, + lines_removed: row.lines_removed, + patch_sha256: row.patch_sha256.clone(), + }; + if existing != incoming { + bail!( + "code change integrity conflict for repository {repository_id}, source instance {source_instance_id}, source diff trace {}", + row.source_row_id + ); + } + stats.already_present += 1; + } else { + txn.execute( + "INSERT INTO code_changes + (repository_id, source_instance_id, source_diff_trace_id, session_id, + time_ms, model_id, tool_name, tool_version, payload_type, files_changed, + lines_added, lines_removed, patch_sha256) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)", + ( + repository_id, + source_instance_id, + row.source_row_id, + row.session_id.as_str(), + row.time_ms, + row.model_id.as_deref(), + row.tool_name + .as_deref() + .expect("tool_name validated before transaction"), + row.tool_version.as_deref(), + row.payload_type.as_str(), + row.files_changed, + row.lines_added, + row.lines_removed, + row.patch_sha256.as_str(), + ), + )?; + stats.inserted += 1; + } + + if fail_after_row == Some(index + 1) { + bail!( + "injected code changes destination failure after row {}", + index + 1 + ); + } + } + + upsert_watermark( + txn, + repository_id, + source_instance_id, + CODE_CHANGES_SOURCE_TABLE, + last_row.source_row_id, + )?; + + Ok(CodeChangesBatchStats { + inserted: stats.inserted, + already_present: stats.already_present, + watermark: stats.watermark, + }) + }) +} + +fn existing_code_change( + txn: &TursoTransaction<'_, crate::services::agent_trace_dwh_db::AgentTraceDwhDbSpec>, + repository_id: &str, + source_instance_id: &str, + source_row_id: i64, +) -> Result> { + let existing = txn.query_map( + "SELECT session_id, time_ms, model_id, tool_name, tool_version, payload_type, + files_changed, lines_added, lines_removed, patch_sha256 + FROM code_changes + WHERE repository_id = ?1 AND source_instance_id = ?2 + AND source_diff_trace_id = ?3", + (repository_id, source_instance_id, source_row_id), + |db_row| { + Ok(StoredCodeChange { + session_id: db_row.get(0)?, + time_ms: db_row.get(1)?, + model_id: db_row.get(2)?, + tool_name: db_row.get(3)?, + tool_version: db_row.get(4)?, + payload_type: db_row.get(5)?, + files_changed: db_row.get(6)?, + lines_added: db_row.get(7)?, + lines_removed: db_row.get(8)?, + patch_sha256: db_row.get(9)?, + }) + }, + )?; + Ok(existing.into_iter().next()) +} + +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(()) +} + +/// Read the code-change watermark, treating an absent row as zero. +fn run_with_destination( + config: CodeChangesEtl, + repository_id: &str, + source_instance_id: &str, + source: &RepositoryAgentTraceDb, + destination: &AgentTraceDwhDb, +) -> Result { + let before_watermark = + read_code_changes_watermark(destination, repository_id, source_instance_id)?; + let mut watermark = before_watermark; + let mut stats = CodeChangesEtlStats { + before_watermark, + after_watermark: before_watermark, + ..Default::default() + }; + + loop { + let rows = extract_diff_trace_batch(source, watermark, config.batch_size)?; + if rows.is_empty() { + break; + } + + let batch = load_code_change_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 incremental code-change ETL through an open replica. +pub fn run_code_changes_etl( + repository_id: &str, + source: &RepositoryAgentTraceDb, + replica: &AgentTraceDwhReplica, +) -> Result { + CodeChangesEtl::default().run(repository_id, source, replica) +} + +/// Read the persisted code-change watermark, treating an absent row as zero. +pub fn read_code_changes_watermark( + db: &AgentTraceDwhDb, + repository_id: &str, + source_instance_id: &str, +) -> Result { + read_watermark( + db, + repository_id, + source_instance_id, + CODE_CHANGES_SOURCE_TABLE, + ) +} + +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 + }, + ) +} + +#[cfg(test)] +mod tests { + use std::{ + fs, + path::PathBuf, + sync::mpsc, + thread, + time::{SystemTime, UNIX_EPOCH}, + }; + + use super::*; + use crate::services::{ + agent_trace_db::{ + DiffTraceInsert, InsertMessageInsert, InsertPartInsert, MessageRole, PartType, + PAYLOAD_TYPE_PATCH, + }, + conversation_etl::{self, ConversationEtl}, + }; + + fn unique_path(label: &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-code-changes-etl-{label}-{}-{nonce}", + std::process::id() + )) + .join("agent-trace.db") + } + + fn clean(path: &std::path::Path) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } + } + + fn valid_patch(path: &str, added: &str, removed: Option<&str>) -> String { + let removed_line = removed.map(|line| format!("-{line}\n")).unwrap_or_default(); + format!( + "diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -1,1 +1,1 @@\n{removed_line}+{added}\n" + ) + } + + fn source_row(id: i64, patch: &str) -> SourceDiffTrace { + SourceDiffTrace { + id, + time_ms: 1_000 + id, + session_id: format!("session-{id}"), + patch: patch.to_string(), + model_id: Some(format!("provider/model-{id}")), + tool_name: Some("opencode".to_string()), + tool_version: Some("1.2.3".to_string()), + payload_type: PAYLOAD_TYPE_PATCH.to_string(), + } + } + + fn insert_source_diff_trace(db: &RepositoryAgentTraceDb, id: i64, session_id: &str) { + db.insert_diff_trace(DiffTraceInsert { + time_ms: id, + session_id, + patch: &valid_patch(&format!("file-{id}.rs"), "added", None), + model_id: Some("provider/model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("diff trace insert should succeed"); + } + + #[test] + fn code_changes_etl_defaults_and_validates_batch_size() { + assert_eq!(CodeChangesEtl::default().batch_size(), 500); + assert_eq!(CodeChangesEtl::with_batch_size(2).unwrap().batch_size(), 2); + assert!(CodeChangesEtl::with_batch_size(0).is_err()); + } + + #[test] + fn code_changes_etl_run_processes_incremental_growth_and_noop_reruns() { + let source_path = unique_path("run-source"); + let dwh_path = unique_path("run-dwh"); + let source = RepositoryAgentTraceDb::new_at(&source_path).expect("source DB should open"); + let destination = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let metadata = source + .verify_or_initialize_repository_metadata("repo-a") + .expect("source metadata should initialize"); + + for id in 1..=3 { + insert_source_diff_trace(&source, id, "session-growth"); + } + + let config = CodeChangesEtl::with_batch_size(2).unwrap(); + let first = run_with_destination( + config, + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("initial code-change ETL should succeed"); + assert_eq!( + first, + CodeChangesEtlStats { + extracted: 3, + inserted: 3, + already_present: 0, + batches: 2, + before_watermark: 0, + after_watermark: 3, + } + ); + + for id in 4..=5 { + insert_source_diff_trace(&source, id, "session-growth"); + } + let growth = run_with_destination( + config, + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("growth code-change ETL should succeed"); + assert_eq!(growth.extracted, 2); + assert_eq!(growth.inserted, 2); + assert_eq!(growth.batches, 1); + assert_eq!(growth.before_watermark, 3); + assert_eq!(growth.after_watermark, 5); + + let noop = run_with_destination( + config, + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("a no-op code-change rerun should succeed"); + assert_eq!(noop.extracted, 0); + assert_eq!(noop.inserted, 0); + assert_eq!(noop.already_present, 0); + assert_eq!(noop.batches, 0); + assert_eq!(noop.before_watermark, 5); + assert_eq!(noop.after_watermark, 5); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn code_changes_etl_replays_watermark_behind_failed_transformation() { + let source_path = unique_path("run-invalid"); + let dwh_path = unique_path("run-invalid-dwh"); + let source = RepositoryAgentTraceDb::new_at(&source_path).expect("source DB should open"); + let destination = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let metadata = source + .verify_or_initialize_repository_metadata("repo-a") + .expect("source metadata should initialize"); + insert_source_diff_trace(&source, 1, "session-invalid"); + source + .insert_diff_trace(DiffTraceInsert { + time_ms: 2, + session_id: "session-invalid", + patch: "Index: notes/malformed.md\n===================================================================\n--- notes/malformed.md\n+++ notes/malformed.md\n@@ malformed @@\n+bad\n", + model_id: None, + tool_name: "opencode", + tool_version: None, + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("invalid diff trace insert should succeed"); + insert_source_diff_trace(&source, 3, "session-invalid"); + + let error = run_with_destination( + CodeChangesEtl::with_batch_size(3).unwrap(), + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect_err("invalid transformation should fail the batch"); + assert!(error.to_string().contains("patch")); + assert_eq!( + read_code_changes_watermark(&destination, "repo-a", &metadata.source_instance_id) + .unwrap(), + 0 + ); + assert_eq!( + destination + .query_map("SELECT COUNT(*) FROM code_changes", (), |row| { + row.get::(0).map_err(Into::into) + }) + .unwrap(), + vec![0] + ); + + source + .execute( + "UPDATE diff_traces SET patch = ?1 WHERE id = 2", + (valid_patch("fixed.rs", "fixed", None),), + ) + .expect("invalid source row should be repairable for replay"); + let replay = run_with_destination( + CodeChangesEtl::with_batch_size(3).unwrap(), + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("repaired source rows should replay successfully"); + assert_eq!(replay.extracted, 3); + assert_eq!(replay.inserted, 3); + assert_eq!(replay.after_watermark, 3); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn code_changes_etl_runner_keeps_source_instance_watermarks_independent() { + let first_source_path = unique_path("lineage-runner-a"); + let second_source_path = unique_path("lineage-runner-b"); + let dwh_path = unique_path("lineage-runner-dwh"); + let first_source = + RepositoryAgentTraceDb::new_at(&first_source_path).expect("first source should open"); + let second_source = + RepositoryAgentTraceDb::new_at(&second_source_path).expect("second source should open"); + let destination = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let first_metadata = first_source + .verify_or_initialize_repository_metadata("repo-a") + .expect("first source metadata should initialize"); + let second_metadata = second_source + .verify_or_initialize_repository_metadata("repo-a") + .expect("second source metadata should initialize"); + assert_ne!( + first_metadata.source_instance_id, second_metadata.source_instance_id, + "independent source databases must have independent lineages" + ); + insert_source_diff_trace(&first_source, 1, "session-first-source"); + insert_source_diff_trace(&second_source, 1, "session-second-source"); + + let config = CodeChangesEtl::default(); + let first = run_with_destination( + config, + "repo-a", + &first_metadata.source_instance_id, + &first_source, + &destination, + ) + .expect("first source should load"); + let second = run_with_destination( + config, + "repo-a", + &second_metadata.source_instance_id, + &second_source, + &destination, + ) + .expect("second source should load independently"); + assert_eq!(first.inserted, 1); + assert_eq!(second.inserted, 1); + assert_eq!( + destination + .query_map("SELECT COUNT(*) FROM code_changes", (), |row| { + row.get::(0).map_err(Into::into) + }) + .unwrap(), + vec![2] + ); + assert_eq!( + read_code_changes_watermark(&destination, "repo-a", &first_metadata.source_instance_id) + .unwrap(), + 1 + ); + assert_eq!( + read_code_changes_watermark( + &destination, + "repo-a", + &second_metadata.source_instance_id + ) + .unwrap(), + 1 + ); + + clean(&first_source_path); + clean(&second_source_path); + clean(&dwh_path); + } + + #[test] + fn code_changes_session_relationship_preserves_session_only_conversation_relationship() { + let source_path = unique_path("session-relationship-source"); + let dwh_path = unique_path("session-relationship-dwh"); + let source = RepositoryAgentTraceDb::new_at(&source_path).expect("source DB should open"); + let destination = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let metadata = source + .verify_or_initialize_repository_metadata("repo-a") + .expect("source metadata should initialize"); + source + .insert_message(InsertMessageInsert { + session_id: "session-join".to_string(), + message_id: "message-1".to_string(), + role: MessageRole::User, + generated_at_unix_ms: 1_000, + }) + .expect("source message insert should succeed"); + source + .insert_part(InsertPartInsert { + part_type: PartType::Text, + text: "hello".to_string(), + session_id: "session-join".to_string(), + message_id: "message-1".to_string(), + generated_at_unix_ms: 1_001, + }) + .expect("source part insert should succeed"); + insert_source_diff_trace(&source, 1, "session-join"); + + conversation_etl::run_with_destination( + ConversationEtl::default(), + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("conversation ETL should load the session context"); + let code_changes = run_with_destination( + CodeChangesEtl::default(), + "repo-a", + &metadata.source_instance_id, + &source, + &destination, + ) + .expect("code-change ETL should load the session context"); + assert_eq!(code_changes.inserted, 1); + + let joined = destination + .query_map( + "SELECT c.session_id, m.message_id, p.text + FROM code_changes c + JOIN messages m ON m.repository_id = c.repository_id + AND m.session_id = c.session_id + JOIN message_parts p ON p.repository_id = m.repository_id + AND p.session_id = m.session_id AND p.message_id = m.message_id + WHERE c.repository_id = ?1", + ("repo-a",), + |row| { + Ok(( + row.get::(0)?, + row.get::(1)?, + row.get::(2)?, + )) + }, + ) + .expect("session-level code-change query should succeed"); + assert_eq!( + joined, + vec![( + String::from("session-join"), + String::from("message-1"), + String::from("hello") + )] + ); + let code_changes_schema = destination + .query_map( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'code_changes'", + (), + |row| row.get::(0).map_err(Into::into), + ) + .unwrap() + .into_iter() + .next() + .unwrap(); + assert!(!code_changes_schema.contains("message_id")); + + clean(&source_path); + clean(&dwh_path); + } + + #[test] + fn code_changes_lineage_contention_source_snapshot_does_not_block_concurrent_writers() { + let source_path = unique_path("source-contention"); + let source = RepositoryAgentTraceDb::new_at(&source_path).expect("source DB should open"); + insert_source_diff_trace(&source, 1, "session-contention"); + + 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 source DB should reopen"); + reader_db.read_transaction(|txn| { + let rows = txn.query_map( + SELECT_DIFF_TRACES_BATCH_SQL, + (0_i64, 10_i64), + source_diff_trace_from_row, + )?; + reader_ready_tx + .send(rows) + .expect("reader should signal its snapshot"); + release_reader_rx + .recv() + .expect("reader should wait while holding its snapshot"); + Ok(()) + }) + }); + + let rows = reader_ready_rx + .recv() + .expect("test should observe the source snapshot"); + let writer = RepositoryAgentTraceDb::open_without_migrations_at(&source_path) + .expect("writer source DB should reopen"); + insert_source_diff_trace(&writer, 2, "session-contention"); + release_reader_tx + .send(()) + .expect("test should release the source snapshot"); + reader + .join() + .expect("reader should not panic") + .expect("reader transaction should commit"); + assert_eq!(rows.len(), 1); + assert_eq!(extract_diff_trace_batch(&source, 0, 10).unwrap().len(), 2); + + clean(&source_path); + } + + #[test] + fn code_changes_etl_extraction_uses_zero_watermark_and_ascending_bounded_ids() { + let path = unique_path("extraction"); + let db = RepositoryAgentTraceDb::new_at(&path).expect("source DB should open"); + for id in 1..=3 { + db.insert_diff_trace(DiffTraceInsert { + time_ms: id, + session_id: "session", + patch: &valid_patch(&format!("file-{id}.rs"), "added", None), + model_id: None, + tool_name: "opencode", + tool_version: None, + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("diff trace insert should succeed"); + } + + let rows = extract_diff_trace_batch(&db, 0, 2).expect("bounded extraction should succeed"); + assert_eq!( + rows.iter().map(|row| row.id).collect::>(), + vec![1, 2] + ); + assert_eq!(rows[0].time_ms, 1); + assert_eq!(rows[0].session_id, "session"); + assert_eq!(rows[0].payload_type, PAYLOAD_TYPE_PATCH); + assert!(extract_diff_trace_batch(&db, 3, 2) + .expect("empty extraction should succeed") + .is_empty()); + + clean(&path); + } + + #[test] + fn code_changes_etl_extraction_rejects_zero_batch_size() { + let path = unique_path("zero-batch"); + let db = RepositoryAgentTraceDb::new_at(&path).expect("source DB should open"); + let error = extract_diff_trace_batch(&db, 0, 0).expect_err("zero batch must fail"); + assert!(error.to_string().contains("batch_size")); + clean(&path); + } + + #[test] + fn code_changes_metrics_count_files_and_all_touched_line_kinds() { + let first = source_row(1, &valid_patch("one.rs", "new", Some("old"))); + let second = source_row(2, &valid_patch("two.rs", "another", None)); + let first_transformed = transform_code_change(&first).expect("first patch should parse"); + let second_transformed = transform_code_change(&second).expect("second patch should parse"); + + assert_eq!( + ( + first_transformed.files_changed, + first_transformed.lines_added, + first_transformed.lines_removed, + ), + (1, 1, 1) + ); + assert_eq!( + ( + second_transformed.files_changed, + second_transformed.lines_added, + second_transformed.lines_removed, + ), + (1, 1, 0) + ); + } + + #[test] + fn code_changes_metrics_are_checked_destination_integers() { + let metrics = derive_code_change_metrics(&ParsedPatch { files: vec![] }) + .expect("empty metrics should fit destination integers"); + assert_eq!( + metrics, + CodeChangeMetrics { + files_changed: 0, + lines_added: 0, + lines_removed: 0 + } + ); + } + + #[test] + fn code_changes_hash_uses_exact_original_payload_bytes_and_lowercase_hex() { + let source = source_row(7, &valid_patch("hash.rs", "value", None)); + let transformed = transform_code_change(&source).expect("patch should parse"); + assert_eq!( + transformed.patch_sha256, + "16e02392f2925b987722f8981718dd6ea1ab26841ffc42ee55116e5fb33fbba3" + ); + assert_eq!( + transformed.patch_sha256, + transformed.patch_sha256.to_lowercase() + ); + } + + #[test] + fn code_changes_transformation_preserves_source_metadata_and_rejects_future_payloads() { + let source = source_row(3, &valid_patch("metadata.rs", "value", None)); + let transformed = transform_code_change(&source).expect("patch should parse"); + assert_eq!(transformed.source_row_id, source.id); + assert_eq!(transformed.session_id, source.session_id); + assert_eq!(transformed.time_ms, source.time_ms); + assert_eq!(transformed.model_id, source.model_id); + assert_eq!(transformed.tool_name, source.tool_name); + assert_eq!(transformed.tool_version, source.tool_version); + assert_eq!(transformed.payload_type, PAYLOAD_TYPE_PATCH); + + let mut future = source; + future.payload_type = "future".to_string(); + let error = transform_code_change(&future).expect_err("future payload must fail strictly"); + assert!(error + .to_string() + .contains("unsupported diff-trace payload_type")); + } + + #[test] + fn code_changes_transformation_normalizes_structured_payloads_and_preserves_metadata() { + let fixture_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join( + "src/services/structured_patch/fixtures/edit_single_hunk/claude-post-tool-use.json", + ); + let source = SourceDiffTrace { + id: 9, + time_ms: 2_000, + session_id: "cc-session".to_string(), + patch: fs::read_to_string(fixture_path).expect("structured fixture should be readable"), + model_id: Some("claude/model".to_string()), + tool_name: Some("claude".to_string()), + tool_version: Some("4.0".to_string()), + payload_type: "structured".to_string(), + }; + + let transformed = + transform_code_change(&source).expect("structured payload should normalize"); + assert_eq!(transformed.payload_type, "structured"); + assert_eq!(transformed.session_id, "cc-session"); + assert_eq!(transformed.model_id, Some("claude/model".to_string())); + assert_eq!(transformed.files_changed, 1); + assert!(transformed.lines_added > 0); + } + + #[test] + fn code_changes_identity_inserts_full_content_and_matching_replay_is_already_present() { + let dwh_path = unique_path("identity"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let row = source_row(7, &valid_patch("identity.rs", "new", Some("old"))); + + let first = + load_code_change_batch(&dwh, "repo-a", "instance-a", std::slice::from_ref(&row)) + .expect("first code-change load should insert"); + assert_eq!(first.inserted, 1); + assert_eq!(first.already_present, 0); + assert_eq!(first.watermark, 7); + assert_eq!( + read_code_changes_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 7 + ); + + let values = dwh + .query_map( + "SELECT source_instance_id, source_diff_trace_id, session_id, time_ms, model_id, + tool_name, tool_version, payload_type, files_changed, lines_added, + lines_removed, patch_sha256 + FROM code_changes", + (), + |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)?, + db_row.get::>(6)?, + db_row.get::(7)?, + db_row.get::(8)?, + db_row.get::(9)?, + db_row.get::(10)?, + db_row.get::(11)?, + )) + }, + ) + .expect("code-change content query should succeed"); + assert_eq!(values.len(), 1); + assert_eq!(values[0].0, "instance-a"); + assert_eq!(values[0].1, 7); + assert_eq!(values[0].2, "session-7"); + assert_eq!(values[0].4, Some("provider/model-7".to_string())); + assert_eq!(values[0].5, "opencode"); + assert_eq!(values[0].7, PAYLOAD_TYPE_PATCH); + assert_eq!((values[0].8, values[0].9, values[0].10), (1, 1, 1)); + + let replay = load_code_change_batch(&dwh, "repo-a", "instance-a", &[row]) + .expect("matching replay should succeed"); + assert_eq!(replay.inserted, 0); + assert_eq!(replay.already_present, 1); + assert_eq!( + dwh.query_map("SELECT COUNT(*) FROM code_changes", (), |db_row| db_row + .get::(0) + .map_err(Into::into)) + .unwrap(), + vec![1] + ); + + clean(&dwh_path); + } + + #[test] + fn code_changes_identity_conflict_fails_without_overwrite_or_watermark_change() { + let dwh_path = unique_path("identity-conflict"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let original = source_row(1, &valid_patch("conflict.rs", "original", None)); + load_code_change_batch( + &dwh, + "repo-a", + "instance-a", + std::slice::from_ref(&original), + ) + .expect("original code change should insert"); + + let changed = source_row(1, &valid_patch("conflict.rs", "changed", None)); + let error = load_code_change_batch(&dwh, "repo-a", "instance-a", &[changed]) + .expect_err("changed synchronized content must fail"); + assert!(error.to_string().contains("code change integrity conflict")); + assert_eq!( + read_code_changes_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 1 + ); + assert_eq!( + dwh.query_map( + "SELECT lines_added, patch_sha256 FROM code_changes", + (), + |db_row| { Ok((db_row.get::(0)?, db_row.get::(1)?)) } + ) + .unwrap() + .len(), + 1 + ); + + clean(&dwh_path); + } + + #[test] + fn code_changes_atomic_destination_failure_rolls_back_dimensions_facts_and_watermark() { + let dwh_path = unique_path("atomic"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let rows = vec![ + transform_code_change(&source_row(1, &valid_patch("one.rs", "one", None))).unwrap(), + transform_code_change(&source_row(2, &valid_patch("two.rs", "two", None))).unwrap(), + ]; + + load_transformed_code_change_batch_with_failure( + &dwh, + "repo-a", + "instance-a", + &rows, + Some(1), + ) + .expect_err("injected destination failure should roll back the batch"); + + for table in [ + "repositories", + "source_instances", + "code_changes", + "etl_watermarks", + ] { + assert_eq!( + dwh.query_map(&format!("SELECT COUNT(*) FROM {table}"), (), |db_row| { + db_row.get::(0).map_err(Into::into) + }) + .unwrap(), + vec![0], + "{table} should be rolled back" + ); + } + + let replay = load_code_change_batch( + &dwh, + "repo-a", + "instance-a", + &[ + source_row(1, &valid_patch("one.rs", "one", None)), + source_row(2, &valid_patch("two.rs", "two", None)), + ], + ) + .expect("the failed batch should be replayable"); + assert_eq!(replay.inserted, 2); + assert_eq!(replay.watermark, 2); + + clean(&dwh_path); + } + + #[test] + fn code_changes_same_local_identity_coexists_across_source_instances() { + let dwh_path = unique_path("lineage"); + let dwh = AgentTraceDwhDb::new_at(&dwh_path).expect("DWH DB should open"); + let row = source_row(1, &valid_patch("lineage.rs", "same local id", None)); + + load_code_change_batch(&dwh, "repo-a", "instance-a", std::slice::from_ref(&row)) + .expect("first source instance should load"); + load_code_change_batch(&dwh, "repo-a", "instance-b", &[row]) + .expect("second source instance should load independently"); + + assert_eq!( + dwh.query_map( + "SELECT COUNT(*) FROM code_changes WHERE repository_id = 'repo-a'", + (), + |db_row| db_row.get::(0).map_err(Into::into) + ) + .unwrap(), + vec![2] + ); + assert_eq!( + read_code_changes_watermark(&dwh, "repo-a", "instance-a").unwrap(), + 1 + ); + assert_eq!( + read_code_changes_watermark(&dwh, "repo-a", "instance-b").unwrap(), + 1 + ); + + clean(&dwh_path); + } +} diff --git a/cli/src/services/mod.rs b/cli/src/services/mod.rs index 80d8c20a..2358bcf4 100644 --- a/cli/src/services/mod.rs +++ b/cli/src/services/mod.rs @@ -16,6 +16,8 @@ pub mod auth_db; pub mod bash_policy; pub mod capabilities; pub mod checkout; +#[allow(dead_code)] +pub mod code_changes_etl; pub mod command_registry; pub mod completion; pub mod config; diff --git a/context/architecture.md b/context/architecture.md index 01804300..4c78c9e2 100644 --- a/context/architecture.md +++ b/context/architecture.md @@ -124,8 +124,8 @@ The repository includes a new placeholder Rust binary crate at `cli/`. - `cli/src/services/auth_db/mod.rs` provides the encrypted auth DB spec and `AuthDb` type alias over `EncryptedTursoDb`. `AuthDbSpec` resolves `/sce/auth.db` through the shared default-path seam and embeds ordered auth migrations. Auth DB lifecycle setup/doctor integration is wired through `AuthDbLifecycle`; auth command/token-storage reads/writes are directed through `token_storage.rs`. - `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 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/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()`, `run_code_changes_etl()`, `pull()`, and `push()`; the ETL methods delegate 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 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. `code_changes_etl.rs` adds the strict, bounded `diff_traces` to `code_changes` bridge: it preserves source metadata, hashes the exact original payload, derives checked `ParsedPatch` metrics, validates source-lineage identity, and atomically advances the independent watermark. Its only conversation relationship is `session_id`; no `message_id` or message causality is inferred. The `CodeChangesEtl` and conversation runners 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/cli/structured-patch-service.md b/context/cli/structured-patch-service.md index f4835a4a..5dc749c3 100644 --- a/context/cli/structured-patch-service.md +++ b/context/cli/structured-patch-service.md @@ -28,7 +28,7 @@ The module is wired into `sce hooks diff-trace` for Claude payload classification at intake (T04): when `hook_event_name` is present and the event is a supported `PostToolUse` (`Write` structured update, `Write` content create fallback, or `Edit` structured patch), the raw JSON is persisted as a `structured` payload type in `diff_traces` without conversion to unified-diff text. Unsupported Claude events (non-`PostToolUse`, unsupported tools) produce deterministic no-op results. OpenCode normalized payloads continue to be stored as `patch` payloads unchanged. -Post-commit parsing dispatch through `structured_patch.rs` is implemented (T05): `RepositoryAgentTraceDb::recent_diff_trace_patches` now reads `payload_type` from each `diff_traces` row and dispatches `patch` rows through existing `parse_patch` while dispatching `structured` rows through `derive_claude_structured_patch` at read time, producing `ParsedPatch` for both paths before hunk `model_id` injection and downstream combine/intersect operations. +Post-commit parsing dispatch through `structured_patch.rs` is implemented (T05): `RepositoryAgentTraceDb::recent_diff_trace_patches` now reads `payload_type` from each `diff_traces` row and uses the shared `agent_trace_db::normalize_diff_trace_payload` boundary. That helper dispatches `patch` rows through `parse_patch` and `structured` rows through the existing `derive_claude_structured_patch` logic, producing `ParsedPatch` for both paths before hunk `model_id` injection and downstream combine/intersect operations. Strict ETL callers receive explicit errors for malformed, unsupported, or future payload types; the recent reader maps those errors back to its existing skipped-row accounting. ## Test status diff --git a/context/context-map.md b/context/context-map.md index 05dab237..79aada33 100644 --- a/context/context-map.md +++ b/context/context-map.md @@ -62,9 +62,10 @@ Feature/domain context: - `context/sce/shared-turso-db.md` (current shared `cli/src/services/db/mod.rs` Turso database infrastructure seam, including `DbSpec`, generic `TursoDb`, encrypted `EncryptedTursoDb`, build-time generated migration constants from `cli/build.rs`/Cargo `OUT_DIR`, config-driven constructor/open-connect retry via `run_with_retry_sync`, no-migration `TursoDb::open_without_migrations()` / explicit-path `open_without_migrations_at(path)` for hot runtime paths, migration-running `new()` / explicit-path `new_at(path)` / `run_migrations()` with per-database `__sce_migrations` tracking, config-driven operation retry for `execute`/`query`/`query_values`/`query_map` with a `<= 2_000ms` default query failure budget, raw-value row fetching for deterministic operator-facing rendering, row-mapping excluded from retry, generic embedded migration execution, non-mutating `migration_metadata_problems()` and `ensure_schema_ready(setup_guidance)` readiness methods on `TursoDb`, one-level explicit `transaction()`/`read_transaction()`/`TursoTransaction` seam (`transaction()` issues `BEGIN IMMEDIATE`, `read_transaction()` issues a plain non-blocking `BEGIN`; both commit-on-`Ok` with best-effort rollback on closure error or failed commit, no nested transactions/retry), and concrete wrappers for `LocalDb`, `AuthDb`, plus `RepositoryAgentTraceDb`) - `context/sce/auth-db.md` (encrypted `AuthDb = EncryptedTursoDb` adapter, canonical `/sce/auth.db` path, build-time generated `AUTH_MIGRATIONS` from `cli/migrations/auth/`, auth credential schema and updated-at trigger baseline, lifecycle setup/doctor integration, encrypted token-storage persistence, and `SCE_AUTH_DB_ENCRYPTION_KEY`/OS credential-store key handling) - `context/sce/agent-trace-db.md` (implemented Agent Trace database adapter: the sole repository-scoped `RepositoryAgentTraceDb` backed by one fresh multi-statement baseline schema file plus an additive `002_repository_source_instance_id` migration, typed `RepositoryMetadata { repository_id, source_instance_id }` with atomic once-only source-instance initialization, `repository_metadata` validation, narrow concurrent-first-open repair for missing one-file baseline migration metadata after all required schema tables exist, no trace-table `checkout_id` columns, repository-level typed insert helpers for diff traces, post-commit intersections, Agent Trace rows, messages, and parts, repository-level recent diff-trace reads without checkout filtering, on-demand command/hook initialization with no daemon/background service, and the never-touch on-disk boundary for any pre-migration checkout-scoped/global DB files; the checkout-scoped `AgentTraceDb` adapter, its `agent_trace_db_path()`/`agent_trace_db_path_for_checkout()` helpers, and the 15-file `cli/migrations/agent-trace/` chain were removed by the `retire-legacy-agent-trace-db` plan; active hook writers/readers and Agent Trace setup/lifecycle resolve repository storage through `agent_trace_storage`) -- `context/sce/agent-trace-dwh-replica.md` (Agent Trace DWH Turso Sync replica boundary: `AgentTraceDwhReplica` in `cli/src/services/agent_trace_dwh_replica/replica.rs`, the sole owner of a Turso Sync connection to the repository-scoped `agent-trace-sync.db`; acquires the `BridgeLock` before any Turso access, opens via `turso::sync::Builder` without enabling multiprocess WAL, then classifies the DWH schema via `AgentTraceDwhDb::classify_schema_state()` — a `Ready` schema is left untouched, a genuinely `Empty` schema is initialized locally with `run_migrations()` and published with `push()`, and an `Incompatible` schema fails loudly without repair — exposes lock-lifetime-bound `run_agent_trace_etl()` 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` (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/agent-trace-dwh-replica.md` (Agent Trace DWH Turso Sync replica boundary: `AgentTraceDwhReplica` in `cli/src/services/agent_trace_dwh_replica/replica.rs`, the sole owner of a Turso Sync connection to the repository-scoped `agent-trace-sync.db`; acquires the `BridgeLock` before any Turso access, opens via `turso::sync::Builder` without enabling multiprocess WAL, then classifies the DWH schema via `AgentTraceDwhDb::classify_schema_state()` — a `Ready` schema is left untouched, a genuinely `Empty` schema is initialized locally with `run_migrations()` and published with `push()`, and an `Incompatible` schema fails loudly without repair — exposes lock-lifetime-bound `run_agent_trace_etl()` and `run_code_changes_etl()` plus explicit `pull()`/`push()`, redacts the caller-supplied auth token from every error, and reuses a new `TursoDb::from_connection`/`block_on` seam; credential discovery/persistence and CLI/lifecycle wiring remain deferred) +- `context/sce/agent-trace-dwh-db.md` (Agent Trace DWH: a separate append-oriented destination schema for the CLI-independent ETL consumer, distinct from the repository-scoped source schema above. `AgentTraceDwhDb = TursoDb` in `cli/src/services/agent_trace_dwh_db/mod.rs`, explicit-path only, no lifecycle/CLI wiring yet. One fresh baseline `cli/migrations/agent-trace-dwh/001_dwh_schema.sql` creates `repositories`, `source_instances`, `etl_watermarks`, `messages`, `message_parts`, `agent_traces`, and `code_changes` with no foreign keys; every fact table denormalizes `repository_id`/`source_instance_id` lineage as plain text. Deterministic logical identities (messages, Agent Traces) are unique excluding `source_instance_id` for cross-source-database idempotency; raw local source row IDs (`source_part_id`, `source_diff_trace_id`) are unique per source instance so the same local integer coexists across sources/repositories) +- `context/sce/agent-trace-etl.md` (shared Agent Trace ETL mechanics and the `agent_traces`, `messages`, `parts`, and `code_changes` bridges between the `agent-trace.db` source and DWH replica; covers bounded extraction, exact-content transformation/hashing, table-specific identity validation, atomic per-lineage watermark advancement, source contention handling, stats, replica-owned orchestration, and the code-change session-only relationship) +- `context/sce/code-changes-etl.md` (CLI-independent `CodeChangesEtl` contract for the exact ordered `diff_traces` projection, short source snapshots, strict `patch`/`structured` normalization, source-lineage watermarks, exact payload hashing, checked patch metrics, atomic replay/conflict handling, replica-owned execution, and the session-only relationship to DWH conversations without message-level causality) - `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) @@ -118,4 +119,5 @@ 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-code-changes-etl-watermark-and-replay-contract.md` (code-change ETL uses source-instance-scoped integer-ID watermarks, strict pre-transaction transformation, atomic fact/lineage/progress loading, explicit replay conflicts, and session-only conversation relationships) - `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-code-changes-etl-watermark-and-replay-contract.md b/context/decisions/2026-08-08-code-changes-etl-watermark-and-replay-contract.md new file mode 100644 index 00000000..331335cd --- /dev/null +++ b/context/decisions/2026-08-08-code-changes-etl-watermark-and-replay-contract.md @@ -0,0 +1,85 @@ +# Decision: Use source-lineage ID watermarks with atomic strict replay for code-change ETL + +Date: 2026-08-08 +Status: Accepted +Plan: `context/plans/incremental-code-changes-etl.md` +Task: T01, T02, T03, T04, T05 + +## Context + +Repository-scoped Agent Trace `diff_traces` rows use local integer IDs and may +be copied into the append-oriented DWH from independently created source +instances. The code-change bridge must preserve source lineage, avoid holding +source snapshots during transformation, and make failed batches replayable +without duplicate facts or silently overwriting an existing identity. + +## Decision + +The code-change ETL uses an independent `(repository_id, source_instance_id, +diff_traces)` watermark over ascending `diff_traces.id`; it strictly transforms +an entire bounded batch before destination work, then validates or inserts +`code_changes` rows and advances that watermark in one destination transaction. + +## Rationale + +Integer source IDs provide deterministic, lossless progress without timestamp +ambiguity, while source-instance scoping allows equal local IDs from separate +source databases to coexist. Transforming before opening the destination +transaction prevents malformed payloads from creating partial state. Comparing +all synchronized and derived fields on replay makes conflicts explicit, and the +single transaction preserves dimensions, facts, and progress as one replayable +unit. + +## Alternatives considered + +- **Timestamp-based progress** — not selected because timestamps are not unique + progress identifiers and can miss or reorder source rows. +- **Conflict-ignore or overwrite loading** — not selected because either hides + source inconsistencies or mutates an established fact without evidence. +- **Advancing the watermark per row** — not selected because a later batch + failure would leave partial facts and progress that cannot be replayed as one + unit. + +## Compatibility and risks + +- The contract is additive to the existing source capture schema and DWH + destination; hooks and transport synchronization remain outside this ETL. +- A malformed or unsupported source payload blocks the batch until the source + row is corrected, which is intentional strict behavior; the unchanged + watermark permits safe replay. +- The local source ID is not globally unique, so every code-change identity and + watermark lookup must retain `source_instance_id` as a guardrail. + +## Guardrails + +- Extract only the authored ordered `diff_traces` projection and never use a + timestamp or separate `MAX(id)` query for progress. +- End the plain source read transaction before parsing, hashing, metrics, or + destination work; retry only the shared transient source contention cases. +- Use exactly `(repository_id, source_instance_id, source_diff_trace_id)` for + code-change identity and compare every synchronized and derived field before + counting a replay. +- Keep pull/push, credentials, CLI orchestration, and message-level attribution + outside the ETL boundary; `session_id` is the only conversation relationship. + +## Consequences + +- Code-change ingestion is deterministic, independently watermarked, and safe + to rerun after transformation, integrity, or destination failures. +- Source writers can continue while ETL transforms extracted owned values, and + failed batches leave dimensions, facts, and progress unchanged. +- Future changes to source update semantics or message-level attribution need a + separate design rather than weakening this append-only contract. + +## Follow-up + +None. + +## References + +- Plan: [`incremental-code-changes-etl`](../plans/incremental-code-changes-etl.md) +- Task: `T01, T02, T03, T04, T05` +- Current-state context: [`code-changes-etl.md`](../sce/code-changes-etl.md), [`agent-trace-etl.md`](../sce/agent-trace-etl.md), [`agent-trace-dwh-db.md`](../sce/agent-trace-dwh-db.md) +- Evidence: [`code_changes_etl.rs`](../../cli/src/services/code_changes_etl.rs), [`code_changes_etl` tests](../../cli/src/services/code_changes_etl.rs) +- Related decision: [`agent-trace-dwh-schema-identity-contract`](2026-08-08-agent-trace-dwh-schema-identity-contract.md) +- Related decision: [`agent-trace-dwh-turso-sync-replica-ownership`](2026-08-08-agent-trace-dwh-turso-sync-replica-ownership.md) diff --git a/context/glossary.md b/context/glossary.md index bbad2dbd..6c8a6d28 100644 --- a/context/glossary.md +++ b/context/glossary.md @@ -89,6 +89,7 @@ - `TursoTransaction<'a, M>`: Transaction-scoped handle in `cli/src/services/db/mod.rs` passed into the closure given to `TursoDb::::transaction()` or `TursoDb::::read_transaction()`. Exposes non-retried synchronous `execute()`/`query_map()` over the same connection/runtime so a domain service can perform multiple parameterized reads/writes plus one explicit commit without the raw `turso::Transaction` API escaping the DB module. `transaction()` issues `BEGIN IMMEDIATE` (reserves the write lock); `read_transaction()` issues a plain `BEGIN` instead, so it never reserves the write lock and does not block concurrent writers on other connections — used by read-only callers such as Agent Trace ETL source extraction (see `context/sce/agent-trace-etl.md`). Both commit only on `Ok` and best-effort `ROLLBACK` on closure error or failed commit via a shared internal `run_transaction()` helper; no nested transactions, savepoints, or transaction-level retry exist yet. - `no-migration DB open path`: `TursoDb::open_without_migrations()` / `TursoDb::open_without_migrations_at(path)` plus Agent Trace adapter-specific no-migration seams; opens/connects a local Turso database with parent-directory creation and configured connection-open retry but does not create `__sce_migrations` or run embedded schema migrations. Active Agent Trace hook callers (`agent_trace_storage::resolve_agent_trace_storage_for_hook_runtime(...)`) use only this path: a missing or migration-incomplete database fails readiness with `sce setup` guidance instead of falling back to migration-running initialization. Setup/lifecycle and `sce trace status` callers (`agent_trace_storage::resolve_agent_trace_storage(...)`) still try this path first and fall back to migration-running initialization when readiness or repository metadata validation fails. - `TursoDb migration readiness check`: Public methods on `TursoDb` in `cli/src/services/db/mod.rs` for non-mutating schema-readiness verification: `migration_metadata_problems(&self) -> Result>` queries `__sce_migrations` metadata and compares applied IDs against `M::migrations()`, returning problems (missing table, incomplete migrations, unexpected migrations) or an empty list when ready; `ensure_schema_ready(&self, setup_guidance: &str) -> Result<()>` calls `migration_metadata_problems()` and bails with a formatted error including `M::db_name()` and the caller-provided guidance string when problems are found. `RepositoryAgentTraceDb::ensure_schema_ready_for_hooks()` delegates to `TursoDb::ensure_schema_ready()` with the Agent Trace–specific `AGENT_TRACE_SCHEMA_SETUP_GUIDANCE` constant. +- `code-change ETL`: The `CodeChangesEtl` bridge from repository `diff_traces` to DWH `code_changes`. It extracts the exact ordered source projection in short plain-read snapshots, uses a source-instance-scoped integer-ID watermark, strictly normalizes supported `patch`/`structured` payloads, computes lowercase SHA-256 over the original payload bytes, derives checked `ParsedPatch` metrics, and atomically validates or inserts source-lineage identities before advancing progress. `session_id` is its only conversation relationship; it never adds or infers `message_id` or message causality. See [code-changes-etl.md](sce/code-changes-etl.md). - `database_retry config namespace`: Nested config namespace under `policies.database_retry` in `sce/config.json`, authored in `config/pkl/base/sce-config-schema.pkl` and parsed/resolved in `cli/src/services/config/mod.rs`. Supports per-database overrides (`local_db`, `agent_trace_db`, `auth_db`) each with optional `connection_open` and `query` objects containing `max_attempts`, `timeout_ms`, `initial_backoff_ms`, `max_backoff_ms`. Validated against JSON Schema at config load and surfaced in `sce config show`/`validate`. Wired into DB adapter constructors and operation methods via config-aware retry resolution with fallback to hardcoded defaults. - `DatabaseRetryConfig`: Rust type in `cli/src/services/config/mod.rs` holding parsed and validated per-database retry policy overrides (`local_db`/`agent_trace_db`/`auth_db`, each `Option`) from the `policies.database_retry` config namespace. Initialized at app startup via `DATABASE_RETRY_CONFIG` `OnceLock` and consumed by config-aware retry resolution in DB adapters. - `PerDbRetryConfig`: Rust type in `cli/src/services/config/mod.rs` holding optional `connection_open` and `query` retry policies (`Option`) for one database in the `database_retry` config namespace. diff --git a/context/overview.md b/context/overview.md index 04ef7566..e6ab2064 100644 --- a/context/overview.md +++ b/context/overview.md @@ -12,6 +12,7 @@ It also includes a Rust CLI (`sce`) for Shared Context Engineering workflows: au - **Observability:** config-resolved logging to stderr, optional dated/session-partitioned `log_dir` / `SCE_LOG_DIR` files with retention (see `context/sce/cli-observability-contract.md`). - **Config precedence:** `flags > env > config file > defaults` (see `context/cli/config-precedence-contract.md`). - **Attribution hooks:** enabled by default, gated by staged-diff AI-overlap preflight; `SCE_ATTRIBUTION_HOOKS_DISABLED` opt-out (see `context/sce/agent-trace-commit-msg-coauthor-policy.md`). +- **Code-change ETL:** CLI-independent, source-lineage-watermarked `diff_traces` ingestion into DWH `code_changes`. It uses strict `patch`/`structured` normalization, exact raw-payload hashing, checked parsed-patch metrics, atomic replay/conflict handling, and `session_id` as the only conversation join; it never infers message attribution (see `context/sce/code-changes-etl.md`). - **Install channels:** repo-flake Nix, Cargo, npm, and source-built Flatpak (`dev.crocoder.sce`); Homebrew deferred (see `context/sce/cli-first-install-channels-contract.md`). The CLI crate currently depends on `anyhow`, `chrono`, `clap`, `clap_complete`, `dirs`, `hmac`, `inquire`, `jsonschema`, `keyring-core`, `murmur3`, `owo-colors`, `rand`, `reqwest`, `serde`, `serde_json`, `sha2`, `tokio`, `tracing`, `turso`, and `uuid`, with target-specific keyring backend dependencies for Linux/FreeBSD, macOS, and Windows. No CLI dev-dependencies are currently declared. diff --git a/context/plans/incremental-code-changes-etl.md b/context/plans/incremental-code-changes-etl.md new file mode 100644 index 00000000..a94c1d9e --- /dev/null +++ b/context/plans/incremental-code-changes-etl.md @@ -0,0 +1,169 @@ +# Plan: incremental-code-changes-etl + +## Change summary + +Implement PR 6's CLI-independent incremental ETL from repository-scoped Agent Trace `diff_traces` rows into the existing DWH `code_changes` table. The runner will use the source row's integer `id` as an independent `(repository_id, source_instance_id, diff_traces)` watermark, snapshot only owned source values in a short read transaction, then strictly normalize patch and structured payloads through the canonical patch services before hashing the exact source payload and deriving code-change metrics. + +This extends the completed source identity, DWH schema, single-owner replica, Agent Trace ETL, and conversation ETL boundaries. It preserves `session_id` as the only conversation/code-change join boundary, validates source-lineage-scoped idempotent replay without overwriting conflicts, and commits code-change facts plus watermark atomically. Pull/push, control-plane integration, CLI orchestration, and message-level attribution remain outside the ETL. + +## 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: `CodeChangesEtl` exposes a CLI-independent API consistent with the existing table runners, accepts an open repository source and lock-owning `AgentTraceDwhReplica`, obtains `source_instance_id` from repository metadata, uses `diff_traces` as its independent source table, reports extracted/inserted/already-present/batch and before/after watermark stats, and never calls `pull()` or `push()` or acquires credentials. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_api`; inspect the runner and replica call paths. +- [x] AC2: Source extraction uses exactly `id, time_ms, session_id, patch, model_id, tool_name, tool_version, payload_type` with `id > watermark ORDER BY id ASC LIMIT batch_size`, treats a missing watermark as zero, never queries `MAX(id)` or uses timestamps for progress, validates positive batch sizes, and processes all rows through bounded ordered batches. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_extraction`. +- [x] AC3: Each source batch is copied into owned `SourceDiffTrace` values in a short plain read transaction, commits/releases the source snapshot before parsing, hashing, metric derivation, or destination work, retries only existing transient Busy/database-locked contention with the shared bounded retry/rollback mechanics, and permits concurrent source writers to continue. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_source`. +- [x] AC4: `patch` payloads and currently supported `structured` payloads both normalize through one canonical parsing/derivation path into `ParsedPatch`; existing best-effort recent-diff processing may continue to classify malformed rows as skipped, while strict DWH transformation returns an error for malformed, unsupported, or future payload types and never silently treats them as unified patches. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml patch_payload_normalization`; inspect shared parser callers for preserved best-effort behavior. +- [x] AC5: DWH transformation preserves source metadata (`session_id`, `time_ms`, nullable `model_id`, `tool_name`, nullable `tool_version`, and `payload_type`), retains the source-level `model_id` without deriving a replacement from parsed hunks, derives `files_changed` from `ParsedPatch.files`, derives `lines_added` by counting normalized touched lines with `TouchedLineKind::Added`, derives `lines_removed` by counting normalized touched lines with `TouchedLineKind::Removed` across every file and hunk, and rejects count conversions that cannot be represented by the destination schema. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_metrics`. +- [x] AC6: `patch_sha256` is lowercase hexadecimal SHA-256 of the exact UTF-8 bytes of the original `diff_traces.patch` value for both payload types; normalization never changes the bytes used for hashing. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_hash`. +- [x] AC7: Destination identity is exactly `(repository_id, source_instance_id, source_diff_trace_id)`. A missing identity inserts one `code_changes` row with all required lineage, source metadata, and derived fields; an identical existing row increments `already_present`; any synchronized or derived mismatch returns an integrity error and leaves the existing row unchanged without using silent conflict-ignore or overwrite behavior. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_identity`. +- [x] AC8: Each batch commits repository/source lineage, code-change inserts or verifications, and the `diff_traces` watermark in one destination transaction; transformation failure occurs before destination transaction creation, and any injected destination failure or integrity conflict rolls back all facts, dimensions, and watermark changes so the full batch can be replayed. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_atomic`. +- [x] AC9: Tests prove normal unified patches, valid structured payloads, multiple files, additions/removals including modified/new/deleted files where practical, initial IDs `1..=3`, no-op reruns, growth with IDs `4` and `5`, batch size `2` over five rows, watermark-behind idempotent replay, conflicts, malformed patch/structured/future payload failures, and no watermark advancement past a failed row. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl`. +- [x] AC10: Session-level joins remain the only supported conversation relationship: every DWH code-change row preserves `session_id`, code changes can be queried alongside messages/message parts for the same session, no `message_id` is added or inferred, and documentation states that captured data proves only session membership, not causality to an individual message. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_session_relationship`; inspect `code_changes` schema/API and the synchronized context documents. +- [x] AC11: Two source instances of one repository can each ingest local `diff_traces.id = 1` with independent watermarks and coexist in DWH; source contention tests show writers continue and eventually committed rows are observed; a failed batch can be rerun successfully without duplicates. + - Validate: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_lineage_contention`. +- [x] AC12: Durable context documents the source-to-DWH code-activity path, exact source-instance identity and watermark scope, strict transformation behavior, metric/hash definitions, transactional replay/conflict rules, pull/push separation, and the explicit session-only relationship to conversations with no message-level attribution. + - Validate: inspect `context/sce/agent-trace-etl.md`, `context/sce/agent-trace-dwh-db.md`, `context/sce/agent-trace-dwh-replica.md`, `context/sce/agent-trace-db.md`, `context/overview.md`, `context/architecture.md`, `context/glossary.md`, and `context/context-map.md` against the implementation and tests. + +### 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 completed code-change ETL and session-level relationship. +- Update `context/sce/agent-trace-etl.md`, `context/sce/agent-trace-dwh-db.md`, `context/sce/agent-trace-dwh-replica.md`, and `context/sce/agent-trace-db.md` from their current “code-change ETL future work” statements to the implemented behavior. +- Add a focused `context/sce/code-changes-etl.md` domain document and register it in `context/context-map.md`. + +## Constraints and non-goals + +- **In scope:** shared parser refactoring needed for one strict DWH normalization path; `SourceDiffTrace` extraction; strict patch/structured transformation; exact payload hashing; normalized metrics; source-lineage-scoped `code_changes` loading; independent watermarking; stats; focused ETL, rollback, replay, source-contention, and session-join tests; and durable documentation. +- **Out of scope:** message-level attribution or `message_id` on any diff/code-change row; post-commit intersection ETL; commits table; file-level destination rows; raw diff-trace archival; control-plane calls; DWH provisioning; credential retrieval; OAuth; `sce` sync wiring; background syncing; analytics UI; and remote-to-source synchronization. +- **Constraints:** reuse `RepositoryAgentTraceDb::verify_or_initialize_repository_metadata`, `AgentTraceDwhReplica`, `services::etl` helpers, `parse_patch`, structured-patch derivation, `ParsedPatch`, `sha2`, and existing Turso transactions; do not change live source capture semantics; do not hold source read transactions during transformation or destination work; do not use timestamps or a separate `MAX(id)` query for progress; do not change `cli/migrations/agent-trace-dwh/001_dwh_schema.sql` unless implementation proves an existing contract column is insufficient; do not create a generic ETL trait hierarchy; and do not add a new dependency when existing crates suffice. +- **Non-goal:** infer causality between a diff trace and a conversation message. `session_id` is the supported join boundary until source capture supplies explicit message attribution metadata. + +## Assumptions + +- The existing DWH `code_changes` schema is sufficient: its source-lineage identity, preserved metadata, metric columns, and `patch_sha256` column require no migration. +- The strict structured-payload adapter will wrap or refactor the existing Claude structured derivation so the best-effort recent-window caller retains its current skip classification, while DWH ETL receives `Result` for supported `patch` and `structured` payloads. +- `model_id` and `tool_version` are preserved as nullable source values exactly as extracted; `session_id` is preserved exactly as stored in the source database, including any producer prefix. +- Metric counts are converted to the destination integer type with checked conversion; an impossible overflow is a strict transformation failure rather than truncated output. +- Test-only destination-failure injection is acceptable for proving atomic rollback and does not become a production error path. + +## Task stack + +- [x] T01: `Extract one canonical strict diff-trace normalization path` (status:complete) + - Task ID: T01 + - Goal: Make unified and structured diff payloads available through one parser boundary that supports strict DWH errors without changing best-effort recent-diff behavior. + - Boundaries (in/out of scope): In — parser/structured-patch helper extraction, strict `patch` and `structured` dispatch, explicit unsupported-payload errors, preservation of existing Claude structured formats and model provenance, and pure malformed/valid normalization tests. Out — source SQL, DWH writes, watermarks, ETL orchestration, message attribution, and destination schema changes. + - Dependencies: none + - Done when: a strict helper returns `Result` for both supported payload types, malformed or unsupported inputs fail explicitly, structured payloads use the existing derivation logic rather than a second parser, and current recent-diff best-effort callers still classify malformed rows as skipped with their existing behavior. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml patch_payload_normalization`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Completion evidence: Added public `normalize_diff_trace_payload` dispatching supported `patch` and `structured` payloads into `ParsedPatch`, with explicit malformed, unsupported, and negative-time errors. Reused the strict helper in recent diff-trace parsing while preserving best-effort skip accounting and row-level model provenance injection. Added focused valid/failure normalization tests. + - Verification: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml patch_payload_normalization` passed (3 tests); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml agent_trace_db` passed (20 tests); `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T02: `Add bounded diff-trace extraction and deterministic transformation` (status:complete) + - Task ID: T02 + - Goal: Copy source `diff_traces` rows safely and transform them after the source snapshot into owned code-change values with exact hashing and normalized metrics. + - Boundaries (in/out of scope): In — `SourceDiffTrace`, exact projection/query, shared source contention retry, `TransformedCodeChange`, strict parser invocation, source metadata preservation, checked metric derivation, and raw-payload SHA-256 tests. Out — destination transactions, watermark mutation, identity replay, replica orchestration, and context documentation. + - Dependencies: T01 + - Done when: extraction uses only the specified ordered integer-ID query in a short read transaction; transformation starts after extraction returns; patch and structured rows retain their discriminator and metadata; metrics count `ParsedPatch` files/touched-line kinds; and exact source-payload hashes match deterministic expected values. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_extraction`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_metrics`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_hash`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Completion evidence: Added `cli/src/services/code_changes_etl.rs` with the exact ordered `diff_traces` projection, short read-transaction extraction using shared bounded contention retry, owned source rows, strict patch/structured normalization, source metadata preservation, checked `ParsedPatch` file/touched-line metrics, and lowercase SHA-256 hashing of the original payload bytes. Registered the module in `cli/src/services/mod.rs`; destination loading and watermark mutation remain absent. + - Verification: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_extraction` passed (2 tests); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_metrics` passed (2 tests); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_hash` passed (1 test); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl` passed (7 tests); `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T03: `Load source-lineage code changes atomically with replay validation` (status:complete) + - Task ID: T03 + - Goal: Insert or verify transformed code-change rows and advance the `diff_traces` watermark in one destination transaction. + - Boundaries (in/out of scope): In — repository/source lineage ensuring, source-scoped destination lookup, full synchronized/derived content comparison, explicit conflict errors, insert/already-present accounting, watermark upsert, and test-only failure injection/rollback tests. Out — source extraction, parser changes, public run loop, pull/push, credentials, message joins, and documentation. + - Dependencies: T02 + - Done when: missing `(repository_id, source_instance_id, source_diff_trace_id)` rows insert without conflict-ignore semantics; identical rows count as already present; every compared field mismatch fails without overwrite; code-change rows, dimensions, and watermark roll back together on any destination error; and replay/conflict/rollback tests pass. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_identity`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_atomic`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Completion evidence: Added atomic code-change destination loading with repository/source lineage initialization, exact source-lineage identity lookup, full synchronized and derived-field comparison, explicit integrity conflicts, insert/already-present accounting, and independent `diff_traces` watermark upsert. Added replay, conflict, rollback, and cross-source-instance tests; destination transactions roll back dimensions, facts, and watermarks on injected failure. + - Verification: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_identity` passed (2 tests); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_atomic` passed (1 test); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl` passed (11 tests); `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T04: `Expose CodeChangesEtl and prove incremental session-level behavior` (status:complete) + - Task ID: T04 + - Goal: Add the replica-owned public runner and end-to-end coverage for batching, failure boundaries, independent source instances, contention, and session-level conversation queries. + - Boundaries (in/out of scope): In — default/configurable batch runner, `CodeChangesEtlStats`, metadata-derived source identity, initial/no-op/growth/batch-boundary runs, invalid-row watermark stopping, source-instance coexistence, source writer contention, rollback/replay, session query coverage with messages/parts, and pull/push absence inspection. Out — CLI/control-plane wiring, orchestration, message-level attribution, commits, and remote synchronization. + - Dependencies: T03 + - Done when: `CodeChangesEtl::default().run(repository_id, source, replica)` processes only rows above the independent watermark; invalid transformations fail before destination writes; watermarks never advance past failed rows; all requested incremental/replay/conflict/lineage/contention/session scenarios pass; and ETL does not call transport or credential code. + - Verification notes (commands or checks): `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_lineage_contention`; `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_session_relationship`; `nix develop -c sh -c 'cd cli && cargo fmt'`. + - Completion evidence: Added configurable/default `CodeChangesEtl` and `CodeChangesEtlStats`, metadata-derived source identity handling, incremental bounded runner loop, default free-function entrypoint, and `AgentTraceDwhReplica::run_code_changes_etl`. Added end-to-end coverage for initial/bounded batches, growth, no-op reruns, invalid-row replay with watermark preservation, independent source-instance watermarks, source writer contention, and session-only joins without `message_id` attribution. The runner has no pull/push or credential access. + - Verification: `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl` passed (17 tests); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_lineage_contention` passed (1 test); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_session_relationship` passed (1 test); `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml patch_payload_normalization` passed (3 tests); `nix develop -c sh -c 'cd cli && cargo fmt'` passed; `git diff --check` passed. + +- [x] T05: `Document the session-level code-change ETL contract` (status:complete) + - Task ID: T05 + - Goal: Make the implemented code-change source/DWH boundary, strictness, metrics, hash, identity, rollback, and conversation relationship durable and discoverable. + - Boundaries (in/out of scope): In — focused code-change ETL context, context-map registration, updates to related ETL/source/DWH/replica and root context, and documentation of the session-only join boundary and lack of message causality. Out — implementation changes, CLI/control-plane docs, new attribution metadata, and unrelated context cleanup. + - Dependencies: T04 + - Done when: durable context matches the code and tests, explicitly states `session_id` is the supported relationship between conversations and code changes, explicitly states there is no reliable message-level attribution, and removes stale “code-change ETL remains future work” claims without implying message causality. + - Verification notes (commands or checks): inspect the listed context files against `cli/src/services/code_changes_etl.rs` (or the final module path), the source schema, DWH schema, parser helpers, and test coverage; `git diff --check`. + - Completion evidence: Updated the focused code-change ETL contract and context map, then reconciled overview, architecture, glossary, source DB, DWH schema, DWH replica, and shared ETL context with the implemented strict normalization, exact hashing, checked metrics, source-lineage identity/watermark, atomic replay/conflict behavior, transport separation, and session-only conversation relationship. + - Verification: Read the mandatory root context files and affected domain documents against the implementation, source/DWH schemas, parser boundary, and tests; `git diff --check` passed; all changed context files remain at or below 250 lines. + +## Open questions + +None. The request fixes the source projection, identity, strictness, metrics, hash, session boundary, orchestration exclusions, and test obligations; the existing DWH schema and ETL infrastructure supply the remaining local seams. + +## Validation Report + +**Status:** validated +**Date:** 2026-08-08 + +### Commands run + +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_api` -> exit 0 (no matching tests; API and call paths inspected) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_extraction` -> exit 0 (2 extraction tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl_source` -> exit 0 (no matching tests; source snapshot and retry paths inspected) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml patch_payload_normalization` -> exit 0 (3 normalization tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_metrics` -> exit 0 (2 metric tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_hash` -> exit 0 (1 hash test passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_identity` -> exit 0 (2 identity tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_atomic` -> exit 0 (1 atomic rollback test passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_etl` -> exit 0 (17 ETL tests passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_session_relationship` -> exit 0 (1 session relationship test passed) +- `nix develop -c ./scripts/run-cli-cargo.sh test --manifest-path cli/Cargo.toml code_changes_lineage_contention` -> exit 0 (1 lineage/contention test passed) +- `nix run .#pkl-check-generated` -> exit 0 (ephemeral Pkl generation passed; 101 files) +- `nix flake check` -> exit 0 (all flake checks passed) +- `git diff --check` -> exit 0 (no whitespace errors) + +### Scaffolding removed + +- None. + +### Success-criteria verification + +- [x] AC1: CLI-independent API, metadata-derived source identity, stats, and no transport/credential access -> targeted command exited 0 with no matching tests; `CodeChangesEtl::run` and `AgentTraceDwhReplica::run_code_changes_etl` call paths were inspected. +- [x] AC2: Exact ordered bounded integer-ID extraction and positive batch validation -> extraction tests passed. +- [x] AC3: Owned short source snapshots, contention retry, and writer concurrency -> source path inspection confirmed plain read transactions and shared retry; lineage contention test passed. +- [x] AC4: Strict canonical patch/structured normalization with preserved best-effort recent-diff behavior -> 3 normalization tests passed and shared parser callers were inspected. +- [x] AC5: Source metadata preservation and checked parsed-patch metrics -> 2 metric tests passed. +- [x] AC6: Lowercase SHA-256 of exact original payload bytes -> hash test passed. +- [x] AC7: Source-lineage identity, replay verification, and conflict rejection -> 2 identity tests passed. +- [x] AC8: Atomic fact, lineage, and watermark transaction with rollback -> atomic rollback test passed. +- [x] AC9: Incremental, replay, batching, malformed-row, conflict, structured, and lineage scenarios -> 17 ETL tests passed. +- [x] AC10: Session-only conversation relationship with no `message_id` attribution -> session relationship test passed; schema/API and synchronized context were inspected. +- [x] AC11: Independent source instances, contention, and replay behavior -> lineage/contention test passed and ETL suite covered independent watermarks/replay. +- [x] AC12: Durable context contract for source path, identity, strictness, metrics/hash, replay/conflict, transport separation, and session-only relationship -> all listed context files were inspected against implementation and tests. + +### 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 ea51ed37..16a83b95 100644 --- a/context/sce/agent-trace-db.md +++ b/context/sce/agent-trace-db.md @@ -13,7 +13,8 @@ This is the live-capture **source** schema written by hooks and `sce trace` — - `PAYLOAD_TYPE_PATCH` / `PAYLOAD_TYPE_STRUCTURED`: string constants (`"patch"` / `"structured"`) for the `diff_traces.payload_type` discriminator column; `OpenCode` normalized diff-trace payloads use `patch`, `Claude` structured `PostToolUse` payloads use `structured`. - `insert_diff_trace()`: domain-specific insert helper using parameterized SQL. - `RecentDiffTracePatches`: parsed recent `diff_traces` query result containing valid parsed patches plus skipped-row reports. -- `recent_diff_trace_patches(cutoff_time_ms, end_time_ms)`: chronological `diff_traces` read helper for rows in the inclusive window `time_ms >= cutoff_time_ms AND time_ms <= end_time_ms`; parses raw patch text through `parse_patch` and skips malformed rows without failing the query. +- `normalize_diff_trace_payload(...)`: strict shared normalization boundary returning `Result` for `patch` and `structured` payloads; malformed, unsupported, future, or invalid-time inputs fail explicitly for ETL callers. +- `recent_diff_trace_patches(cutoff_time_ms, end_time_ms)`: chronological `diff_traces` read helper for rows in the inclusive window `time_ms >= cutoff_time_ms AND time_ms <= end_time_ms`; uses the shared strict normalizer and maps malformed or unsupported rows to best-effort skipped results without failing the query. - `PostCommitPatchIntersectionInsert<'a>`: insert payload for post-commit intersection results with commit metadata, window bounds, loaded/skipped counts, and serialized patch JSON. - `insert_post_commit_patch_intersection()`: domain-specific insert helper using parameterized SQL. - `AgentTraceInsert<'a>`: insert payload for built Agent Trace rows with `commit_id`, `commit_time_ms`, serialized `trace_json`, `agent_trace_id`, non-null `url`, and required `remote_url: &'a str` (Rust-API-only; DB column stays nullable). @@ -42,7 +43,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`. 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. +`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`, the source-lineage-scoped message-part pipeline is implemented in `cli/src/services/conversation_parts_etl.rs`, and code-change extraction/transformation, atomic destination loading, and incremental `CodeChangesEtl` orchestration are implemented in `cli/src/services/code_changes_etl.rs`; `AgentTraceDwhReplica::run_code_changes_etl()` exposes the lock-owned boundary. Code-change ingestion treats the source `diff_traces.id` as the local source row identity and preserves `session_id` without adding message-level attribution. `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. @@ -218,7 +219,7 @@ The messages/parts ETL uses integer-ID watermarks and deliberately does not prov - SQL reads `id`, `time_ms`, `session_id`, `patch`, nullable `model_id` + `tool_name` + `tool_version`, and `payload_type` from `diff_traces` where `time_ms >= cutoff_time_ms AND time_ms <= end_time_ms`. - Rows are ordered by `time_ms ASC, id ASC` for deterministic chronological processing. -- Valid row patches are parsed through `cli/src/services/patch.rs` `parse_patch` for `payload_type="patch"` rows (OpenCode unified-diff payloads), while `payload_type="structured"` rows (Claude `PostToolUse` structured payloads) are parsed from stored JSON through `cli/src/services/structured_patch.rs` `derive_claude_structured_patch` at read time to produce `ParsedPatch` without pre-rendered unified-diff text. +- Valid row patches use the shared `cli/src/services/agent_trace_db/mod.rs` `normalize_diff_trace_payload` boundary: `payload_type="patch"` rows (OpenCode unified-diff payloads) delegate to `cli/src/services/patch.rs` `parse_patch`, while `payload_type="structured"` rows (Claude `PostToolUse` structured payloads) delegate through the existing `cli/src/services/structured_patch.rs` `derive_claude_structured_patch` logic to produce `ParsedPatch` without pre-rendered unified-diff text. Strict ETL callers receive errors for malformed, unsupported, future, or invalid-time payloads. - Each produced `PatchHunk` is annotated with the originating row `model_id` (`Some(value)` propagated verbatim, `NULL` propagated as `None`) for both patch and structured paths; parsed row records also carry nullable `tool_name`/`tool_version` and `payload_type` from the same source row and are returned as `ParsedDiffTracePatch` records. - Malformed recent row patches (invalid unified-diff text, invalid structured JSON, unsupported payload types, or unsupported Claude structured payloads) are returned as `SkippedDiffTracePatch` records with deterministic parse-error or derivation-skip reasons; malformed historical rows do not fail the operation. - `RecentDiffTracePatches::loaded_count()` and `skipped_count()` expose accounting for later hook output and persistence metadata. diff --git a/context/sce/agent-trace-dwh-db.md b/context/sce/agent-trace-dwh-db.md index 2d11bd02..7d9382af 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 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. +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`, source-lineage-scoped `message_parts`, and source-lineage-scoped `code_changes` through independent table runners; transport synchronization remains a separate caller concern. ## Adapter @@ -30,8 +30,18 @@ Two different uniqueness scopes are used, chosen by whether the source identity `message_parts` also carries `idx_dwh_message_parts_order` on `(repository_id, session_id, message_id, generated_at_unix_ms, source_part_id)`, so deterministic message-part reconstruction orders by source timestamp and falls back to `source_part_id` when multiple parts share the same `generated_at_unix_ms`. +## Code-change loading + +`load_code_change_batch()` in `cli/src/services/code_changes_etl.rs` transforms all source rows before opening the destination transaction. The loader ensures repository/source dimensions, looks up each `(repository_id, source_instance_id, source_diff_trace_id)`, compares session/time/attribution/payload/metrics/hash content, inserts missing rows without conflict-ignore or overwrite behavior, and counts identical rows as already present. A mismatch is an integrity error. The same transaction upserts the `diff_traces` watermark only after every row succeeds, so conflicts or other destination failures roll back dimensions, facts, and progress together. + +`CodeChangesEtl` runs the source `diff_traces` bridge in bounded ordered batches. It extracts the exact eight-column source projection in a short plain read transaction, strictly normalizes supported `patch` and `structured` payloads through the canonical parser before destination work, preserves source metadata, counts parsed files and touched lines with checked integer conversion, and hashes exact source payload bytes. A null source `tool_name` is rejected because the destination contract requires it. Its replica-owned runner is `AgentTraceDwhReplica::run_code_changes_etl()`; neither it nor the destination adapter performs pull/push or credential handling. + +## Session-only conversation relationship + +`code_changes.session_id` is the only supported relationship from code changes to conversation facts. Queries may join code changes with `messages` and `message_parts` for the same `(repository_id, session_id)`, but `code_changes` has no `message_id` and the ETL never infers one. This proves session membership only, not causality to an individual message. + ## 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, 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. +`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, the `PartsEtl` slice computes and populates the lowercase SHA-256 of exact source part text bytes, and code-change loading computes the lowercase SHA-256 of exact source `diff_traces.patch` bytes for both supported payload types. Message rows require no content hash. 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 7eac06f5..934b72cd 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` and `ConversationEtl` bridges 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`, `ConversationEtl`, and `CodeChangesEtl` bridges for local fact/watermark loading; ETL never performs pull/push, credential discovery/persistence, or background sync. ## Ownership and lock-before-open @@ -45,7 +45,7 @@ Observed real-SDK behavior (recorded across repeated runs, including with tempor ## ETL separation -`AgentTraceDwhReplica::run_agent_trace_etl()` accepts an already-open repository source and an `AgentTraceEtl` configuration, then delegates while retaining the replica's bridge-lock ownership. The ETL verifies source metadata, extracts short bounded read snapshots, and commits facts plus per-lineage watermarks locally. Pull/push remain explicit operations owned by the caller and are never invoked by ETL. +`AgentTraceDwhReplica::run_agent_trace_etl()` accepts an already-open repository source and an `AgentTraceEtl` configuration, then delegates while retaining the replica's bridge-lock ownership. `run_code_changes_etl()` provides the same boundary for `CodeChangesEtl` and the source `diff_traces` to `code_changes` bridge. These ETLs verify source metadata, extract short bounded read snapshots, and commit facts plus per-lineage watermarks locally. Pull/push remain explicit operations owned by the caller and are never invoked by ETL. Code-change transformation preserves `session_id` for conversation queries; it has no `message_id` field and does not infer attribution or causality to an individual message. See [code-changes-etl.md](code-changes-etl.md). Control-plane/provisioning calls, OAuth or credential discovery/persistence, token rotation, CLI/lifecycle/setup/doctor/hook wiring, automatic/background sync, archive/retention behavior, and partial sync remain out of scope for this boundary. diff --git a/context/sce/agent-trace-etl.md b/context/sce/agent-trace-etl.md index a544a537..8f75db22 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 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. +`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, while `code_changes_etl` applies them to source `diff_traces`; each table has its own watermark. None of these pipelines acquires credentials, invokes `pull()`/`push()`, or depends on CLI orchestration. ## Incremental run contract @@ -16,7 +16,9 @@ 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; `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. +`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. + +The code-change ETL includes destination-independent `diff_traces` extraction/transformation plus atomic destination loading in `cli/src/services/code_changes_etl.rs`; its complete contract is documented in [code-changes-etl.md](code-changes-etl.md). It copies exactly `id`, `time_ms`, `session_id`, `patch`, `model_id`, `tool_name`, `tool_version`, and `payload_type` in a short read snapshot, then strictly normalizes `patch` and `structured` payloads through the canonical parser, preserves source metadata, counts `ParsedPatch` files and touched-line kinds with checked destination-sized metrics, and hashes the original payload bytes with lowercase SHA-256. Loading ensures repository/source lineage, uses `(repository_id, source_instance_id, source_diff_trace_id)` identity, compares all synchronized and derived fields before counting a replay, rejects conflicts without overwrite, and commits facts, dimensions, and the `diff_traces` watermark atomically. Transformation failures and null destination-required `tool_name` values stop before destination writes. `CodeChangesEtl` and `AgentTraceDwhReplica::run_code_changes_etl()` provide the public replica-owned run boundary without pull/push or credential access. ## Source contention retry diff --git a/context/sce/code-changes-etl.md b/context/sce/code-changes-etl.md new file mode 100644 index 00000000..74a9f800 --- /dev/null +++ b/context/sce/code-changes-etl.md @@ -0,0 +1,32 @@ +# Code-Change ETL + +`CodeChangesEtl` is the CLI-independent incremental bridge from repository-scoped Agent Trace `diff_traces` rows to the DWH `code_changes` fact table. It accepts an open `RepositoryAgentTraceDb` and an `AgentTraceDwhReplica`; source metadata supplies the stable `source_instance_id`. The runner owns neither credentials nor remote transport. + +```mermaid +flowchart LR + S[repository agent-trace.db\ndiff_traces] -->|id watermark, short read snapshot| E[CodeChangesEtl] + E -->|strict normalization, metrics, exact hash| L[atomic DWH transaction] + L --> F[code_changes fact] + L --> W[diff_traces watermark] + C[messages and message_parts] -. session_id join only .-> F +``` + +## Incremental contract + +`CodeChangesEtl::default()` uses batches of 500. `with_batch_size()` rejects zero and provides the bounded configuration seam. Each run reads the `(repository_id, source_instance_id, diff_traces)` watermark, treating an absent watermark as zero, and extracts only `id > watermark ORDER BY id ASC LIMIT batch_size`. It repeats until an empty batch and reports extracted, inserted, already-present, batch, and before/after watermark statistics. + +Extraction selects exactly `id`, `time_ms`, `session_id`, `patch`, `model_id`, `tool_name`, `tool_version`, and `payload_type` into owned `SourceDiffTrace` values. The plain read transaction ends before parsing, hashing, metrics, or destination work. Only transient Busy/database-locked source contention is retried with the shared bounded rollback/backoff mechanics, so source writers can continue. + +## Transformation and loading + +`patch` and `structured` payloads use the shared strict normalization boundary and produce `ParsedPatch`; malformed, unsupported, future, and invalid-time payloads fail before destination work. Extraction preserves the exact source projection (`id`, `time_ms`, `session_id`, `patch`, `model_id`, `tool_name`, `tool_version`, and `payload_type`), including nullable source metadata where applicable. The DWH destination requires `tool_name` and rejects a null value before opening its transaction. `files_changed` counts parsed files, while `lines_added` and `lines_removed` count every touched line by `TouchedLineKind` with checked destination-sized conversion. `patch_sha256` is lowercase hexadecimal SHA-256 over the exact original UTF-8 `patch` bytes, regardless of payload type. + +A destination batch ensures repository/source lineage and uses `(repository_id, source_instance_id, source_diff_trace_id)` as its identity. Missing rows are inserted; identical synchronized and derived values count as `already_present`; any mismatch is an integrity conflict and never overwrites the existing row. Transformation completes before the destination transaction begins. Facts, lineage dimensions, and the watermark commit together, so a failure leaves progress behind for complete replay without duplicates. + +`AgentTraceDwhReplica::run_code_changes_etl()` delegates to `CodeChangesEtl` while retaining replica lock ownership. Pull and push remain explicit caller operations; the runner does not acquire credentials or call either transport method. + +## Conversation relationship + +A code-change row preserves `session_id` and can be queried with DWH `messages` and `message_parts` for that same repository/session. `code_changes` has no `message_id`, and the ETL does not infer one. The captured relationship proves session membership only; it does not establish causality between a code change and an individual conversation message. + +See also: [agent-trace-etl.md](agent-trace-etl.md), [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), [conversation-etl.md](conversation-etl.md), [../architecture.md](../architecture.md), [../glossary.md](../glossary.md), [../decisions/2026-08-08-code-changes-etl-watermark-and-replay-contract.md](../decisions/2026-08-08-code-changes-etl-watermark-and-replay-contract.md), and [../context-map.md](../context-map.md).