diff --git a/cli/src/services/agent_trace_db/mod.rs b/cli/src/services/agent_trace_db/mod.rs index c972c0ac..15b8426f 100644 --- a/cli/src/services/agent_trace_db/mod.rs +++ b/cli/src/services/agent_trace_db/mod.rs @@ -147,7 +147,8 @@ pub struct AgentTraceInsert<'a> { } /// Message role constraint for the `messages` table. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(rename_all = "lowercase")] pub enum MessageRole { User, Assistant, diff --git a/cli/src/services/agent_trace_export/mod.rs b/cli/src/services/agent_trace_export/mod.rs new file mode 100644 index 00000000..7995c70a --- /dev/null +++ b/cli/src/services/agent_trace_export/mod.rs @@ -0,0 +1,1535 @@ +//! Read-only incremental export readers for the Agent Trace capture streams. +//! +//! This module establishes the local read/export boundary: cursor in, owned +//! wire-compatible rows out. It performs no database mutation, holds no local +//! sync cursor, and makes no network calls. + +use anyhow::{bail, Context, Result}; +use serde::Serialize; + +use crate::services::agent_trace_db::{repository::RepositoryAgentTraceDb, MessageRole}; + +/// Maximum number of rows a single export reader call may return. +pub const AGENT_TRACE_EXPORT_BATCH_SIZE: usize = 500; + +/// Largest integer value that round-trips exactly through an IEEE-754 double +/// (`Number.MAX_SAFE_INTEGER`). +pub const JS_MAX_SAFE_INTEGER: i64 = 9_007_199_254_740_991; + +/// Rejects a negative cursor. +pub fn validate_cursor(cursor: i64) -> Result<()> { + if cursor < 0 { + bail!("agent trace export cursor must be >= 0, got {cursor}"); + } + + Ok(()) +} + +/// Rejects a zero limit or a limit above [`AGENT_TRACE_EXPORT_BATCH_SIZE`]. +pub fn validate_limit(limit: usize) -> Result<()> { + if limit == 0 { + bail!("agent trace export limit must be greater than 0"); + } + + if limit > AGENT_TRACE_EXPORT_BATCH_SIZE { + bail!( + "agent trace export limit {limit} exceeds maximum batch size {AGENT_TRACE_EXPORT_BATCH_SIZE}" + ); + } + + Ok(()) +} + +/// Rejects a value outside `0..=JS_MAX_SAFE_INTEGER`, the range an exportable +/// numeric field must stay within to survive JSON round-trip without +/// truncation or casting. +pub fn validate_js_safe_integer(value: i64) -> Result<()> { + if !(0..=JS_MAX_SAFE_INTEGER).contains(&value) { + bail!("agent trace export value {value} is outside the JS-safe-integer range 0..={JS_MAX_SAFE_INTEGER}"); + } + + Ok(()) +} + +/// Owned, wire-compatible export row for the `messages` capture stream. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTraceMessageExportRow { + pub source_row_id: i64, + pub session_id: String, + pub message_id: String, + pub role: MessageRole, + pub generated_at_unix_ms: i64, +} + +/// Owned, wire-compatible export row for the `parts` capture stream. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTracePartExportRow { + pub source_row_id: i64, + pub session_id: String, + pub message_id: String, + #[serde(rename = "type")] + pub part_type: String, + pub text: String, + pub generated_at_unix_ms: i64, +} + +/// Owned, wire-compatible export row for the `diff_traces` capture stream. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTraceDiffTraceExportRow { + pub source_row_id: i64, + pub session_id: String, + pub time_ms: i64, + pub patch: String, + pub model_id: Option, + pub tool_name: Option, + pub tool_version: Option, + pub payload_type: String, +} + +/// Owned, wire-compatible export row for the `agent_traces` capture stream. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTraceAgentTraceExportRow { + pub source_row_id: i64, + pub agent_trace_id: String, + pub commit_id: String, + pub commit_time_ms: i64, + pub trace_json: String, + pub url: String, + pub remote_url: Option, +} + +const SELECT_MESSAGES_AFTER_SQL: &str = + "SELECT id, session_id, message_id, role, generated_at_unix_ms +FROM messages +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +const SELECT_PARTS_AFTER_SQL: &str = + "SELECT id, session_id, message_id, type, text, generated_at_unix_ms +FROM parts +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +const SELECT_DIFF_TRACES_AFTER_SQL: &str = + "SELECT id, session_id, time_ms, patch, model_id, tool_name, tool_version, payload_type +FROM diff_traces +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +const SELECT_AGENT_TRACES_AFTER_SQL: &str = + "SELECT id, agent_trace_id, commit_id, commit_time_ms, trace_json, url, remote_url +FROM agent_traces +WHERE id > ?1 +ORDER BY id ASC +LIMIT ?2"; + +/// Read-only incremental export reader over one repository-scoped Agent Trace +/// database. Holds no local cursor, performs no mutation, and makes no +/// network calls; the caller supplies the last server-accepted `id` as +/// `cursor` on every call. +pub struct AgentTraceExportReader<'a> { + db: &'a RepositoryAgentTraceDb, +} + +impl<'a> AgentTraceExportReader<'a> { + pub fn new(db: &'a RepositoryAgentTraceDb) -> Self { + Self { db } + } + + /// Read `messages` rows with `id > cursor`, ordered by `id ASC`, capped + /// at `limit`. + pub fn read_messages_after( + &self, + cursor: i64, + limit: usize, + ) -> Result> { + validate_cursor(cursor)?; + validate_limit(limit)?; + + let rows = self.db.query_map( + SELECT_MESSAGES_AFTER_SQL, + (cursor, limit_as_i64(limit)), + message_export_row_from_turso, + )?; + + for row in &rows { + validate_js_safe_integer(row.source_row_id)?; + validate_js_safe_integer(row.generated_at_unix_ms)?; + } + + Ok(rows) + } + + /// Read `parts` rows with `id > cursor`, ordered by `id ASC`, capped at + /// `limit`. + pub fn read_parts_after( + &self, + cursor: i64, + limit: usize, + ) -> Result> { + validate_cursor(cursor)?; + validate_limit(limit)?; + + let rows = self.db.query_map( + SELECT_PARTS_AFTER_SQL, + (cursor, limit_as_i64(limit)), + part_export_row_from_turso, + )?; + + for row in &rows { + validate_js_safe_integer(row.source_row_id)?; + validate_js_safe_integer(row.generated_at_unix_ms)?; + } + + Ok(rows) + } + + /// Read `diff_traces` rows with `id > cursor`, ordered by `id ASC`, + /// capped at `limit`. `patch` and `payload_type` are returned raw and + /// unmodified: no patch parsing or normalization is performed. + pub fn read_diff_traces_after( + &self, + cursor: i64, + limit: usize, + ) -> Result> { + validate_cursor(cursor)?; + validate_limit(limit)?; + + let rows = self.db.query_map( + SELECT_DIFF_TRACES_AFTER_SQL, + (cursor, limit_as_i64(limit)), + diff_trace_export_row_from_turso, + )?; + + for row in &rows { + validate_js_safe_integer(row.source_row_id)?; + validate_js_safe_integer(row.time_ms)?; + } + + Ok(rows) + } + + /// Read `agent_traces` rows with `id > cursor`, ordered by `id ASC`, + /// capped at `limit`. `trace_json` is returned as the exact raw string + /// from `SQLite`: no parse/reserialize is performed. + pub fn read_agent_traces_after( + &self, + cursor: i64, + limit: usize, + ) -> Result> { + validate_cursor(cursor)?; + validate_limit(limit)?; + + let rows = self.db.query_map( + SELECT_AGENT_TRACES_AFTER_SQL, + (cursor, limit_as_i64(limit)), + agent_trace_export_row_from_turso, + )?; + + for row in &rows { + validate_js_safe_integer(row.source_row_id)?; + validate_js_safe_integer(row.commit_time_ms)?; + } + + Ok(rows) + } +} + +fn message_export_row_from_turso(row: &turso::Row) -> Result { + Ok(AgentTraceMessageExportRow { + source_row_id: row.get(0).context("failed to read messages.id")?, + session_id: row.get(1).context("failed to read messages.session_id")?, + message_id: row.get(2).context("failed to read messages.message_id")?, + role: message_role_from_column( + row.get::(3) + .context("failed to read messages.role")? + .as_str(), + )?, + generated_at_unix_ms: row + .get(4) + .context("failed to read messages.generated_at_unix_ms")?, + }) +} + +fn part_export_row_from_turso(row: &turso::Row) -> Result { + Ok(AgentTracePartExportRow { + source_row_id: row.get(0).context("failed to read parts.id")?, + session_id: row.get(1).context("failed to read parts.session_id")?, + message_id: row.get(2).context("failed to read parts.message_id")?, + part_type: row.get(3).context("failed to read parts.type")?, + text: row.get(4).context("failed to read parts.text")?, + generated_at_unix_ms: row + .get(5) + .context("failed to read parts.generated_at_unix_ms")?, + }) +} + +fn diff_trace_export_row_from_turso(row: &turso::Row) -> Result { + Ok(AgentTraceDiffTraceExportRow { + source_row_id: row.get(0).context("failed to read diff_traces.id")?, + session_id: row + .get(1) + .context("failed to read diff_traces.session_id")?, + time_ms: row.get(2).context("failed to read diff_traces.time_ms")?, + 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")?, + }) +} + +fn agent_trace_export_row_from_turso(row: &turso::Row) -> Result { + Ok(AgentTraceAgentTraceExportRow { + source_row_id: row.get(0).context("failed to read agent_traces.id")?, + agent_trace_id: row + .get(1) + .context("failed to read agent_traces.agent_trace_id")?, + commit_id: row + .get(2) + .context("failed to read agent_traces.commit_id")?, + commit_time_ms: row + .get(3) + .context("failed to read agent_traces.commit_time_ms")?, + trace_json: row + .get(4) + .context("failed to read agent_traces.trace_json")?, + url: row.get(5).context("failed to read agent_traces.url")?, + remote_url: row + .get(6) + .context("failed to read agent_traces.remote_url")?, + }) +} + +fn message_role_from_column(value: &str) -> Result { + match value { + "user" => Ok(MessageRole::User), + "assistant" => Ok(MessageRole::Assistant), + other => bail!("agent trace export encountered unknown messages.role value: {other}"), + } +} + +/// Converts a validated `limit` (already bounded by [`validate_limit`] to +/// `1..=AGENT_TRACE_EXPORT_BATCH_SIZE`) into the `i64` the SQL `LIMIT` +/// parameter requires. +fn limit_as_i64(limit: usize) -> i64 { + i64::try_from(limit).expect("validated limit should fit in i64") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn message_export_row_serializes_to_camel_case_contract() { + let row = AgentTraceMessageExportRow { + source_row_id: 1, + session_id: "sess-1".to_string(), + message_id: "msg-1".to_string(), + role: MessageRole::Assistant, + generated_at_unix_ms: 1_700_000_000_000, + }; + + assert_eq!( + serde_json::to_value(&row).expect("row should serialize"), + serde_json::json!({ + "sourceRowId": 1, + "sessionId": "sess-1", + "messageId": "msg-1", + "role": "assistant", + "generatedAtUnixMs": 1_700_000_000_000_i64, + }) + ); + } + + #[test] + fn message_export_row_serializes_user_role_lowercase() { + let row = AgentTraceMessageExportRow { + source_row_id: 2, + session_id: "sess-2".to_string(), + message_id: "msg-2".to_string(), + role: MessageRole::User, + generated_at_unix_ms: 1_700_000_000_001, + }; + + let value = serde_json::to_value(&row).expect("row should serialize"); + assert_eq!(value["role"], serde_json::json!("user")); + } + + #[test] + fn part_export_row_serializes_to_camel_case_contract() { + let row = AgentTracePartExportRow { + source_row_id: 5, + session_id: "sess-1".to_string(), + message_id: "msg-1".to_string(), + part_type: "text".to_string(), + text: "hello".to_string(), + generated_at_unix_ms: 1_700_000_000_002, + }; + + assert_eq!( + serde_json::to_value(&row).expect("row should serialize"), + serde_json::json!({ + "sourceRowId": 5, + "sessionId": "sess-1", + "messageId": "msg-1", + "type": "text", + "text": "hello", + "generatedAtUnixMs": 1_700_000_000_002_i64, + }) + ); + } + + #[test] + fn diff_trace_export_row_serializes_with_all_fields_populated() { + let row = AgentTraceDiffTraceExportRow { + source_row_id: 7, + session_id: "sess-3".to_string(), + time_ms: 1_700_000_000_003, + patch: "Index: a\n".to_string(), + model_id: Some("test-provider/test-model".to_string()), + tool_name: Some("opencode".to_string()), + tool_version: Some("1.2.3".to_string()), + payload_type: "patch".to_string(), + }; + + assert_eq!( + serde_json::to_value(&row).expect("row should serialize"), + serde_json::json!({ + "sourceRowId": 7, + "sessionId": "sess-3", + "timeMs": 1_700_000_000_003_i64, + "patch": "Index: a\n", + "modelId": "test-provider/test-model", + "toolName": "opencode", + "toolVersion": "1.2.3", + "payloadType": "patch", + }) + ); + } + + #[test] + fn diff_trace_export_row_serializes_nullable_fields_as_null() { + let row = AgentTraceDiffTraceExportRow { + source_row_id: 8, + session_id: "sess-4".to_string(), + time_ms: 1_700_000_000_004, + patch: "Index: b\n".to_string(), + model_id: None, + tool_name: None, + tool_version: None, + payload_type: "structured".to_string(), + }; + + assert_eq!( + serde_json::to_value(&row).expect("row should serialize"), + serde_json::json!({ + "sourceRowId": 8, + "sessionId": "sess-4", + "timeMs": 1_700_000_000_004_i64, + "patch": "Index: b\n", + "modelId": null, + "toolName": null, + "toolVersion": null, + "payloadType": "structured", + }) + ); + } + + #[test] + fn agent_trace_export_row_serializes_with_remote_url_null() { + let row = AgentTraceAgentTraceExportRow { + source_row_id: 9, + agent_trace_id: "trace-1".to_string(), + commit_id: "abc123".to_string(), + commit_time_ms: 1_700_000_000_005, + trace_json: "{\"steps\":[]}".to_string(), + url: "https://example.com/trace/1".to_string(), + remote_url: None, + }; + + assert_eq!( + serde_json::to_value(&row).expect("row should serialize"), + serde_json::json!({ + "sourceRowId": 9, + "agentTraceId": "trace-1", + "commitId": "abc123", + "commitTimeMs": 1_700_000_000_005_i64, + "traceJson": "{\"steps\":[]}", + "url": "https://example.com/trace/1", + "remoteUrl": null, + }) + ); + } + + #[test] + fn agent_trace_export_row_serializes_with_remote_url_populated() { + let row = AgentTraceAgentTraceExportRow { + source_row_id: 10, + agent_trace_id: "trace-2".to_string(), + commit_id: "def456".to_string(), + commit_time_ms: 1_700_000_000_006, + trace_json: "{\"steps\":[1]}".to_string(), + url: "https://example.com/trace/2".to_string(), + remote_url: Some("https://github.com/org/repo/commit/def456".to_string()), + }; + + let value = serde_json::to_value(&row).expect("row should serialize"); + assert_eq!( + value["remoteUrl"], + serde_json::json!("https://github.com/org/repo/commit/def456") + ); + } + + #[test] + fn validate_cursor_rejects_negative() { + let error = validate_cursor(-1).expect_err("negative cursor should error"); + assert!(error.to_string().contains("cursor")); + } + + #[test] + fn validate_cursor_accepts_zero_and_positive() { + assert!(validate_cursor(0).is_ok()); + assert!(validate_cursor(42).is_ok()); + } + + #[test] + fn validate_limit_rejects_zero() { + let error = validate_limit(0).expect_err("zero limit should error"); + assert!(error.to_string().contains("limit")); + } + + #[test] + fn validate_limit_rejects_above_batch_size() { + let error = validate_limit(AGENT_TRACE_EXPORT_BATCH_SIZE + 1) + .expect_err("limit above batch size should error"); + assert!(error.to_string().contains("limit")); + } + + #[test] + fn validate_limit_accepts_batch_size() { + assert!(validate_limit(AGENT_TRACE_EXPORT_BATCH_SIZE).is_ok()); + } + + #[test] + fn validate_limit_accepts_one() { + assert!(validate_limit(1).is_ok()); + } + + #[test] + fn validate_js_safe_integer_accepts_zero() { + assert!(validate_js_safe_integer(0).is_ok()); + } + + #[test] + fn validate_js_safe_integer_accepts_max_safe_integer() { + assert!(validate_js_safe_integer(JS_MAX_SAFE_INTEGER).is_ok()); + } + + #[test] + fn validate_js_safe_integer_rejects_above_max_safe_integer() { + let error = validate_js_safe_integer(JS_MAX_SAFE_INTEGER + 1) + .expect_err("value above max safe integer should error"); + assert!(error.to_string().contains("JS-safe-integer")); + } + + #[test] + fn validate_js_safe_integer_rejects_negative() { + let error = validate_js_safe_integer(-1).expect_err("negative value should error"); + assert!(error.to_string().contains("JS-safe-integer")); + } + + use std::{ + fs, + path::PathBuf, + time::{SystemTime, UNIX_EPOCH}, + }; + + use crate::services::agent_trace_db::{ + AgentTraceInsert, DiffTraceInsert, InsertMessageInsert, InsertPartInsert, PartType, + PAYLOAD_TYPE_PATCH, PAYLOAD_TYPE_STRUCTURED, + }; + + fn unique_test_db_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-agent-trace-export-{label}-{}-{nonce}", + std::process::id() + )) + .join("agent-trace.db") + } + + fn remove_test_db(db_path: &std::path::Path) { + if let Some(parent) = db_path.parent() { + fs::remove_dir_all(parent).expect("test DB directory should be removed"); + } + } + + fn row_count(db: &RepositoryAgentTraceDb, table: &str) -> i64 { + db.query_map(&format!("SELECT COUNT(*) FROM {table}"), (), |row| { + row.get::(0).map_err(Into::into) + }) + .expect("count query should succeed") + .into_iter() + .next() + .expect("count row should exist") + } + + fn insert_message_row_with_id(db: &RepositoryAgentTraceDb, id: i64, message_id: &str) { + db.execute( + "INSERT INTO messages (id, session_id, message_id, role, generated_at_unix_ms) VALUES (?1, ?2, ?3, ?4, ?5)", + (id, "sess-1", message_id, "assistant", 1_000 + id), + ) + .expect("direct message insert should succeed"); + } + + fn insert_part_row_with_id(db: &RepositoryAgentTraceDb, id: i64, message_id: &str) { + db.execute( + "INSERT INTO parts (id, type, text, message_id, session_id, generated_at_unix_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + (id, "text", "hello", message_id, "sess-1", 1_000 + id), + ) + .expect("direct part insert should succeed"); + } + + fn insert_diff_trace_row_with_id(db: &RepositoryAgentTraceDb, id: i64, session_id: &str) { + db.execute( + "INSERT INTO diff_traces (id, time_ms, session_id, patch, model_id, tool_name, tool_version, payload_type) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + ( + id, + 1_000 + id, + session_id, + "Index: a\n", + Option::<&str>::None, + Option::<&str>::None, + Option::<&str>::None, + PAYLOAD_TYPE_PATCH, + ), + ) + .expect("direct diff_trace insert should succeed"); + } + + fn insert_agent_trace_row_with_id(db: &RepositoryAgentTraceDb, id: i64, agent_trace_id: &str) { + db.execute( + "INSERT INTO agent_traces (id, commit_id, commit_time_ms, trace_json, agent_trace_id, url, remote_url) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + ( + id, + "abc123", + 1_000 + id, + "{\"steps\":[]}", + agent_trace_id, + "https://example.com/trace", + Option::<&str>::None, + ), + ) + .expect("direct agent_trace insert should succeed"); + } + + #[test] + fn read_messages_after_returns_rows_after_cursor_in_order() { + let db_path = unique_test_db_path("messages-basic"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.insert_messages(vec![ + InsertMessageInsert { + session_id: "sess-1".to_string(), + message_id: "msg-1".to_string(), + role: MessageRole::User, + generated_at_unix_ms: 1_001, + }, + InsertMessageInsert { + session_id: "sess-1".to_string(), + message_id: "msg-2".to_string(), + role: MessageRole::Assistant, + generated_at_unix_ms: 1_002, + }, + InsertMessageInsert { + session_id: "sess-1".to_string(), + message_id: "msg-3".to_string(), + role: MessageRole::Assistant, + generated_at_unix_ms: 1_003, + }, + ]) + .expect("seed messages should insert"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_messages_after(1, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![2, 3] + ); + assert_eq!(rows[0].message_id, "msg-2"); + assert_eq!(rows[0].role, MessageRole::Assistant); + assert_eq!(rows[0].generated_at_unix_ms, 1_002); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_returns_non_contiguous_ids() { + let db_path = unique_test_db_path("messages-gap"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for (id, message_id) in [ + (11, "msg-11"), + (15, "msg-15"), + (19, "msg-19"), + (30, "msg-30"), + ] { + insert_message_row_with_id(&db, id, message_id); + } + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_messages_after(10, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![11, 15, 19, 30] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_limit_truncates_and_follow_up_continues() { + let db_path = unique_test_db_path("messages-limit"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for id in 1..=10 { + insert_message_row_with_id(&db, id, &format!("msg-{id}")); + } + + let reader = AgentTraceExportReader::new(&db); + let first_batch = reader + .read_messages_after(0, 3) + .expect("first limited read should succeed"); + assert_eq!( + first_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![1, 2, 3] + ); + + let next_cursor = first_batch + .last() + .expect("first batch non-empty") + .source_row_id; + let second_batch = reader + .read_messages_after(next_cursor, 3) + .expect("follow-up read should succeed"); + assert_eq!( + second_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![4, 5, 6] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_returns_empty_at_or_beyond_max_id() { + let db_path = unique_test_db_path("messages-empty-tail"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_message_row_with_id(&db, 1, "msg-1"); + + let reader = AgentTraceExportReader::new(&db); + assert!(reader + .read_messages_after(1, 500) + .expect("read at max id should succeed") + .is_empty()); + assert!(reader + .read_messages_after(100, 500) + .expect("read beyond max id should succeed") + .is_empty()); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_rejects_invalid_cursor_and_limit() { + let db_path = unique_test_db_path("messages-invalid"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + let reader = AgentTraceExportReader::new(&db); + + let cursor_error = reader + .read_messages_after(-1, 500) + .expect_err("negative cursor should error"); + assert!(cursor_error.to_string().contains("cursor")); + + let zero_limit_error = reader + .read_messages_after(0, 0) + .expect_err("zero limit should error"); + assert!(zero_limit_error.to_string().contains("limit")); + + let excess_limit_error = reader + .read_messages_after(0, AGENT_TRACE_EXPORT_BATCH_SIZE + 1) + .expect_err("excess limit should error"); + assert!(excess_limit_error.to_string().contains("limit")); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_rejects_row_above_safe_integer_bound() { + let db_path = unique_test_db_path("messages-unsafe-integer"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.execute( + "INSERT INTO messages (id, session_id, message_id, role, generated_at_unix_ms) VALUES (?1, ?2, ?3, ?4, ?5)", + (1_i64, "sess-1", "msg-1", "assistant", JS_MAX_SAFE_INTEGER + 1), + ) + .expect("direct message insert should succeed"); + + let reader = AgentTraceExportReader::new(&db); + let error = reader + .read_messages_after(0, 500) + .expect_err("row above safe-integer bound should error"); + assert!(error.to_string().contains("JS-safe-integer")); + + remove_test_db(&db_path); + } + + #[test] + fn read_messages_after_performs_no_mutation() { + let db_path = unique_test_db_path("messages-no-mutation"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_message_row_with_id(&db, 1, "msg-1"); + insert_message_row_with_id(&db, 2, "msg-2"); + + let messages_before = row_count(&db, "messages"); + let metadata_before = row_count(&db, "repository_metadata"); + + let reader = AgentTraceExportReader::new(&db); + reader + .read_messages_after(0, 500) + .expect("read should succeed"); + + assert_eq!(row_count(&db, "messages"), messages_before); + assert_eq!(row_count(&db, "repository_metadata"), metadata_before); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_returns_rows_after_cursor_in_order() { + let db_path = unique_test_db_path("parts-basic"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.insert_parts(vec![ + InsertPartInsert { + part_type: PartType::Text, + text: "first".to_string(), + session_id: "sess-1".to_string(), + message_id: "msg-1".to_string(), + generated_at_unix_ms: 1_001, + }, + InsertPartInsert { + part_type: PartType::Patch, + text: "second".to_string(), + session_id: "sess-1".to_string(), + message_id: "msg-2".to_string(), + generated_at_unix_ms: 1_002, + }, + ]) + .expect("seed parts should insert"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_parts_after(0, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![1, 2] + ); + assert_eq!(rows[1].part_type, "patch"); + assert_eq!(rows[1].text, "second"); + assert_eq!(rows[1].message_id, "msg-2"); + assert_eq!(rows[1].generated_at_unix_ms, 1_002); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_returns_non_contiguous_ids() { + let db_path = unique_test_db_path("parts-gap"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for (id, message_id) in [ + (11, "msg-11"), + (15, "msg-15"), + (19, "msg-19"), + (30, "msg-30"), + ] { + insert_part_row_with_id(&db, id, message_id); + } + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_parts_after(10, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![11, 15, 19, 30] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_limit_truncates_and_follow_up_continues() { + let db_path = unique_test_db_path("parts-limit"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for id in 1..=10 { + insert_part_row_with_id(&db, id, &format!("msg-{id}")); + } + + let reader = AgentTraceExportReader::new(&db); + let first_batch = reader + .read_parts_after(0, 3) + .expect("first limited read should succeed"); + assert_eq!( + first_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![1, 2, 3] + ); + + let next_cursor = first_batch + .last() + .expect("first batch non-empty") + .source_row_id; + let second_batch = reader + .read_parts_after(next_cursor, 3) + .expect("follow-up read should succeed"); + assert_eq!( + second_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![4, 5, 6] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_returns_empty_at_or_beyond_max_id() { + let db_path = unique_test_db_path("parts-empty-tail"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_part_row_with_id(&db, 1, "msg-1"); + + let reader = AgentTraceExportReader::new(&db); + assert!(reader + .read_parts_after(1, 500) + .expect("read at max id should succeed") + .is_empty()); + assert!(reader + .read_parts_after(100, 500) + .expect("read beyond max id should succeed") + .is_empty()); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_rejects_invalid_cursor_and_limit() { + let db_path = unique_test_db_path("parts-invalid"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + let reader = AgentTraceExportReader::new(&db); + + let cursor_error = reader + .read_parts_after(-1, 500) + .expect_err("negative cursor should error"); + assert!(cursor_error.to_string().contains("cursor")); + + let zero_limit_error = reader + .read_parts_after(0, 0) + .expect_err("zero limit should error"); + assert!(zero_limit_error.to_string().contains("limit")); + + let excess_limit_error = reader + .read_parts_after(0, AGENT_TRACE_EXPORT_BATCH_SIZE + 1) + .expect_err("excess limit should error"); + assert!(excess_limit_error.to_string().contains("limit")); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_rejects_row_above_safe_integer_bound() { + let db_path = unique_test_db_path("parts-unsafe-integer"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.execute( + "INSERT INTO parts (id, type, text, message_id, session_id, generated_at_unix_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + (1_i64, "text", "hello", "msg-1", "sess-1", JS_MAX_SAFE_INTEGER + 1), + ) + .expect("direct part insert should succeed"); + + let reader = AgentTraceExportReader::new(&db); + let error = reader + .read_parts_after(0, 500) + .expect_err("row above safe-integer bound should error"); + assert!(error.to_string().contains("JS-safe-integer")); + + remove_test_db(&db_path); + } + + #[test] + fn read_parts_after_performs_no_mutation() { + let db_path = unique_test_db_path("parts-no-mutation"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_part_row_with_id(&db, 1, "msg-1"); + insert_part_row_with_id(&db, 2, "msg-2"); + + let parts_before = row_count(&db, "parts"); + let metadata_before = row_count(&db, "repository_metadata"); + + let reader = AgentTraceExportReader::new(&db); + reader + .read_parts_after(0, 500) + .expect("read should succeed"); + + assert_eq!(row_count(&db, "parts"), parts_before); + assert_eq!(row_count(&db, "repository_metadata"), metadata_before); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_returns_rows_after_cursor_in_order() { + let db_path = unique_test_db_path("diff-traces-basic"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.insert_diff_trace(DiffTraceInsert { + time_ms: 1_001, + session_id: "sess-1", + patch: "Index: a\n", + model_id: Some("test-provider/test-model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("seed diff_trace should insert"); + db.insert_diff_trace(DiffTraceInsert { + time_ms: 1_002, + session_id: "sess-1", + patch: "{\"tool\":\"Edit\"}", + model_id: None, + tool_name: "claude", + tool_version: None, + payload_type: PAYLOAD_TYPE_STRUCTURED, + }) + .expect("seed diff_trace should insert"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_diff_traces_after(1, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![2] + ); + assert_eq!(rows[0].patch, "{\"tool\":\"Edit\"}"); + assert_eq!(rows[0].payload_type, "structured"); + assert_eq!(rows[0].model_id, None); + assert_eq!(rows[0].tool_name, Some("claude".to_string())); + assert_eq!(rows[0].tool_version, None); + assert_eq!(rows[0].time_ms, 1_002); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_preserves_populated_nullable_fields() { + let db_path = unique_test_db_path("diff-traces-nullable-populated"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.insert_diff_trace(DiffTraceInsert { + time_ms: 1_001, + session_id: "sess-1", + patch: "Index: a\n", + model_id: Some("test-provider/test-model"), + tool_name: "opencode", + tool_version: Some("1.2.3"), + payload_type: PAYLOAD_TYPE_PATCH, + }) + .expect("seed diff_trace should insert"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_diff_traces_after(0, 500) + .expect("read after cursor should succeed"); + + assert_eq!(rows.len(), 1); + assert_eq!( + rows[0].model_id, + Some("test-provider/test-model".to_string()) + ); + assert_eq!(rows[0].tool_name, Some("opencode".to_string())); + assert_eq!(rows[0].tool_version, Some("1.2.3".to_string())); + assert_eq!(rows[0].patch, "Index: a\n"); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_returns_non_contiguous_ids() { + let db_path = unique_test_db_path("diff-traces-gap"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for (id, session_id) in [ + (11, "sess-11"), + (15, "sess-15"), + (19, "sess-19"), + (30, "sess-30"), + ] { + insert_diff_trace_row_with_id(&db, id, session_id); + } + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_diff_traces_after(10, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![11, 15, 19, 30] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_limit_truncates_and_follow_up_continues() { + let db_path = unique_test_db_path("diff-traces-limit"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for id in 1..=10 { + insert_diff_trace_row_with_id(&db, id, &format!("sess-{id}")); + } + + let reader = AgentTraceExportReader::new(&db); + let first_batch = reader + .read_diff_traces_after(0, 3) + .expect("first limited read should succeed"); + assert_eq!( + first_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![1, 2, 3] + ); + + let next_cursor = first_batch + .last() + .expect("first batch non-empty") + .source_row_id; + let second_batch = reader + .read_diff_traces_after(next_cursor, 3) + .expect("follow-up read should succeed"); + assert_eq!( + second_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![4, 5, 6] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_returns_empty_at_or_beyond_max_id() { + let db_path = unique_test_db_path("diff-traces-empty-tail"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_diff_trace_row_with_id(&db, 1, "sess-1"); + + let reader = AgentTraceExportReader::new(&db); + assert!(reader + .read_diff_traces_after(1, 500) + .expect("read at max id should succeed") + .is_empty()); + assert!(reader + .read_diff_traces_after(100, 500) + .expect("read beyond max id should succeed") + .is_empty()); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_rejects_invalid_cursor_and_limit() { + let db_path = unique_test_db_path("diff-traces-invalid"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + let reader = AgentTraceExportReader::new(&db); + + let cursor_error = reader + .read_diff_traces_after(-1, 500) + .expect_err("negative cursor should error"); + assert!(cursor_error.to_string().contains("cursor")); + + let zero_limit_error = reader + .read_diff_traces_after(0, 0) + .expect_err("zero limit should error"); + assert!(zero_limit_error.to_string().contains("limit")); + + let excess_limit_error = reader + .read_diff_traces_after(0, AGENT_TRACE_EXPORT_BATCH_SIZE + 1) + .expect_err("excess limit should error"); + assert!(excess_limit_error.to_string().contains("limit")); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_rejects_row_above_safe_integer_bound() { + let db_path = unique_test_db_path("diff-traces-unsafe-integer"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.execute( + "INSERT INTO diff_traces (id, time_ms, session_id, patch, model_id, tool_name, tool_version, payload_type) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + ( + 1_i64, + JS_MAX_SAFE_INTEGER + 1, + "sess-1", + "Index: a\n", + Option::<&str>::None, + Option::<&str>::None, + Option::<&str>::None, + PAYLOAD_TYPE_PATCH, + ), + ) + .expect("direct diff_trace insert should succeed"); + + let reader = AgentTraceExportReader::new(&db); + let error = reader + .read_diff_traces_after(0, 500) + .expect_err("row above safe-integer bound should error"); + assert!(error.to_string().contains("JS-safe-integer")); + + remove_test_db(&db_path); + } + + #[test] + fn read_diff_traces_after_performs_no_mutation() { + let db_path = unique_test_db_path("diff-traces-no-mutation"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_diff_trace_row_with_id(&db, 1, "sess-1"); + insert_diff_trace_row_with_id(&db, 2, "sess-2"); + + let diff_traces_before = row_count(&db, "diff_traces"); + let metadata_before = row_count(&db, "repository_metadata"); + + let reader = AgentTraceExportReader::new(&db); + reader + .read_diff_traces_after(0, 500) + .expect("read should succeed"); + + assert_eq!(row_count(&db, "diff_traces"), diff_traces_before); + assert_eq!(row_count(&db, "repository_metadata"), metadata_before); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_returns_rows_after_cursor_in_order() { + let db_path = unique_test_db_path("agent-traces-basic"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.insert_agent_trace(AgentTraceInsert { + commit_id: "abc123", + commit_time_ms: 1_001, + trace_json: "{\"steps\":[]}", + agent_trace_id: "trace-1", + url: "https://example.com/trace/1", + remote_url: "", + }) + .expect("seed agent_trace should insert"); + db.insert_agent_trace(AgentTraceInsert { + commit_id: "def456", + commit_time_ms: 1_002, + trace_json: "{\"steps\":[1]}", + agent_trace_id: "trace-2", + url: "https://example.com/trace/2", + remote_url: "https://github.com/org/repo/commit/def456", + }) + .expect("seed agent_trace should insert"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_agent_traces_after(1, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![2] + ); + assert_eq!(rows[0].agent_trace_id, "trace-2"); + assert_eq!(rows[0].commit_id, "def456"); + assert_eq!(rows[0].commit_time_ms, 1_002); + assert_eq!(rows[0].trace_json, "{\"steps\":[1]}"); + assert_eq!(rows[0].url, "https://example.com/trace/2"); + assert_eq!( + rows[0].remote_url, + Some("https://github.com/org/repo/commit/def456".to_string()) + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_preserves_null_remote_url() { + let db_path = unique_test_db_path("agent-traces-remote-url-null"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_agent_trace_row_with_id(&db, 1, "trace-null"); + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_agent_traces_after(0, 500) + .expect("read after cursor should succeed"); + + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].remote_url, None); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_returns_non_contiguous_ids() { + let db_path = unique_test_db_path("agent-traces-gap"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for (id, agent_trace_id) in [ + (11, "trace-11"), + (15, "trace-15"), + (19, "trace-19"), + (30, "trace-30"), + ] { + insert_agent_trace_row_with_id(&db, id, agent_trace_id); + } + + let reader = AgentTraceExportReader::new(&db); + let rows = reader + .read_agent_traces_after(10, 500) + .expect("read after cursor should succeed"); + + assert_eq!( + rows.iter().map(|row| row.source_row_id).collect::>(), + vec![11, 15, 19, 30] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_limit_truncates_and_follow_up_continues() { + let db_path = unique_test_db_path("agent-traces-limit"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + for id in 1..=10 { + insert_agent_trace_row_with_id(&db, id, &format!("trace-{id}")); + } + + let reader = AgentTraceExportReader::new(&db); + let first_batch = reader + .read_agent_traces_after(0, 3) + .expect("first limited read should succeed"); + assert_eq!( + first_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![1, 2, 3] + ); + + let next_cursor = first_batch + .last() + .expect("first batch non-empty") + .source_row_id; + let second_batch = reader + .read_agent_traces_after(next_cursor, 3) + .expect("follow-up read should succeed"); + assert_eq!( + second_batch + .iter() + .map(|row| row.source_row_id) + .collect::>(), + vec![4, 5, 6] + ); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_returns_empty_at_or_beyond_max_id() { + let db_path = unique_test_db_path("agent-traces-empty-tail"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_agent_trace_row_with_id(&db, 1, "trace-1"); + + let reader = AgentTraceExportReader::new(&db); + assert!(reader + .read_agent_traces_after(1, 500) + .expect("read at max id should succeed") + .is_empty()); + assert!(reader + .read_agent_traces_after(100, 500) + .expect("read beyond max id should succeed") + .is_empty()); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_rejects_invalid_cursor_and_limit() { + let db_path = unique_test_db_path("agent-traces-invalid"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + let reader = AgentTraceExportReader::new(&db); + + let cursor_error = reader + .read_agent_traces_after(-1, 500) + .expect_err("negative cursor should error"); + assert!(cursor_error.to_string().contains("cursor")); + + let zero_limit_error = reader + .read_agent_traces_after(0, 0) + .expect_err("zero limit should error"); + assert!(zero_limit_error.to_string().contains("limit")); + + let excess_limit_error = reader + .read_agent_traces_after(0, AGENT_TRACE_EXPORT_BATCH_SIZE + 1) + .expect_err("excess limit should error"); + assert!(excess_limit_error.to_string().contains("limit")); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_rejects_row_above_safe_integer_bound() { + let db_path = unique_test_db_path("agent-traces-unsafe-integer"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + db.execute( + "INSERT INTO agent_traces (id, commit_id, commit_time_ms, trace_json, agent_trace_id, url, remote_url) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + ( + 1_i64, + "abc123", + JS_MAX_SAFE_INTEGER + 1, + "{\"steps\":[]}", + "trace-unsafe", + "https://example.com/trace", + Option::<&str>::None, + ), + ) + .expect("direct agent_trace insert should succeed"); + + let reader = AgentTraceExportReader::new(&db); + let error = reader + .read_agent_traces_after(0, 500) + .expect_err("row above safe-integer bound should error"); + assert!(error.to_string().contains("JS-safe-integer")); + + remove_test_db(&db_path); + } + + #[test] + fn read_agent_traces_after_performs_no_mutation() { + let db_path = unique_test_db_path("agent-traces-no-mutation"); + let db = RepositoryAgentTraceDb::new_at(&db_path).expect("test DB should open"); + insert_agent_trace_row_with_id(&db, 1, "trace-1"); + insert_agent_trace_row_with_id(&db, 2, "trace-2"); + + let agent_traces_before = row_count(&db, "agent_traces"); + let metadata_before = row_count(&db, "repository_metadata"); + + let reader = AgentTraceExportReader::new(&db); + reader + .read_agent_traces_after(0, 500) + .expect("read should succeed"); + + assert_eq!(row_count(&db, "agent_traces"), agent_traces_before); + assert_eq!(row_count(&db, "repository_metadata"), metadata_before); + + remove_test_db(&db_path); + } + + #[test] + fn source_instance_integration() { + use crate::services::agent_trace_storage::{ + resolve_agent_trace_storage_at_state_root, AgentTraceStorageContext, + }; + + let state_root = unique_storage_temp_dir("state"); + let repo_root = + init_git_repo_with_remote("repo", "git@github.com:acme/agent-trace-export-readers.git"); + + let context = AgentTraceStorageContext { + repository_root: &repo_root, + explicit_repository_id: None, + repository_remote: "origin", + }; + + let storage = resolve_agent_trace_storage_at_state_root(&context, &state_root) + .expect("repository-scoped Agent Trace storage should resolve"); + + assert!(!storage.metadata.repository_id.trim().is_empty()); + assert!(!storage.metadata.source_instance_id.trim().is_empty()); + + insert_message_row_with_id(&storage.db, 1, "msg-1"); + + let reader = AgentTraceExportReader::new(&storage.db); + let rows = reader + .read_messages_after(0, AGENT_TRACE_EXPORT_BATCH_SIZE) + .expect("read through storage-resolved db should succeed"); + + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].source_row_id, 1); + assert_eq!(rows[0].message_id, "msg-1"); + + fs::remove_dir_all(&state_root).expect("clean up state root"); + fs::remove_dir_all(&repo_root).expect("clean up repo root"); + } + + fn unique_storage_temp_dir(label: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after Unix epoch") + .as_nanos(); + let dir = std::env::temp_dir().join(format!( + "sce-agent-trace-export-storage-{label}-{}-{nonce}", + std::process::id() + )); + fs::create_dir_all(&dir).expect("create temp dir"); + dir + } + + fn init_git_repo_with_remote(label: &str, remote_url: &str) -> PathBuf { + let repo = unique_storage_temp_dir(label); + git(&repo, &["init", "-q"]); + git(&repo, &["remote", "add", "origin", remote_url]); + repo + } + + fn git(repo_root: &std::path::Path, args: &[&str]) { + let output = std::process::Command::new("git") + .args(args) + .current_dir(repo_root) + .output() + .unwrap_or_else(|error| panic!("git {args:?} failed to spawn: {error}")); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + } +} diff --git a/cli/src/services/mod.rs b/cli/src/services/mod.rs index 0f2251b1..3d814bd3 100644 --- a/cli/src/services/mod.rs +++ b/cli/src/services/mod.rs @@ -1,6 +1,8 @@ pub mod agent_trace; pub mod agent_trace_db; #[allow(dead_code)] +pub mod agent_trace_export; +#[allow(dead_code)] pub mod agent_trace_storage; pub mod app_support; pub mod auth; diff --git a/context/context-map.md b/context/context-map.md index aa291d8a..4f64b6b8 100644 --- a/context/context-map.md +++ b/context/context-map.md @@ -62,6 +62,7 @@ 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`, 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 the fresh multi-statement baseline schema plus the additive `source_instance_id` migration, with `repository_metadata` carrying both `repository_id` and a concurrency-safe atomic-claim `source_instance_id` (physical database identity, independent of `repository_id`), 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-export-readers.md` (implemented `AgentTraceExportReader<'a>` in `cli/src/services/agent_trace_export/mod.rs`: the read-only local export boundary over `RepositoryAgentTraceDb` — `read_messages_after`/`read_parts_after`/`read_diff_traces_after`/`read_agent_traces_after`, each cursor/limit/JS-safe-integer validated, materialized into owned camelCase `serde::Serialize` DTOs; composes directly with `ResolvedAgentTraceStorage` without owning `source_instance_id`; no local sync cursor, no `agent-trace-sync.db`, no Turso Sync, no ETL, no DWH) - `context/sce/agent-trace-core-schema-migrations.md` (historical reference for removed local DB schema bootstrap behavior; T03 now implements the actual local DB with migrations) - `context/sce/agent-trace-retry-queue-observability.md` (inactive local-hook retry path plus historical retry/metrics reference) - `context/sce/agent-trace-local-hooks-mvp-contract-gap-matrix.md` (T01 Local Hooks MVP production contract freeze and deterministic gap matrix for `agent-trace-local-hooks-production-mvp`) diff --git a/context/glossary.md b/context/glossary.md index 2988023c..ef768430 100644 --- a/context/glossary.md +++ b/context/glossary.md @@ -237,3 +237,4 @@ - `musl static Linux release`: The Linux binary release targets (`x86_64-unknown-linux-musl` and `aarch64-unknown-linux-musl`) compile against musl libc and link fully statically. The resulting binary has no runtime libc dependency and zero `/nix/store/` references in ELF metadata, strings, or dynamic-linker fields, satisfying the native portability audit. The musl targets replace the previous glibc-linked `*-unknown-linux-gnu` targets; macOS (`aarch64-apple-darwin`) is unchanged. Introduced in the `musl-static-linux-release` plan. - `parts table (Agent Trace DB)`: Agent Trace DB table created by migration `009_create_parts.sql`; stores append-only message parts with columns `type` (typed by Rust as `text`/`reasoning`/`patch`/`question` and stored as unconstrained `TEXT NOT NULL`), `text`, `message_id`, `session_id`, `generated_at_unix_ms`, `created_at`, `updated_at`. Uses only the internal `id` for row identity (no upsert/dedup). Multiple parts can exist for the same `(session_id, message_id)`. A compound index on `(session_id, message_id, generated_at_unix_ms, id)` enables ordered joins. No foreign keys to `messages` or any other table, so parts may be inserted before their parent message exists. +- `AgentTraceExportReader`: Read-only incremental export reader in `cli/src/services/agent_trace_export/mod.rs` over one `RepositoryAgentTraceDb`, exposing `read_messages_after`/`read_parts_after`/`read_diff_traces_after`/`read_agent_traces_after`, each `(cursor: i64, limit: usize) -> Result>` over `WHERE id > cursor ORDER BY id ASC LIMIT limit`. Holds no local cursor, performs no mutation, makes no network calls, and returns owned camelCase `serde::Serialize` export-row DTOs matching the shipped control-plane ingestion contract. See `context/sce/agent-trace-export-readers.md`. diff --git a/context/plans/agent-trace-export-readers.md b/context/plans/agent-trace-export-readers.md new file mode 100644 index 00000000..c92b9402 --- /dev/null +++ b/context/plans/agent-trace-export-readers.md @@ -0,0 +1,145 @@ +# Plan: agent-trace-export-readers + +## Change summary + +Add incremental local Agent Trace export readers for the four capture streams +(`messages`, `parts`, `diff_traces`, `agent_traces`) stored in the +repository-scoped `RepositoryAgentTraceDb` established by the `source-instance` +plan (PR #197). This establishes the local read/export boundary — cursor in, +owned wire-compatible rows out — that the next PR will compose with a +control-plane HTTP client and `sce trace sync` orchestration. + +This plan adds a new `AgentTraceExportReader` seam plus four owned, +`serde::Serialize`-derived export-row DTOs with camelCase JSON matching the +already-shipped control-plane ingestion contract. Each reader method takes a +`cursor: i64` (last server-accepted `table.id`) and a `limit: usize`, runs +`SELECT ... WHERE id > ?1 ORDER BY id ASC LIMIT ?2`, validates cursor/limit/ +JS-safe-integer bounds, and returns a fully materialized `Vec<...>` — no open +transaction, no iterator, no borrowed row state, no network, no local cursor, +no auth. `post_commit_patch_intersections` is not exported. + +This is purely additive: no existing writer, schema, or hook behavior changes. + +## Acceptance criteria + +- [x] AC1: `AgentTraceExportReader::read_messages_after(cursor, limit)` returns owned `AgentTraceMessageExportRow` values for `messages.id > cursor`, ordered by `id ASC`, capped at `limit`, with `sourceRowId` equal to the local `id` unmodified. + - Validate: `cargo test -p shared-context-engineering --lib services::agent_trace_export` +- [x] AC2: The same contract holds for `read_parts_after`, `read_diff_traces_after`, and `read_agent_traces_after`, including exact column mapping, nullable-field preservation as `Option` → JSON `null`, and no gap/contiguity assumption. + - Validate: `cargo test -p shared-context-engineering --lib services::agent_trace_export` +- [x] AC3: Every export row type serializes via `serde_json` to the exact camelCase shape already shipped by the control-plane ingestion contract (field names and value shapes as specified in this plan's DTO sections). + - Validate: `cargo test -p shared-context-engineering --lib services::agent_trace_export::tests` (serialization contract tests) +- [x] AC4: Readers reject `cursor < 0`, `limit == 0`, and `limit > AGENT_TRACE_EXPORT_BATCH_SIZE` (500) with a clear error and no query execution; readers reject rows whose exportable numeric fields fall outside `0..=9_007_199_254_740_991` with a clear export error instead of truncating/casting. + - Validate: `cargo test -p shared-context-engineering --lib services::agent_trace_export::tests` (validation tests) +- [x] AC5: Invoking any reader method performs no database mutation (no inserts, no cursor table, no metadata writes). + - Validate: inspection of `AgentTraceExportReader` (read-only `SELECT` methods only, no `INSERT`/`UPDATE`/`DDL`) plus a test asserting row counts/`repository_metadata` are unchanged after a read. +- [x] AC6: A `RepositoryAgentTraceDb` opened through the existing repository storage resolver (from PR #197) can be passed directly into `AgentTraceExportReader::new(&storage.db)` and read at least one stream, proving the composition point without the reader owning or generating `source_instance_id`. + - Validate: `cargo test -p shared-context-engineering --lib services::agent_trace_export::tests::source_instance_integration` (or equivalently named integration test) + +### Full validation + +- `nix run .#pkl-check-generated` +- `nix flake check` + +### Context sync + +- `context/sce/agent-trace-db.md` and/or a new `context/sce/agent-trace-export-readers.md` documenting the reader boundary, the four stream queries, and the explicit no-local-cursor / no-sync-db statement. +- `context/context-map.md` entry for the new domain file if a dedicated file is created. + +## Constraints and non-goals + +- **In scope:** a new `cli/src/services/agent_trace_export/` (or `agent_trace_export.rs`) module; four export DTOs; `AgentTraceExportReader` with four `read_*_after` methods; the `AGENT_TRACE_EXPORT_BATCH_SIZE` constant; cursor/limit/safe-integer validation; reader unit + integration tests; durable context documentation. +- **Out of scope:** `sce trace sync`, any HTTP client or control-plane request/response types beyond the export row DTOs, WorkOS/auth changes, server cursor fetching, `POST /agent-trace/ingestion/batch`, retry/backoff against a remote server, local cursor persistence, any new database or table, patch parsing/normalization, `code_changes` derivation, analytics. +- **Constraints:** no new local sync database; no bridge lock; no changes to `RepositoryAgentTraceDb` write semantics, migrations, or schema; must not derive or accept a source identity other than the existing `RepositoryMetadata.source_instance_id`; must not UUID-parse `source_instance_id`; JSON field names are camelCase and must match the already-shipped control-plane contract exactly as specified in this plan. +- **Non-goal:** building any part of the future `sce trace sync` command, config, or CLI surface. This plan produces a library-level reader only; nothing here is user-invocable. + +## Task stack + +- [x] T01: `Add export batch-size constant, error handling for cursor/limit, and safe-integer validation helper` (status:done) + - Task ID: T01 + - Goal: Establish the shared validation primitives every reader method depends on: `pub const AGENT_TRACE_EXPORT_BATCH_SIZE: usize = 500;`, a cursor validator rejecting `cursor < 0`, a limit validator rejecting `limit == 0` and `limit > AGENT_TRACE_EXPORT_BATCH_SIZE`, and a JS-safe-integer validator rejecting values outside `0..=9_007_199_254_740_991` (`Number.MAX_SAFE_INTEGER`), all returning `anyhow::Result` with clear, distinct error messages per failure mode (matching this repository's established `anyhow` error convention). + - Boundaries (in/out of scope): In — the constant, validator functions/helpers, and their unit tests in a new `cli/src/services/agent_trace_export/mod.rs`. Out — any reader method, any DTO, any SQL. + - Dependencies: none + - Done when: `cargo test -p shared-context-engineering --lib services::agent_trace_export` passes with unit tests for cursor `< 0` rejection, limit `0`/`501` rejection, limit `500` acceptance, and safe-integer boundary acceptance/rejection at `0`, `9_007_199_254_740_991`, and `9_007_199_254_740_992`. + - Verification notes (commands or checks): `cargo test -p shared-context-engineering --lib services::agent_trace_export` + - Evidence: Added `cli/src/services/agent_trace_export/mod.rs` with `AGENT_TRACE_EXPORT_BATCH_SIZE = 500`, `JS_MAX_SAFE_INTEGER`, and `validate_cursor`/`validate_limit`/`validate_js_safe_integer` (each returning `anyhow::Result<()>` with a distinct `bail!` message per failure mode). Registered the module in `cli/src/services/mod.rs` with `#[allow(dead_code)]` (unused until later tasks wire it up). 9 unit tests cover cursor `<0` rejection and `>=0` acceptance, limit `0`/`501` rejection and `1`/`500` acceptance, and safe-integer rejection at `-1`/`9_007_199_254_740_992` with acceptance at `0`/`9_007_199_254_740_991`. + - Verification run: `nix flake check` (repository policy blocks direct `cargo test`; the `cli-tests` flake check runs the full workspace suite including the new `services::agent_trace_export` tests) — all 3 checks (`cli-tests`, `cli-clippy`, `cli-fmt`) passed. + - Deviations: none. + +- [x] T02: `Define the four export DTOs with camelCase Serialize and serialization contract tests` (status:done) + - Task ID: T02 + - Goal: Define `AgentTraceMessageExportRow`, `AgentTracePartExportRow`, `AgentTraceDiffTraceExportRow`, and `AgentTraceAgentTraceExportRow` as owned structs deriving `serde::Serialize` with `#[serde(rename_all = "camelCase")]`, matching field-for-field the JSON shapes specified in the change request (messages: `sourceRowId, sessionId, messageId, role, generatedAtUnixMs`; parts: `sourceRowId, sessionId, messageId, type, text, generatedAtUnixMs`; diff_traces: `sourceRowId, sessionId, timeMs, patch, modelId, toolName, toolVersion, payloadType` with `modelId`/`toolName`/`toolVersion` as `Option`; agent_traces: `sourceRowId, agentTraceId, commitId, commitTimeMs, traceJson, url, remoteUrl` with `remoteUrl: Option`). Reuse the existing `agent_trace_db::MessageRole` enum for `role` by adding a `Serialize` derive with lowercase rename (`user`/`assistant`) rather than introducing a parallel role type; do not export local `created_at`/`updated_at`. + - Boundaries (in/out of scope): In — the four struct definitions, `MessageRole`'s added `Serialize` derive, and `serde_json`-based serialization contract tests asserting exact JSON shape for representative rows of each type (including `null` for `None` fields). Out — any reader method, any SQL, any DB access. + - Dependencies: T01 + - Done when: `cargo test` for the new serialization contract tests passes, asserting JSON output byte-for-byte (via `serde_json::json!` comparison or exact string) for one representative row per stream, including a diff_trace row with all-`None` nullable fields and one with all populated, and an agent_trace row with `remoteUrl: null`. + - Verification notes (commands or checks): `cargo test -p shared-context-engineering --lib services::agent_trace_export` + - Evidence: Added `AgentTraceMessageExportRow`, `AgentTracePartExportRow`, `AgentTraceDiffTraceExportRow`, and `AgentTraceAgentTraceExportRow` to `cli/src/services/agent_trace_export/mod.rs`, each `#[derive(Clone, Debug, PartialEq, Serialize)]` with `#[serde(rename_all = "camelCase")]` (the `parts` row's `type` field uses `#[serde(rename = "type")]` since `type` is a Rust keyword). Added `#[derive(..., serde::Serialize)]` with `#[serde(rename_all = "lowercase")]` to `MessageRole` in `cli/src/services/agent_trace_db/mod.rs`, reused directly as the `role` field type — no parallel role type introduced. Added 8 serialization contract tests via `serde_json::to_value`/`json!` comparison covering: message row with `assistant` role, message row with `user` role (lowercase check), part row full shape, diff_trace row with all nullable fields populated, diff_trace row with all nullable fields `None` → JSON `null`, agent_trace row with `remoteUrl: null`, and agent_trace row with `remoteUrl` populated. + - Verification run: `nix flake check` — all 3 checks (`cli-tests`, `cli-clippy`, `cli-fmt`) passed. + - Deviations: none. + +- [x] T03: `Implement AgentTraceExportReader::read_messages_after and read_parts_after with full test coverage` (status:done) + - Task ID: T03 + - Goal: Implement `pub struct AgentTraceExportReader<'a> { db: &'a RepositoryAgentTraceDb }` with `pub fn new(db: &'a RepositoryAgentTraceDb) -> Self`, `read_messages_after(&self, cursor: i64, limit: usize) -> Result>`, and `read_parts_after(&self, cursor: i64, limit: usize) -> Result>`. Each validates cursor/limit via T01 helpers, runs `SELECT id, session_id, message_id, role, generated_at_unix_ms FROM messages WHERE id > ?1 ORDER BY id ASC LIMIT ?2` (and the equivalent for `parts`) via the existing `TursoDb` query API, validates safe-integer bounds on `sourceRowId`/`generatedAtUnixMs` per row before returning, and materializes fully into owned `Vec<...>` before returning (no borrowed row/iterator/transaction survives the call). + - Boundaries (in/out of scope): In — the reader struct, `new`, `read_messages_after`, `read_parts_after`, and their tests (incremental read with a gap, limit truncation plus follow-up read, empty result at/above max ID, invalid-cursor test, invalid-limit tests, safe-integer rejection test, no-mutation test). Out — `read_diff_traces_after`, `read_agent_traces_after`, any HTTP/network code. + - Dependencies: T02 + - Done when: tests seed a temporary `RepositoryAgentTraceDb` (following the existing `unique_test_db_path`/`RepositoryAgentTraceDb::new_at` pattern in `cli/src/services/agent_trace_db/repository.rs`), insert messages/parts via the existing `insert_message`/`insert_messages`/`insert_part`/`insert_parts` helpers or direct SQL to control exact `id` values, and assert: seeded IDs `1,2,3` with `cursor=1` returns `[2,3]` in order with exact field mapping; a non-contiguous ID set (e.g. `11,15,19,30` after `cursor=10`) returns all four; `limit=3` over 10 rows returns IDs `1..3` and a follow-up read from `cursor=3` continues correctly; `cursor` at/beyond the max ID returns an empty `Vec`; `cursor=-1` errors with no query executed; `limit=0` and `limit=501` error; a row with `generated_at_unix_ms > 9_007_199_254_740_991` (seeded via direct SQL) errors; and read calls leave row counts and `repository_metadata` unchanged. + - Verification notes (commands or checks): `cargo test -p shared-context-engineering --lib services::agent_trace_export` + - Evidence: Added `AgentTraceExportReader<'a>` (wrapping `&'a RepositoryAgentTraceDb`), `new`, `read_messages_after`, and `read_parts_after` to `cli/src/services/agent_trace_export/mod.rs`, plus the `SELECT_MESSAGES_AFTER_SQL`/`SELECT_PARTS_AFTER_SQL` query constants, row-mapping functions (`message_export_row_from_turso`, `part_export_row_from_turso`), a `message_role_from_column` helper reused for role decoding, and a `limit_as_i64` helper for the clippy-clean `usize` → `i64` `LIMIT` param conversion. Each method validates cursor/limit via the T01 helpers before querying and validates `source_row_id`/`generated_at_unix_ms` per returned row via `validate_js_safe_integer` before returning the fully materialized `Vec`. Added 16 new tests (8 per method) covering: cursor-ordered read after a gap, non-contiguous IDs, limit truncation plus a follow-up read continuing from the last returned ID, empty result at/beyond the max ID, negative-cursor and zero/excess-limit rejection, per-row safe-integer rejection (seeded via direct SQL), and a no-mutation assertion comparing `messages`/`parts`/`repository_metadata` row counts before and after a read. + - Verification run: `nix flake check` (repository policy blocks direct `cargo test`; the `cli-tests` flake check runs the full workspace suite including the new reader tests) — all 3 checks (`cli-tests`, `cli-clippy`, `cli-fmt`) passed. + - Deviations: none. + +- [x] T04: `Implement AgentTraceExportReader::read_diff_traces_after and read_agent_traces_after with full test coverage` (status:done) + - Task ID: T04 + - Goal: Implement `read_diff_traces_after(&self, cursor: i64, limit: usize) -> Result>` and `read_agent_traces_after(&self, cursor: i64, limit: usize) -> Result>` on `AgentTraceExportReader`, following the same cursor/limit/safe-integer validation and materialization discipline as T03. `read_diff_traces_after` selects `id, session_id, time_ms, patch, model_id, tool_name, tool_version, payload_type` from `diff_traces` and preserves `patch`/`payload_type` raw and unmodified (no patch parsing, no normalizer call). `read_agent_traces_after` selects `id, agent_trace_id, commit_id, commit_time_ms, trace_json, url, remote_url` from `agent_traces` and preserves `trace_json` as the exact raw string from SQLite (no parse/reserialize). + - Boundaries (in/out of scope): In — the two reader methods and their tests (nullable `model_id`/`tool_name`/`tool_version` both populated and `NULL`; raw `patch`/`payload_type` passthrough; `remote_url = NULL` → JSON `null`; `trace_json` byte-for-byte passthrough; gap/limit/empty/invalid-cursor/invalid-limit/safe-integer/no-mutation tests mirroring T03's coverage for these two streams). Out — `read_messages_after`, `read_parts_after` (T03), any HTTP/network code, any use of `cli/src/services/patch.rs` or `structured_patch.rs`. + - Dependencies: T03 + - Done when: all diff_traces/agent_traces reader tests pass per the Goal, and `AgentTraceExportReader` now exposes all four `read_*_after` methods with the complete `pub` API shape from the change request (`reader.read_messages_after(cursor, limit)`, etc.), ready for direct use by a future `sce trace sync` command. + - Verification notes (commands or checks): `cargo test -p shared-context-engineering --lib services::agent_trace_export` + - Evidence: Added `read_diff_traces_after` and `read_agent_traces_after` to `AgentTraceExportReader` in `cli/src/services/agent_trace_export/mod.rs`, plus `SELECT_DIFF_TRACES_AFTER_SQL`/`SELECT_AGENT_TRACES_AFTER_SQL` query constants and `diff_trace_export_row_from_turso`/`agent_trace_export_row_from_turso` row mappers, mirroring T03's validate-then-query-then-per-row-safe-integer-check discipline (`source_row_id`/`time_ms` for diff_traces, `source_row_id`/`commit_time_ms` for agent_traces). `patch`/`payload_type`/`trace_json` are passed through unmodified with no parsing. Added 20 new tests (10 per method) covering: cursor-ordered read with mixed `patch`/`structured` payload types, populated-nullable-fields passthrough, non-contiguous IDs, limit truncation plus follow-up continuation, empty result at/beyond max ID, negative-cursor and zero/excess-limit rejection, per-row safe-integer rejection (seeded via direct SQL), null `remote_url` passthrough as `None`, and a no-mutation assertion comparing table/`repository_metadata` row counts before and after a read. `AgentTraceExportReader` now exposes all four `read_*_after` methods (`read_messages_after`, `read_parts_after`, `read_diff_traces_after`, `read_agent_traces_after`). + - Verification run: `nix flake check` (repository policy blocks direct `cargo test`; the `cli-tests` flake check runs the full workspace suite including the new reader tests) — all 3 checks (`cli-tests`, `cli-clippy`, `cli-fmt`) passed. + - Deviations: none. + +- [x] T05: `Add source-instance storage-resolver integration test and document the export reader boundary` (status:done) + - Task ID: T05 + - Goal: Add one integration-style test that resolves `ResolvedAgentTraceStorage` through the existing PR #197 storage resolver (`resolve_agent_trace_storage_at_state_root` or equivalent test entrypoint), asserts `storage.metadata.repository_id` and `storage.metadata.source_instance_id` remain available, constructs `AgentTraceExportReader::new(&storage.db)`, and successfully reads at least one stream — proving the `ResolvedAgentTraceStorage → metadata + db → AgentTraceExportReader` composition point without the reader generating or owning `source_instance_id`. Then document the new boundary: create or extend Agent Trace context documentation describing the layering (`SCE local source DB → incremental export reader → future control-plane client`), the fact that `repository_id`/`source_instance_id` identify the source while `table.id` is the per-stream progress marker, the exact `WHERE id > cursor ORDER BY id ASC LIMIT batch_size` query shape for all four streams, and an explicit statement that there is no local sync cursor, no `agent-trace-sync.db`, no Turso Sync, no ETL, and no DWH. Update `context/context-map.md` if a new domain file is added. + - Boundaries (in/out of scope): In — the one integration test, and durable-context documentation edits/additions. Out — any further reader behavior changes; any control-plane client code. + - Dependencies: T04 + - Done when: the integration test passes, durable context documents the export-reader boundary and the four exact stream queries, and `nix run .#pkl-check-generated` plus `nix flake check` both pass. + - Verification notes (commands or checks): `cargo test -p shared-context-engineering --lib services::agent_trace_export`; `nix run .#pkl-check-generated`; `nix flake check` + - Evidence: Added `source_instance_integration` to `cli/src/services/agent_trace_export/mod.rs`'s test module (plus local `unique_storage_temp_dir`/`init_git_repo_with_remote`/`git` helpers mirroring `agent_trace_storage`'s test pattern): resolves `ResolvedAgentTraceStorage` via `resolve_agent_trace_storage_at_state_root`, asserts `storage.metadata.repository_id` and `storage.metadata.source_instance_id` are non-empty, builds `AgentTraceExportReader::new(&storage.db)`, seeds one `messages` row via direct SQL, and reads it back through `read_messages_after`. Documented the export-reader boundary in a new `context/sce/agent-trace-export-readers.md` (layering diagram, storage-resolver composition point, identity-vs-progress-marker distinction, the four query shapes/validation rules, and the explicit no-local-cursor/no-`agent-trace-sync.db`/no-Turso-Sync/no-ETL/no-DWH statement), trimmed `context/sce/agent-trace-db.md`'s "Export reader" section to a pointer to keep it under the file-hygiene line budget, and added a `context/context-map.md` entry for the new domain file. + - Verification run: `nix flake check` — all 3 checks (`cli-tests`, `cli-clippy`, `cli-fmt`) passed (including the new `source_instance_integration` test); `nix run .#pkl-check-generated` — passed (101 files, ephemeral generation matched inventory). + - Deviations: none. + +## Open questions + +None. The change request is fully specified end-to-end (module boundary, DTO shapes, query semantics, validation rules, and test coverage), and it does not duplicate or extend PR #197's identity work — it consumes `RepositoryMetadata` as-is and adds a strictly read-only, additive seam. + +## Validation Report + +**Status:** validated +**Date:** 2026-08-10 + +### Commands run + +- `nix run .#pkl-check-generated` -> exit 0 (Ephemeral Pkl generation passed: 101 files, inventory sha256 766f5111af2434d6c345d07ae0aeb8b276aeeb94e7ecb7d39688c3f1267c8971) +- `nix flake check` -> exit 0 (all checks passed, including `cli-tests`, `cli-clippy`, `cli-fmt`; `cli-tests` derivation built and its store output verified present) + +### Scaffolding removed + +- None. + +### Success-criteria verification + +- [x] AC1: `read_messages_after` cursor/order/limit/`sourceRowId` contract -> `services::agent_trace_export::tests::read_messages_after_returns_rows_after_cursor_in_order`, `..._returns_non_contiguous_ids`, `..._limit_truncates_and_follow_up_continues`, `..._returns_empty_at_or_beyond_max_id` all `ok` in `cli-tests` +- [x] AC2: same contract for `read_parts_after`/`read_diff_traces_after`/`read_agent_traces_after`, nullable-as-null, no contiguity assumption -> corresponding `read_parts_after_*`, `read_diff_traces_after_*`, `read_agent_traces_after_*` tests (incl. `..._preserves_populated_nullable_fields`, `..._preserves_null_remote_url`) all `ok` +- [x] AC3: exact camelCase serialization contract -> `message_export_row_serializes_to_camel_case_contract`, `message_export_row_serializes_user_role_lowercase`, `part_export_row_serializes_to_camel_case_contract`, `diff_trace_export_row_serializes_with_all_fields_populated`, `diff_trace_export_row_serializes_nullable_fields_as_null`, `agent_trace_export_row_serializes_with_remote_url_populated`, `agent_trace_export_row_serializes_with_remote_url_null` all `ok` +- [x] AC4: cursor/limit/safe-integer rejection -> `validate_cursor_rejects_negative`, `validate_limit_rejects_zero`, `validate_limit_rejects_above_batch_size`, `validate_js_safe_integer_rejects_negative`, `validate_js_safe_integer_rejects_above_max_safe_integer`, plus per-stream `read_*_after_rejects_row_above_safe_integer_bound` / `read_*_after_rejects_invalid_cursor_and_limit` all `ok` +- [x] AC5: no database mutation -> inspection of `cli/src/services/agent_trace_export/mod.rs`: all four reader methods issue only `SELECT_MESSAGES_AFTER_SQL` / `SELECT_PARTS_AFTER_SQL` / `SELECT_DIFF_TRACES_AFTER_SQL` / `SELECT_AGENT_TRACES_AFTER_SQL` constants (no `INSERT`/`UPDATE`/DDL outside `#[cfg(test)]` seeding helpers); confirmed by `read_messages_after_performs_no_mutation`, `read_parts_after_performs_no_mutation`, `read_diff_traces_after_performs_no_mutation`, `read_agent_traces_after_performs_no_mutation` all `ok` +- [x] AC6: storage-resolver composition point -> `services::agent_trace_export::tests::source_instance_integration` `ok` + +### 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 d53ce6f4..5b03e939 100644 --- a/context/sce/agent-trace-db.md +++ b/context/sce/agent-trace-db.md @@ -213,6 +213,10 @@ Post-commit intersection rows are written by the active `post-commit` hook flow - 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. +## Export reader (read-only) + +`cli/src/services/agent_trace_export/mod.rs` defines `AgentTraceExportReader<'a>`, a read-only incremental reader over one repository-scoped `RepositoryAgentTraceDb` (`AgentTraceExportReader::new(&db)`), composing directly with `ResolvedAgentTraceStorage` from `agent_trace_storage` without generating or owning `source_instance_id` itself. See [agent-trace-export-readers.md](agent-trace-export-readers.md) for the full reader boundary, the storage-resolver composition point, the four stream query shapes, and the explicit no-local-cursor / no-sync-db / no-Turso-Sync / no-ETL / no-DWH statement. + ## Staged-diff AI-overlap evidence gate `cli/src/services/agent_trace.rs` owns the pure patch-overlap helper `patches_have_overlap`, which is consumed by the commit-msg staged-diff AI-overlap evidence gate in `cli/src/services/hooks/mod.rs`: diff --git a/context/sce/agent-trace-export-readers.md b/context/sce/agent-trace-export-readers.md new file mode 100644 index 00000000..63f59a68 --- /dev/null +++ b/context/sce/agent-trace-export-readers.md @@ -0,0 +1,64 @@ +# Agent Trace export readers (read-only) + +`cli/src/services/agent_trace_export/mod.rs` defines `AgentTraceExportReader<'a>`, the local read/export boundary between one repository-scoped Agent Trace source database and any future outbound sync. It is purely additive over the existing schema: it adds no table, no migration, and no writer. + +## Layering + +```mermaid +flowchart LR + A["SCE local source DB\n(RepositoryAgentTraceDb)"] --> B["Incremental export reader\n(AgentTraceExportReader)"] + B --> C["Future control-plane client\n(not built by this plan)"] +``` + +- **SCE local source DB** — the existing repository-scoped `RepositoryAgentTraceDb` (see [agent-trace-db.md](agent-trace-db.md)), written by the hook/lifecycle paths already documented there. This plan does not change its writer, schema, or migrations. +- **Incremental export reader** — `AgentTraceExportReader<'a>`, described below. Read-only, stateless across calls, no network. +- **Future control-plane client** — out of scope for this plan. A later plan composes this reader with an HTTP client and `sce trace sync` orchestration; nothing in this reader assumes or depends on that client existing. + +## Composition point + +`AgentTraceExportReader::new(&db)` takes a `&RepositoryAgentTraceDb` directly. The reader does not resolve storage, open a database, or generate identity itself. The existing storage resolver composes cleanly with it: + +```rust +let storage = resolve_agent_trace_storage_at_state_root(&context, &state_root)?; +let reader = AgentTraceExportReader::new(&storage.db); +let rows = reader.read_messages_after(cursor, limit)?; +``` + +`storage.metadata` (`RepositoryMetadata { repository_id, source_instance_id }`, see [agent-trace-db.md](agent-trace-db.md#repository-scoped-adapter-seam)) identifies *which* physical database produced the rows; the reader never reads, generates, or accepts a source identity of its own. Identity and progress tracking are deliberately separate concerns: + +- `repository_id` / `source_instance_id` identify the *source* (the physical database). +- Each stream's `table.id` (returned as `sourceRowId`) is the per-stream *progress marker* — the caller's cursor. It has no relationship to `source_instance_id` and is never used to derive or validate it. + +Test coverage: `cli/src/services/agent_trace_export/mod.rs::tests::source_instance_integration` resolves storage through `resolve_agent_trace_storage_at_state_root`, asserts both metadata fields are populated, and reads one stream through a reader built from `storage.db`. + +## Reader contract + +Four methods, one per capture stream, sharing one shape: `(cursor: i64, limit: usize) -> Result>`, running + +```sql +SELECT ... FROM WHERE id > ?1 ORDER BY id ASC LIMIT ?2 +``` + +against `messages`, `parts`, `diff_traces`, and `agent_traces` respectively (`read_messages_after`, `read_parts_after`, `read_diff_traces_after`, `read_agent_traces_after`). `cursor` is the last server-accepted `id` for that stream; the reader makes no gap or contiguity assumption about IDs. `diff_traces.patch` / `payload_type` and `agent_traces.trace_json` are returned raw and unmodified — no patch parsing, no JSON reparsing. + +Every call validates, before executing any query: + +- `cursor >= 0` +- `1 <= limit <= AGENT_TRACE_EXPORT_BATCH_SIZE` (500) + +and validates, per returned row before returning: + +- every exportable numeric field falls within `0..=9_007_199_254_740_991` (`Number.MAX_SAFE_INTEGER`), rejecting out-of-range rows instead of truncating or casting. + +Each stream has an owned `serde::Serialize` export-row DTO (`AgentTraceMessageExportRow`, `AgentTracePartExportRow`, `AgentTraceDiffTraceExportRow`, `AgentTraceAgentTraceExportRow`) with `#[serde(rename_all = "camelCase")]` matching the shipped control-plane ingestion contract; `sourceRowId` is the local `id` unmodified. `post_commit_patch_intersections` is not exported by any reader method. + +## What does not exist + +This reader introduces no local sync state and no outbound transport: + +- No local sync cursor is stored anywhere; the caller (a future `sce trace sync`) owns cursor persistence entirely outside this module. +- No `agent-trace-sync.db` or any other new database or table exists. +- No Turso Sync, no ETL pipeline, and no data-warehouse (DWH) integration exists. +- No network call, no HTTP client, and no auth/WorkOS code exists in this module. + +See also: [agent-trace-db.md](agent-trace-db.md), [context-map.md](../context-map.md)