From 2ca1be958eb1ffed6288f47fe56821d970e6791c Mon Sep 17 00:00:00 2001 From: samzong Date: Wed, 7 Oct 2026 07:27:48 -0400 Subject: [PATCH 1/6] feat(adapters): add Warp agent conversation adapter Decode persisted task protobufs instead of synthesizing assistant messages and tool calls from query history or aggregate statistics. Verified with make check and isolated CLI sync, export, search, and event backfill against synthetic fixtures encoded with the official Warp protocol. Real Warp conversation acceptance remains pending because the installed app did not initialize. Signed-off-by: samzong --- Cargo.lock | 24 ++ Cargo.toml | 1 + src/adapters/mod.rs | 3 + src/adapters/warp.rs | 589 ++++++++++++++++++++++++++ src/adapters/warp/proto.rs | 369 ++++++++++++++++ src/adapters/warp/tests.rs | 483 +++++++++++++++++++++ src/usage.rs | 19 + tests/fixtures/warp/README.md | 36 ++ tests/fixtures/warp/origins.bin | Bin 0 -> 501 bytes tests/fixtures/warp/origins.textproto | 12 + tests/fixtures/warp/schema.sql | 15 + tests/fixtures/warp/task.bin | Bin 0 -> 853 bytes tests/fixtures/warp/task.textproto | 182 ++++++++ 13 files changed, 1733 insertions(+) create mode 100644 src/adapters/warp.rs create mode 100644 src/adapters/warp/proto.rs create mode 100644 src/adapters/warp/tests.rs create mode 100644 tests/fixtures/warp/README.md create mode 100644 tests/fixtures/warp/origins.bin create mode 100644 tests/fixtures/warp/origins.textproto create mode 100644 tests/fixtures/warp/schema.sql create mode 100644 tests/fixtures/warp/task.bin create mode 100644 tests/fixtures/warp/task.textproto diff --git a/Cargo.lock b/Cargo.lock index 6a32820e..320c2891 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3275,6 +3275,29 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools 0.14.0", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "pulldown-cmark" version = "0.13.4" @@ -3477,6 +3500,7 @@ dependencies = [ "globset", "hf-hub", "kitup", + "prost", "pulldown-cmark", "ratatui", "rmcp", diff --git a/Cargo.toml b/Cargo.toml index 0dfe7168..799637d9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,6 +55,7 @@ fs2 = "0.4" pulldown-cmark = "0.13" zip = { version = "7", default-features = false, features = ["deflate"] } zstd = { version = "0.13", default-features = false } +prost = { version = "0.14", default-features = false, features = ["derive", "std"] } [target.'cfg(target_os = "macos")'.dependencies] candle-core = { version = "0.10", features = ["metal"] } diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs index b14bf37a..2170db5a 100644 --- a/src/adapters/mod.rs +++ b/src/adapters/mod.rs @@ -35,6 +35,7 @@ pub(crate) mod roo; pub(crate) mod sync_state; pub(crate) mod trae_cli; pub(crate) mod usage; +pub(crate) mod warp; pub(crate) mod zcode; use std::cell::RefCell; @@ -447,6 +448,7 @@ pub(crate) fn all_adapters() -> Vec> { Box::new(muse_code::MuseCodeAdapter), Box::new(codebuddy::CodeBuddyAdapter), Box::new(trae_cli::TraeCliAdapter), + Box::new(warp::WarpAdapter), ] } @@ -494,6 +496,7 @@ pub(crate) fn source_supports_event_backfill(source_id: &str) -> bool { | "muse-code" | "codebuddy" | "trae-cli" + | "warp" ) } diff --git a/src/adapters/warp.rs b/src/adapters/warp.rs new file mode 100644 index 00000000..1934791e --- /dev/null +++ b/src/adapters/warp.rs @@ -0,0 +1,589 @@ +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use chrono::{DateTime, NaiveDateTime}; +use prost::Message as _; +use rusqlite::{Connection, OpenFlags}; +use serde_json::{Value, json}; +use tracing::warn; + +use crate::adapters::events::{self, EventContext}; +use crate::adapters::{ + AdapterSyncContext, RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanOutput, + SyncScanResult, +}; +use crate::types::{ + FileEvidence, FileOperation, ParentLink, ParentRelation, RawSessionEvent, RawUsageEvent, Role, + ThreadRole, +}; + +mod proto; + +#[cfg(test)] +mod tests; + +const METADATA_PARSER_VERSION: u32 = 1; +const EVENT_PARSER_VERSION: u32 = 2; +const USAGE_PARSER_VERSION: u32 = 1; + +pub(crate) struct WarpAdapter; + +impl SourceAdapter for WarpAdapter { + fn id(&self) -> &str { + "warp" + } + + fn label(&self) -> &str { + "WP" + } + + fn usage_parser_version(&self) -> Option { + Some(USAGE_PARSER_VERSION) + } + + fn resume_command(&self, _source_id: &str) -> Option { + None + } + + fn scan(&self) -> Result> { + Ok(scan_paths(&default_db_paths(), None, true)?.sessions) + } + + fn scan_for_sync_output( + &self, + context: &AdapterSyncContext, + _since_ts: Option, + include_events: bool, + _force: bool, + ) -> Result> { + Ok(Some(SyncScanOutput { + scan: scan_paths(&default_db_paths(), Some(context), include_events)?, + reconcile: None, + })) + } +} + +fn default_db_paths() -> Vec { + let Some(home) = dirs::home_dir() else { + return Vec::new(); + }; + let dirs = if cfg!(target_os = "macos") { + ["Stable", "Preview"] + .into_iter() + .flat_map(|channel| { + let app = format!("dev.warp.Warp-{channel}"); + [ + home.join( + "Library/Group Containers/2BBY89MBSN.dev.warp/Library/Application Support", + ) + .join(&app), + home.join("Library/Application Support").join(app), + ] + }) + .collect::>() + } else if cfg!(target_os = "windows") { + let base = std::env::var_os("LOCALAPPDATA") + .map(PathBuf::from) + .unwrap_or_else(|| home.join("AppData/Local")); + ["Warp", "WarpPreview"].map(|app| base.join("warp").join(app).join("data")).to_vec() + } else { + let base = std::env::var_os("XDG_STATE_HOME") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + .unwrap_or_else(|| home.join(".local/state")); + ["warp-terminal", "warp-terminal-preview"].map(|app| base.join(app)).to_vec() + }; + dirs.into_iter() + .flat_map(|dir| [dir.join("warp.sqlite"), dir.join("tui/warp.sqlite")]) + .collect() +} + +fn scan_paths( + paths: &[PathBuf], + context: Option<&AdapterSyncContext>, + include_events: bool, +) -> Result { + let mut result = SyncScanResult::default(); + let mut seen = HashSet::new(); + for path in paths { + let mut scan = SyncScanResult::default(); + let outcome = scan_db_with_seen(Some(path), context, include_events, &mut seen, &mut scan); + result.absorb(scan); + if let Err(error) = outcome { + if error.is::() { + return Err(error); + } + warn!("failed to scan Warp database {}: {error:#}", path.display()); + } + } + Ok(result) +} + +#[derive(Debug)] +struct SessionWriteError; + +impl std::fmt::Display for SessionWriteError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("writing Warp session") + } +} + +impl std::error::Error for SessionWriteError {} + +fn has_table(conn: &Connection, name: &str) -> Result { + Ok(conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)", + [name], + |row| row.get(0), + )?) +} + +#[cfg(test)] +fn scan_db( + path: Option<&Path>, + context: Option<&AdapterSyncContext>, + include_events: bool, +) -> Result { + let mut result = SyncScanResult::default(); + scan_db_with_seen(path, context, include_events, &mut HashSet::new(), &mut result)?; + Ok(result) +} + +fn scan_db_with_seen( + path: Option<&Path>, + context: Option<&AdapterSyncContext>, + include_events: bool, + seen: &mut HashSet, + result: &mut SyncScanResult, +) -> Result<()> { + let Some(path) = path.filter(|path| path.is_file()) else { + return Ok(()); + }; + let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY) + .context("opening Warp database read-only")?; + conn.busy_timeout(Duration::from_secs(3))?; + let snapshot = conn.unchecked_transaction()?; + if !has_table(&snapshot, "agent_conversations")? { + return Ok(()); + } + let has_tasks = has_table(&snapshot, "agent_tasks")?; + let has_queries = has_table(&snapshot, "ai_queries")?; + let mut stmt = snapshot.prepare( + "SELECT conversation_id, last_modified_at, conversation_data FROM agent_conversations + WHERE (?1 IS NULL OR conversation_id = ?1) ORDER BY conversation_id", + )?; + let mut rows = stmt.query([context.and_then(AdapterSyncContext::target_source_id)])?; + while let Some(row) = rows.next()? { + result.stats.candidates += 1; + let parsed = (|| { + let id: String = row.get(0)?; + let modified: String = row.get(1)?; + parse_conversation( + &snapshot, + &id, + &modified, + path, + has_tasks, + has_queries, + include_events, + ) + })(); + match parsed { + Ok(Some(mut raw)) => { + let data: String = row.get(2)?; + match serde_json::from_str::(&data) { + Ok(data) => { + if let Some(parent) = data + .get("parent_conversation_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|parent| !parent.is_empty()) + { + raw.thread_role = Some(ThreadRole::Subagent); + raw.parent_links = vec![ParentLink { + relation: ParentRelation::Spawn, + source: "warp".into(), + source_id: parent.into(), + }]; + } + } + Err(error) => warn!("failed to parse Warp conversation metadata: {error}"), + } + raw.metadata_parser_version = Some(METADATA_PARSER_VERSION); + if !seen.insert(raw.source_id.clone()) { + result.stats.filtered_sessions += 1; + continue; + } + result.stats.parsed += 1; + if let Some(raw) = match context { + Some(context) => context.stream_session(raw).context(SessionWriteError)?, + None => Some(raw), + } { + result.sessions.push(raw); + } + } + Ok(None) => result.stats.filtered_sessions += 1, + Err(error) => warn!("failed to parse Warp conversation: {error:#}"), + } + } + Ok(()) +} + +fn parse_timestamp(text: &str) -> Option { + DateTime::parse_from_rfc3339(text.trim()).map(|time| time.timestamp_millis()).ok().or_else( + || { + NaiveDateTime::parse_from_str(text.trim(), "%Y-%m-%d %H:%M:%S%.f") + .ok() + .map(|time| time.and_utc().timestamp_millis()) + }, + ) +} + +impl proto::Timestamp { + fn millis(&self) -> Option { + if !(0..1_000_000_000).contains(&self.nanos) { + return None; + } + self.seconds.checked_mul(1000)?.checked_add(i64::from(self.nanos) / 1_000_000) + } +} + +fn cwd(context: Option<&proto::InputContext>) -> Option { + context + .and_then(|context| context.directory.as_ref()) + .map(|directory| directory.pwd.clone()) + .filter(|pwd| !pwd.trim().is_empty()) +} + +fn parse_conversation( + conn: &Connection, + id: &str, + modified: &str, + path: &Path, + has_tasks: bool, + has_queries: bool, + include_events: bool, +) -> Result> { + let mut updated_at = parse_timestamp(modified); + let mut records = Vec::new(); + let mut title = None; + if has_tasks { + let mut stmt = conn.prepare( + "SELECT task, last_modified_at FROM agent_tasks WHERE conversation_id = ?1 ORDER BY id", + )?; + let mut rows = stmt.query([id])?; + while let Some(row) = rows.next()? { + let bytes: Vec = row.get(0)?; + let task = + proto::Task::decode(bytes.as_slice()).context("decoding Warp task protobuf")?; + let modified: String = row.get(1)?; + updated_at = updated_at.max(parse_timestamp(&modified)); + if task.dependencies.as_ref().is_none_or(|deps| deps.parent_task_id.is_empty()) + && !task.description.trim().is_empty() + && title.is_none() + { + title = Some(task.description); + } + let mut ordering_time = task + .messages + .iter() + .find_map(|message| message.timestamp.as_ref().and_then(proto::Timestamp::millis)); + for (seq, message) in task.messages.into_iter().enumerate() { + let key = if message.id.is_empty() { + format!("{}:{seq}", task.id) + } else { + message.id.clone() + }; + ordering_time = + message.timestamp.as_ref().and_then(proto::Timestamp::millis).or(ordering_time); + records.push((ordering_time, key, task.id.clone(), message)); + } + } + } + records.sort_by_key(|(time, _, _, _)| *time); + let mut seen = HashSet::new(); + records.retain(|(_, key, _, _)| seen.insert(key.clone())); + let mut messages = Vec::new(); + let mut session_events = Vec::new(); + let mut usage_events = Vec::new(); + let mut directory = None; + let mut task_directories = HashMap::new(); + let mut tool_names = HashMap::new(); + let shell_commands: HashMap<_, _> = records + .iter() + .filter_map(|(_, _, _, message)| message.tool_call.as_ref()) + .filter(|call| !call.tool_call_id.is_empty()) + .filter_map(|call| { + call.run_shell_command + .as_ref() + .map(|shell| (call.tool_call_id.clone(), shell.command.clone())) + }) + .collect(); + let mut started_at = None; + let source_path = path.to_string_lossy().into_owned(); + for (_, key, task_id, message) in records { + let timestamp = message.timestamp.as_ref().and_then(proto::Timestamp::millis); + started_at = started_at.or(timestamp); + updated_at = updated_at.max(timestamp); + let context = + message.user_query.as_ref().and_then(|query| query.context.as_ref()).or_else(|| { + message.tool_call_result.as_ref().and_then(|result| result.context.as_ref()) + }); + if let Some(pwd) = cwd(context) { + directory = directory.or_else(|| Some(pwd.clone())); + task_directories.insert(task_id.clone(), pwd); + } + let active_cwd = task_directories.get(&task_id).map(String::as_str); + if let Some(query) = message.user_query { + if !query.origin.as_ref().is_some_and(|origin| { + matches!( + origin.variant, + Some( + proto::QueryOrigin::ParentAgent(_) + | proto::QueryOrigin::AgentMessageWake(_) + | proto::QueryOrigin::Schedule(_) + | proto::QueryOrigin::Automation(_) + | proto::QueryOrigin::ServerSynthesized(_) + ) + ) + }) && !query.query.trim().is_empty() + { + messages.push(RawMessage { role: Role::User, content: query.query, timestamp }); + } + } else if let Some(output) = message.agent_output { + if !output.text.trim().is_empty() { + messages.push(RawMessage { + role: Role::Assistant, + content: output.text, + timestamp, + }); + } + } else if include_events { + let context = EventContext { + event_seq: session_events.len() as u32, + timestamp, + source_path: Some(source_path.clone()), + source_event_id: Some(key.clone()), + message_seq: None, + parser_version: EVENT_PARSER_VERSION, + }; + if let Some(call) = message.tool_call { + let event = call_event(&call, context, active_cwd); + tool_names.insert(call.tool_call_id, event.name.clone()); + session_events.push(event); + } else if let Some(result) = message.tool_call_result { + let name = tool_names.get(&result.tool_call_id).cloned().flatten(); + let command = shell_commands.get(&result.tool_call_id).map(String::as_str); + session_events.push(result_event(&result, context, name, command, active_cwd)); + } + } + if let Some(metadata) = message.request_metadata + && let Some(timestamp) = timestamp + { + parse_usage(&metadata, &key, timestamp, &source_path, &mut usage_events)?; + } + } + if !has_tasks && has_queries { + let mut stmt = conn.prepare( + "SELECT input, working_directory, start_ts FROM ai_queries + WHERE conversation_id = ?1 ORDER BY start_ts, id", + )?; + let mut rows = stmt.query([id])?; + while let Some(row) = rows.next()? { + let input: Option = row.get(0)?; + let pwd: Option = row.get(1)?; + let time: String = row.get(2)?; + let timestamp = parse_timestamp(&time); + started_at = started_at.or(timestamp); + updated_at = updated_at.max(timestamp); + directory = directory.or(pwd.filter(|pwd| !pwd.trim().is_empty())); + if let Some(input) = input { + let input: Vec = serde_json::from_str(&input)?; + for item in input { + if let Some(text) = item.pointer("/Query/text").and_then(Value::as_str) + && !text.trim().is_empty() + { + messages.push(RawMessage { + role: Role::User, + content: text.into(), + timestamp, + }); + } + } + } + } + } + if messages.is_empty() && session_events.is_empty() && usage_events.is_empty() { + return Ok(None); + } + let started_at = + started_at.or(updated_at).context("Warp conversation has no valid timestamp")?; + let mut raw = RawSession::search_only(id, directory, started_at, updated_at, None, messages) + .with_usage(usage_events, USAGE_PARSER_VERSION); + if include_events { + raw = raw.with_events(session_events, EVENT_PARSER_VERSION); + } + raw.source_file_path = Some(source_path); + raw.custom_title = title; + Ok(Some(raw)) +} + +fn call_event(call: &proto::ToolCall, context: EventContext, cwd: Option<&str>) -> RawSessionEvent { + let mut files = Vec::new(); + let (name, args) = if let Some(shell) = &call.run_shell_command { + ("run_shell_command", Some(json!({"command": shell.command}))) + } else if let Some(read) = &call.read_files { + files.extend(read.files.iter().filter(|file| !file.name.trim().is_empty()).map(|file| { + FileEvidence::call(file.name.clone(), FileOperation::Read, cwd.map(str::to_string)) + })); + ( + "read_files", + Some(json!({"paths": read.files.iter().map(|file| &file.name).collect::>()})), + ) + } else if let Some(diff) = &call.apply_file_diffs { + for (paths, operation) in [ + (&diff.diffs, FileOperation::Write), + (&diff.new_files, FileOperation::Write), + (&diff.deleted_files, FileOperation::Delete), + ] { + for file in paths.iter().filter(|file| !file.file_path.trim().is_empty()) { + files.push(FileEvidence::call( + file.file_path.clone(), + operation.clone(), + cwd.map(str::to_string), + )); + } + } + for update in &diff.v4a_updates { + if update.file_path.trim().is_empty() { + continue; + } + let moved = !update.move_to.trim().is_empty(); + files.push(FileEvidence::call( + update.file_path.clone(), + if moved { FileOperation::MoveFrom } else { FileOperation::Write }, + cwd.map(str::to_string), + )); + if moved { + files.push(FileEvidence::call( + update.move_to.clone(), + FileOperation::MoveTo, + cwd.map(str::to_string), + )); + } + } + ( + "apply_file_diffs", + Some( + json!({"summary": diff.summary, "paths": files.iter().map(|file| &file.path).collect::>()}), + ), + ) + } else { + (call.other.as_ref().map(proto::OtherTool::name).unwrap_or("unknown"), None) + }; + let mut event = events::tool_call_event(context, name.into(), args.as_ref()); + event.tool_call_id = Some(call.tool_call_id.clone()).filter(|id| !id.is_empty()); + if name == "run_shell_command" { + let (evidence, status) = + events::shell_file_evidence(&call.run_shell_command.as_ref().unwrap().command, cwd); + files = evidence; + event.command_evidence_status = Some(status); + } else if name == "read_files" || name == "apply_file_diffs" { + event.kind = if name == "read_files" { "file_read" } else { "file_write" }.into(); + event.target = files.first().map(|file| file.path.clone()); + } + event.files = files; + event +} + +fn result_event( + result: &proto::ToolCallResult, + context: EventContext, + name: Option, + call_command: Option<&str>, + cwd: Option<&str>, +) -> RawSessionEvent { + let mut event = events::tool_result_event(context, name, None); + event.tool_call_id = Some(result.tool_call_id.clone()).filter(|id| !id.is_empty()); + if let Some(shell) = &result.run_shell_command { + event.name = Some("run_shell_command".into()); + let exit_code = + shell.command_finished.as_ref().map(|finished| finished.exit_code).or(shell.exit_code); + let output = shell + .command_finished + .as_ref() + .map(|finished| &finished.output) + .unwrap_or(&shell.output); + event.summary = + Some(events::bounded_summary(output.clone())).filter(|text| !text.is_empty()); + event.status = if shell.permission_denied.is_some() { + Some("denied".into()) + } else if shell.terminal_busy.is_some() { + Some("terminal_busy".into()) + } else { + exit_code.map(|code| if code == 0 { "success" } else { "error" }.into()) + }; + if !shell.command.trim().is_empty() { + event.target = Some(shell.command.clone()); + if call_command != Some(shell.command.as_str()) { + let (files, status) = events::shell_file_evidence(&shell.command, cwd); + event.files = files; + event.command_evidence_status = Some(status); + } + } + event.attrs_json = Some(json!({"exit_code": exit_code}).to_string()); + } else if result.cancel.is_some() { + event.status = Some("cancelled".into()); + } else if let Some(outcome) = result.read_files.as_ref().or(result.apply_file_diffs.as_ref()) { + event.status = if outcome.error.is_some() { + Some("error".into()) + } else if outcome.success.is_some() || outcome.any_files_success.is_some() { + Some("success".into()) + } else { + None + }; + } + event +} + +fn parse_usage( + metadata: &proto::RequestMetadata, + message_id: &str, + timestamp: i64, + source_path: &str, + events: &mut Vec, +) -> Result<()> { + let Some(charges) = &metadata.charges else { + return Ok(()); + }; + for (category, usage) in &charges.usage_by_category { + for (origin, models) in [ + ("warp", &usage.direct_api_inference_usage), + ("byok", &usage.byok_inference_usage), + ("custom_endpoint", &usage.custom_endpoint_inference_usage), + ] { + for (model, usage) in models { + let Some(tokens) = &usage.token_count else { + continue; + }; + let mut event = RawUsageEvent::observed( + json!([message_id, category, origin, model]).to_string(), + events.len() as u32, + timestamp, + USAGE_PARSER_VERSION, + ); + event.model = model.clone(); + event.input_tokens = tokens.input.try_into()?; + event.output_tokens = tokens.output.try_into()?; + event.cache_read_tokens = tokens.input_cache_read.try_into()?; + event.cache_write_tokens = tokens.input_cache_write.try_into()?; + event.source_path = Some(source_path.into()); + event.raw_usage_json = Some(json!({"category": category, "origin": origin, "input": tokens.input, "output": tokens.output, "input_cache_read": tokens.input_cache_read, "input_cache_write": tokens.input_cache_write}).to_string()); + events.push(event); + } + } + } + Ok(()) +} diff --git a/src/adapters/warp/proto.rs b/src/adapters/warp/proto.rs new file mode 100644 index 00000000..b22eb5b3 --- /dev/null +++ b/src/adapters/warp/proto.rs @@ -0,0 +1,369 @@ +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Task { + #[prost(string, tag = "1")] + pub(super) id: String, + #[prost(string, tag = "2")] + pub(super) description: String, + #[prost(message, optional, tag = "3")] + pub(super) dependencies: Option, + #[prost(message, repeated, tag = "5")] + pub(super) messages: Vec, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Dependencies { + #[prost(string, tag = "1")] + pub(super) parent_task_id: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Message { + #[prost(string, tag = "1")] + pub(super) id: String, + #[prost(message, optional, tag = "2")] + pub(super) user_query: Option, + #[prost(message, optional, tag = "3")] + pub(super) agent_output: Option, + #[prost(message, optional, tag = "4")] + pub(super) tool_call: Option, + #[prost(message, optional, tag = "5")] + pub(super) tool_call_result: Option, + #[prost(string, tag = "11")] + pub(super) task_id: String, + #[prost(string, tag = "13")] + pub(super) request_id: String, + #[prost(message, optional, tag = "14")] + pub(super) timestamp: Option, + #[prost(message, optional, tag = "30")] + pub(super) request_metadata: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Timestamp { + #[prost(int64, tag = "1")] + pub(super) seconds: i64, + #[prost(int32, tag = "2")] + pub(super) nanos: i32, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct UserQuery { + #[prost(string, tag = "1")] + pub(super) query: String, + #[prost(message, optional, tag = "2")] + pub(super) context: Option, + #[prost(message, optional, tag = "6")] + pub(super) origin: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct UserQueryOrigin { + #[prost(oneof = "QueryOrigin", tags = "1, 2, 3, 4, 5, 6, 7, 8")] + pub(super) variant: Option, +} + +#[derive(Clone, PartialEq, prost::Oneof)] +pub(super) enum QueryOrigin { + #[prost(bytes, tag = "1")] + WarpClient(Vec), + #[prost(bytes, tag = "2")] + ExternalPlatform(Vec), + #[prost(bytes, tag = "3")] + ParentAgent(Vec), + #[prost(bytes, tag = "4")] + AgentMessageWake(Vec), + #[prost(bytes, tag = "5")] + Schedule(Vec), + #[prost(bytes, tag = "6")] + Automation(Vec), + #[prost(bytes, tag = "7")] + PublicApi(Vec), + #[prost(bytes, tag = "8")] + ServerSynthesized(Vec), +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct AgentOutput { + #[prost(string, tag = "1")] + pub(super) text: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct InputContext { + #[prost(message, optional, tag = "1")] + pub(super) directory: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Directory { + #[prost(string, tag = "1")] + pub(super) pwd: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ToolCall { + #[prost(string, tag = "1")] + pub(super) tool_call_id: String, + #[prost(message, optional, tag = "2")] + pub(super) run_shell_command: Option, + #[prost(message, optional, tag = "5")] + pub(super) read_files: Option, + #[prost(message, optional, tag = "6")] + pub(super) apply_file_diffs: Option, + #[prost( + oneof = "OtherTool", + tags = "3, 4, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 30, 31, 32, 34, 35, 36, 37, 38" + )] + pub(super) other: Option, +} + +#[derive(Clone, PartialEq, prost::Oneof)] +pub(super) enum OtherTool { + #[prost(bytes, tag = "3")] + SearchCodebase(Vec), + #[prost(bytes, tag = "4")] + Server(Vec), + #[prost(bytes, tag = "7")] + SuggestPlan(Vec), + #[prost(bytes, tag = "8")] + SuggestCreatePlan(Vec), + #[prost(bytes, tag = "9")] + Grep(Vec), + #[prost(bytes, tag = "10")] + FileGlob(Vec), + #[prost(bytes, tag = "11")] + ReadMcpResource(Vec), + #[prost(bytes, tag = "12")] + CallMcpTool(Vec), + #[prost(bytes, tag = "13")] + WriteToLongRunningShellCommand(Vec), + #[prost(bytes, tag = "14")] + SuggestNewConversation(Vec), + #[prost(bytes, tag = "15")] + FileGlobV2(Vec), + #[prost(bytes, tag = "16")] + SuggestPrompt(Vec), + #[prost(bytes, tag = "17")] + OpenCodeReview(Vec), + #[prost(bytes, tag = "18")] + InitProject(Vec), + #[prost(bytes, tag = "19")] + Subagent(Vec), + #[prost(bytes, tag = "20")] + ReadDocuments(Vec), + #[prost(bytes, tag = "21")] + EditDocuments(Vec), + #[prost(bytes, tag = "22")] + CreateDocuments(Vec), + #[prost(bytes, tag = "23")] + ReadShellCommandOutput(Vec), + #[prost(bytes, tag = "24")] + UseComputer(Vec), + #[prost(bytes, tag = "25")] + InsertReviewComments(Vec), + #[prost(bytes, tag = "26")] + ReadSkill(Vec), + #[prost(bytes, tag = "27")] + RequestComputerUse(Vec), + #[prost(bytes, tag = "28")] + FetchConversation(Vec), + #[prost(bytes, tag = "30")] + SendMessageToAgent(Vec), + #[prost(bytes, tag = "31")] + TransferShellCommandControlToUser(Vec), + #[prost(bytes, tag = "32")] + AskUserQuestion(Vec), + #[prost(bytes, tag = "34")] + UploadFileArtifact(Vec), + #[prost(bytes, tag = "35")] + RunAgents(Vec), + #[prost(bytes, tag = "36")] + WaitForEvents(Vec), + #[prost(bytes, tag = "37")] + StartRecording(Vec), + #[prost(bytes, tag = "38")] + StopRecording(Vec), +} + +impl OtherTool { + pub(super) fn name(&self) -> &str { + match self { + Self::SearchCodebase(_) => "search_codebase", + Self::Server(_) => "server", + Self::SuggestPlan(_) => "suggest_plan", + Self::SuggestCreatePlan(_) => "suggest_create_plan", + Self::Grep(_) => "grep", + Self::FileGlob(_) => "file_glob", + Self::ReadMcpResource(_) => "read_mcp_resource", + Self::CallMcpTool(_) => "call_mcp_tool", + Self::WriteToLongRunningShellCommand(_) => "write_to_long_running_shell_command", + Self::SuggestNewConversation(_) => "suggest_new_conversation", + Self::FileGlobV2(_) => "file_glob_v2", + Self::SuggestPrompt(_) => "suggest_prompt", + Self::OpenCodeReview(_) => "open_code_review", + Self::InitProject(_) => "init_project", + Self::Subagent(_) => "subagent", + Self::ReadDocuments(_) => "read_documents", + Self::EditDocuments(_) => "edit_documents", + Self::CreateDocuments(_) => "create_documents", + Self::ReadShellCommandOutput(_) => "read_shell_command_output", + Self::UseComputer(_) => "use_computer", + Self::InsertReviewComments(_) => "insert_review_comments", + Self::ReadSkill(_) => "read_skill", + Self::RequestComputerUse(_) => "request_computer_use", + Self::FetchConversation(_) => "fetch_conversation", + Self::SendMessageToAgent(_) => "send_message_to_agent", + Self::TransferShellCommandControlToUser(_) => "transfer_shell_command_control_to_user", + Self::AskUserQuestion(_) => "ask_user_question", + Self::UploadFileArtifact(_) => "upload_file_artifact", + Self::RunAgents(_) => "run_agents", + Self::WaitForEvents(_) => "wait_for_events", + Self::StartRecording(_) => "start_recording", + Self::StopRecording(_) => "stop_recording", + } + } +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct RunShellCommand { + #[prost(string, tag = "1")] + pub(super) command: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ReadFiles { + #[prost(message, repeated, tag = "1")] + pub(super) files: Vec, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct FileName { + #[prost(string, tag = "1")] + pub(super) name: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ApplyFileDiffs { + #[prost(string, tag = "1")] + pub(super) summary: String, + #[prost(message, repeated, tag = "2")] + pub(super) diffs: Vec, + #[prost(message, repeated, tag = "3")] + pub(super) new_files: Vec, + #[prost(message, repeated, tag = "4")] + pub(super) deleted_files: Vec, + #[prost(message, repeated, tag = "5")] + pub(super) v4a_updates: Vec, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct FilePath { + #[prost(string, tag = "1")] + pub(super) file_path: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct FileUpdate { + #[prost(string, tag = "1")] + pub(super) file_path: String, + #[prost(string, tag = "2")] + pub(super) move_to: String, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ToolCallResult { + #[prost(string, tag = "1")] + pub(super) tool_call_id: String, + #[prost(message, optional, tag = "2")] + pub(super) run_shell_command: Option, + #[prost(message, optional, tag = "5")] + pub(super) read_files: Option, + #[prost(message, optional, tag = "6")] + pub(super) apply_file_diffs: Option, + #[prost(message, optional, tag = "11")] + pub(super) context: Option, + #[prost(bytes = "vec", optional, tag = "14")] + pub(super) cancel: Option>, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct Outcome { + #[prost(bytes = "vec", optional, tag = "1")] + pub(super) success: Option>, + #[prost(bytes = "vec", optional, tag = "2")] + pub(super) error: Option>, + #[prost(bytes = "vec", optional, tag = "3")] + pub(super) any_files_success: Option>, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ShellResult { + #[prost(string, tag = "1")] + pub(super) output: String, + #[prost(int32, optional, tag = "2")] + pub(super) exit_code: Option, + #[prost(string, tag = "3")] + pub(super) command: String, + #[prost(bytes = "vec", optional, tag = "4")] + pub(super) long_running: Option>, + #[prost(message, optional, tag = "5")] + pub(super) command_finished: Option, + #[prost(bytes = "vec", optional, tag = "6")] + pub(super) permission_denied: Option>, + #[prost(bytes = "vec", optional, tag = "7")] + pub(super) terminal_busy: Option>, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ShellFinished { + #[prost(string, tag = "1")] + pub(super) output: String, + #[prost(int32, tag = "2")] + pub(super) exit_code: i32, + #[prost(string, tag = "3")] + pub(super) command_id: String, + #[prost(message, optional, tag = "4")] + pub(super) start_ts: Option, + #[prost(message, optional, tag = "5")] + pub(super) finish_ts: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct RequestMetadata { + #[prost(message, optional, tag = "2")] + pub(super) charges: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct RequestCharges { + #[prost(btree_map = "string, message", tag = "1")] + pub(super) usage_by_category: std::collections::BTreeMap, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct ChargedUsage { + #[prost(btree_map = "string, message", tag = "1")] + pub(super) direct_api_inference_usage: std::collections::BTreeMap, + #[prost(btree_map = "string, message", tag = "2")] + pub(super) byok_inference_usage: std::collections::BTreeMap, + #[prost(btree_map = "string, message", tag = "3")] + pub(super) custom_endpoint_inference_usage: std::collections::BTreeMap, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct InferenceUsage { + #[prost(message, optional, tag = "1")] + pub(super) token_count: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +pub(super) struct TokenCount { + #[prost(uint64, tag = "1")] + pub(super) input: u64, + #[prost(uint64, tag = "2")] + pub(super) output: u64, + #[prost(uint64, tag = "3")] + pub(super) input_cache_read: u64, + #[prost(uint64, tag = "4")] + pub(super) input_cache_write: u64, +} diff --git a/src/adapters/warp/tests.rs b/src/adapters/warp/tests.rs new file mode 100644 index 00000000..2a383ba0 --- /dev/null +++ b/src/adapters/warp/tests.rs @@ -0,0 +1,483 @@ +use super::*; +use rusqlite::params; + +const TASK: &[u8] = include_bytes!("../../../tests/fixtures/warp/task.bin"); +const SCHEMA: &str = include_str!("../../../tests/fixtures/warp/schema.sql"); + +fn database() -> (tempfile::TempDir, PathBuf, Connection) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("warp.sqlite"); + let conn = Connection::open(&path).unwrap(); + conn.execute_batch(SCHEMA).unwrap(); + conn.execute( + "INSERT INTO agent_conversations (conversation_id, conversation_data, last_modified_at) + VALUES ('conversation-1', ?1, '2026-09-30 16:00:09.250')", + [r#"{"conversation_usage_metadata":{"token_usage":[{"warp_tokens":9999}],"tool_usage_metadata":{"run_command_stats":{"count":100}}}}"#], + ) + .unwrap(); + conn.execute( + "INSERT INTO agent_tasks (conversation_id, task_id, task, last_modified_at) + VALUES ('conversation-1', 'task-root', ?1, '2026-09-30 16:00:09.250')", + [TASK], + ) + .unwrap(); + (dir, path, conn) +} + +fn write_task(conn: &Connection, task: &proto::Task) { + conn.execute( + "UPDATE agent_tasks SET task = ?1 WHERE task_id = 'task-root'", + [task.encode_to_vec()], + ) + .unwrap(); +} + +#[test] +fn official_wire_fixture_yields_messages_tools_and_observed_usage() { + let (_dir, path, conn) = database(); + let before = std::fs::read(&path).unwrap(); + let scan = scan_db(Some(&path), None, true).unwrap(); + assert_eq!(scan.sessions.len(), 1); + assert_eq!(scan.stats.candidates, 1); + assert_eq!(scan.stats.parsed, 1); + let raw = &scan.sessions[0]; + assert_eq!(raw.source_id, "conversation-1"); + assert_eq!(raw.custom_title.as_deref(), Some("Warp fixture task")); + assert_eq!(raw.directory.as_deref(), Some("/tmp/recall-warp-project")); + assert_eq!(raw.started_at, 1790784000125); + assert_eq!(raw.updated_at, Some(1790784009250)); + assert_eq!(raw.source_file_path.as_deref(), path.to_str()); + assert_eq!(raw.messages.len(), 3); + assert_eq!(raw.messages[0].role, Role::User); + assert_eq!(raw.messages[1].role, Role::Assistant); + assert_eq!( + raw.messages[2].content, + "Read the file, counted two lines, and updated output.txt." + ); + assert_eq!(raw.messages[2].timestamp, Some(1790784008125)); + assert_eq!(raw.events.len(), 6); + assert_eq!(raw.events[0].kind, "file_read"); + assert_eq!(raw.events[0].files[0].path, "input.txt"); + assert_eq!(raw.events[0].files[0].operation, FileOperation::Read); + assert_eq!(raw.events[0].files[0].cwd.as_deref(), raw.directory.as_deref()); + assert_eq!(raw.events[0].tool_call_id.as_deref(), Some("read-1")); + assert_eq!(raw.events[1].status.as_deref(), Some("success")); + assert_eq!(raw.events[2].kind, "command"); + assert_eq!(raw.events[2].target.as_deref(), Some("wc -l input.txt")); + assert_eq!(raw.events[3].status.as_deref(), Some("success")); + assert_eq!(raw.events[3].attrs_json.as_deref(), Some(r#"{"exit_code":0}"#)); + assert_eq!(raw.events[4].kind, "file_write"); + assert_eq!(raw.events[4].files[0].path, "output.txt"); + assert_eq!(raw.events[4].files[0].operation, FileOperation::Write); + assert_eq!(raw.events[5].status.as_deref(), Some("success")); + assert_eq!(raw.usage_events.len(), 1); + let usage = &raw.usage_events[0]; + assert_eq!(usage.model, "test-model"); + assert_eq!(usage.input_tokens, 11); + assert_eq!(usage.output_tokens, 7); + assert_eq!(usage.cache_read_tokens, 3); + assert_eq!(usage.cache_write_tokens, 2); + assert_eq!(usage.token_source, crate::types::TokenSource::Observed); + assert_eq!(std::fs::read(path).unwrap(), before); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM agent_tasks", [], |row| row.get::<_, i64>(0)).unwrap(), + 1 + ); +} + +#[test] +fn live_wal_snapshot_excludes_uncommitted_changes_and_sees_committed_changes() { + let (_dir, path, conn) = database(); + conn.execute_batch("PRAGMA journal_mode=WAL; BEGIN IMMEDIATE;").unwrap(); + let mut task = proto::Task::decode(TASK).unwrap(); + task.messages[0].user_query.as_mut().unwrap().query = "Updated prompt".into(); + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert!(raw.messages[0].content.starts_with("Read input.txt")); + conn.execute_batch("COMMIT;").unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.messages[0].content, "Updated prompt"); +} + +#[test] +fn interleaved_subtasks_do_not_replace_the_root_tasks_tool_directory() { + let (_dir, path, conn) = database(); + let child = proto::Task { + id: "child-task".into(), + dependencies: Some(proto::Dependencies { parent_task_id: "task-root".into() }), + messages: vec![proto::Message { + id: "child-prompt".into(), + timestamp: Some(proto::Timestamp { seconds: 1790784001, nanos: 500000000 }), + user_query: Some(proto::UserQuery { + query: "Child task".into(), + origin: None, + context: Some(proto::InputContext { + directory: Some(proto::Directory { pwd: "/tmp/child-project".into() }), + }), + }), + ..Default::default() + }], + ..Default::default() + }; + conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('conversation-1', 'child-task', ?1)", [child.encode_to_vec()]).unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.directory.as_deref(), Some("/tmp/recall-warp-project")); + assert_eq!(raw.events[0].files[0].cwd.as_deref(), Some("/tmp/recall-warp-project")); +} + +#[test] +fn corrupt_task_skips_its_conversation_without_losing_other_conversations() { + let (_dir, path, conn) = database(); + conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('bad', '{}'); + INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('bad', 'bad-task', x'ff');").unwrap(); + let scan = scan_db(Some(&path), None, true).unwrap(); + assert_eq!(scan.stats.candidates, 2); + assert_eq!(scan.sessions.len(), 1); + assert_eq!(scan.sessions[0].source_id, "conversation-1"); +} + +#[test] +fn aggregate_stats_and_absent_token_counts_do_not_invent_usage_or_tool_calls() { + let (_dir, path, conn) = database(); + let mut task = proto::Task::decode(TASK).unwrap(); + for message in &mut task.messages { + message.request_metadata = None; + message.tool_call = None; + message.tool_call_result = None; + } + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.messages.len(), 3); + assert!(raw.events.is_empty()); + assert!(raw.usage_events.is_empty()); +} + +#[test] +fn missing_timestamps_preserve_message_order_and_duplicates_are_not_reindexed() { + let (_dir, path, conn) = database(); + let mut task = proto::Task::decode(TASK).unwrap(); + task.messages[1].timestamp = None; + let duplicate = task.messages[0].clone(); + task.messages.push(duplicate); + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.messages.len(), 3); + assert_eq!(raw.messages[0].role, Role::User); + assert_eq!(raw.messages[1].content, "I will inspect the file first."); + assert_eq!(raw.messages[1].timestamp, None); + assert_eq!(raw.messages[2].timestamp, Some(1790784008125)); +} + +#[test] +fn restricted_streaming_scan_and_usage_only_scan_obey_the_core_contract() { + let (_dir, path, conn) = database(); + conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('other', '{}');").unwrap(); + let mut streamed = Vec::new(); + let mut sink = |raw: RawSession| { + streamed.push(raw); + Ok(()) + }; + let context = AdapterSyncContext::empty_for_test("warp") + .restricted_to("conversation-1") + .with_session_sink(&mut sink); + let scan = scan_db(Some(&path), Some(&context), false).unwrap(); + assert_eq!(scan.stats.candidates, 1); + assert!(scan.sessions.is_empty()); + drop(context); + assert_eq!(streamed.len(), 1); + assert!(streamed[0].events.is_empty()); + assert_eq!(streamed[0].event_parser_version, None); + assert_eq!(streamed[0].usage_events.len(), 1); +} + +#[test] +fn legacy_queries_do_not_replace_missing_tasks_in_a_modern_database() { + let (_dir, path, conn) = database(); + conn.execute_batch("CREATE TABLE ai_queries (id INTEGER PRIMARY KEY, conversation_id TEXT, input TEXT, working_directory TEXT, start_ts TEXT); + INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('legacy', '{}');").unwrap(); + let input = r#"[{"Context":{"text":"hidden context"}},{"Query":{"text":"Legacy prompt"}}]"#; + for id in ["legacy", "conversation-1"] { + conn.execute("INSERT INTO ai_queries (conversation_id, input, working_directory, start_ts) VALUES (?1, ?2, '/tmp/legacy', '2026-09-30 15:00:00')", params![id, input]).unwrap(); + } + let scan = scan_db(Some(&path), None, true).unwrap(); + assert_eq!(scan.sessions.len(), 1); + assert_eq!(scan.sessions[0].messages.len(), 3); + assert_eq!(scan.sessions[0].source_id, "conversation-1"); + assert_eq!(scan.stats.filtered_sessions, 1); +} + +#[test] +fn shell_results_keep_executed_commands_and_failure_status() { + let (_dir, path, conn) = database(); + let mut task = proto::Task::decode(TASK).unwrap(); + let shell = + task.messages[5].tool_call_result.as_mut().unwrap().run_shell_command.as_mut().unwrap(); + shell.command = "mv input.txt changed.txt".into(); + shell.command_finished.as_mut().unwrap().exit_code = 1; + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let result = &raw.events[3]; + assert_eq!(result.status.as_deref(), Some("error")); + assert_eq!(result.target.as_deref(), Some("mv input.txt changed.txt")); + assert!(result.files.iter().any(|file| file.path == "changed.txt")); + assert_eq!(result.attrs_json.as_deref(), Some(r#"{"exit_code":1}"#)); +} + +#[test] +fn file_delete_move_and_opaque_tool_variants_preserve_source_evidence() { + let context = || EventContext { + event_seq: 0, + timestamp: None, + source_path: None, + source_event_id: None, + message_seq: None, + parser_version: EVENT_PARSER_VERSION, + }; + let call = proto::ToolCall { + apply_file_diffs: Some(proto::ApplyFileDiffs { + deleted_files: vec![proto::FilePath { file_path: "deleted.txt".into() }], + v4a_updates: vec![proto::FileUpdate { + file_path: "old.txt".into(), + move_to: "new.txt".into(), + }], + ..Default::default() + }), + ..Default::default() + }; + let event = call_event(&call, context(), Some("/tmp/project")); + assert_eq!( + event.files.iter().map(|file| file.operation.clone()).collect::>(), + [FileOperation::Delete, FileOperation::MoveFrom, FileOperation::MoveTo] + ); + let call = proto::ToolCall { + other: Some(proto::OtherTool::CallMcpTool(Vec::new())), + ..Default::default() + }; + assert_eq!(call_event(&call, context(), None).name.as_deref(), Some("call_mcp_tool")); +} + +#[test] +fn missing_sources_are_empty_and_not_created() { + let dir = tempfile::tempdir().unwrap(); + let missing = dir.path().join("missing.sqlite"); + assert!(scan_db(Some(&missing), None, true).unwrap().sessions.is_empty()); + assert!(!missing.exists()); + let path = dir.path().join("empty.sqlite"); + let _conn = Connection::open(&path).unwrap(); + assert!(scan_db(Some(&path), None, true).unwrap().sessions.is_empty()); +} + +#[test] +fn sqlite_and_protobuf_timestamps_are_milliseconds_without_fabricated_values() { + assert_eq!(parse_timestamp("2026-09-30 16:00:09.250"), Some(1790784009250)); + assert_eq!(parse_timestamp("2026-09-30T18:00:09.250+02:00"), Some(1790784009250)); + assert_eq!(parse_timestamp("not a date"), None); + assert_eq!(proto::Timestamp { seconds: 1, nanos: 1_000_000_000 }.millis(), None); + assert_eq!(proto::Timestamp { seconds: i64::MAX, nanos: 0 }.millis(), None); +} + +#[test] +fn official_origin_fixture_excludes_agent_and_automated_prompts() { + let (_dir, path, conn) = database(); + let task = + proto::Task::decode(include_bytes!("../../../tests/fixtures/warp/origins.bin").as_slice()) + .unwrap(); + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!( + raw.messages.iter().map(|message| message.content.as_str()).collect::>(), + ["origin-legacy", "origin-client", "origin-external", "origin-api", "origin-unknown"] + ); + assert!(raw.messages.iter().all(|message| message.role == Role::User)); +} + +#[test] +fn forked_tasks_preserve_message_based_usage_keys_and_parent_links() { + let (_dir, path, conn) = database(); + let mut fork = proto::Task::decode(TASK).unwrap(); + fork.id = "fork-task".into(); + for message in &mut fork.messages { + message.task_id = fork.id.clone(); + } + conn.execute( + "INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('fork', ?1)", + [r#"{"parent_conversation_id":"conversation-1"}"#], + ) + .unwrap(); + conn.execute( + "INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('fork', 'fork-task', ?1)", + [fork.encode_to_vec()], + ) + .unwrap(); + let scan = scan_db(Some(&path), None, true).unwrap(); + let root = &scan.sessions[0]; + let child = &scan.sessions[1]; + assert_eq!(root.usage_events[0].event_key, child.usage_events[0].event_key); + assert_eq!( + serde_json::from_str::(&child.usage_events[0].event_key).unwrap(), + json!(["usage-1", "primary_agent", "warp", "test-model"]) + ); + assert_eq!(child.thread_role, Some(ThreadRole::Subagent)); + assert_eq!( + child.parent_links, + vec![ParentLink { + relation: ParentRelation::Spawn, + source: "warp".into(), + source_id: "conversation-1".into() + }] + ); + assert_eq!(child.metadata_parser_version, Some(METADATA_PARSER_VERSION)); + assert!(root.thread_role.is_none()); + assert!(root.parent_links.is_empty()); +} + +#[test] +fn multiple_database_scan_streams_unique_conversations_across_channels() { + let (_dir, path, _conn) = database(); + let (_other_dir, other_path, conn) = database(); + conn.execute("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('other', '{}')", []).unwrap(); + conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('other', 'other-task', ?1)", [TASK]).unwrap(); + let mut streamed = Vec::new(); + let mut sink = |raw: RawSession| { + streamed.push(raw); + Ok(()) + }; + let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); + let scan = scan_paths(&[path, other_path], Some(&context), true).unwrap(); + assert_eq!(scan.stats.candidates, 3); + assert_eq!(scan.stats.parsed, 2); + assert_eq!(scan.stats.filtered_sessions, 1); + assert!(scan.sessions.is_empty()); + drop(context); + assert_eq!( + streamed.iter().map(|raw| raw.source_id.as_str()).collect::>(), + ["conversation-1", "other"] + ); +} + +#[test] +fn invalid_parent_metadata_does_not_discard_task_content() { + let (_dir, path, conn) = database(); + for data in [r#"{"parent_conversation_id":" "}"#, "malformed"] { + conn.execute("UPDATE agent_conversations SET conversation_data = ?1", [data]).unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.messages.len(), 3); + assert!(raw.parent_links.is_empty()); + } +} + +#[test] +fn legacy_queries_are_read_only_when_the_task_table_is_absent() { + let (_dir, path, conn) = database(); + conn.execute_batch("DROP TABLE agent_tasks; + CREATE TABLE ai_queries (id INTEGER PRIMARY KEY, conversation_id TEXT, input TEXT, working_directory TEXT, start_ts TEXT);").unwrap(); + let input = r#"[{"Context":{"text":"hidden context"}},{"Query":{"text":"Legacy prompt"}}]"#; + conn.execute("INSERT INTO ai_queries (conversation_id, input, working_directory, start_ts) VALUES ('conversation-1', ?1, '/tmp/legacy', '2026-09-30 15:00:00')", [input]).unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.messages.len(), 1); + assert_eq!(raw.messages[0].content, "Legacy prompt"); + assert_eq!(raw.directory.as_deref(), Some("/tmp/legacy")); + assert!(raw.events.is_empty()); + assert!(raw.usage_events.is_empty()); +} + +#[test] +fn unchanged_shell_results_do_not_repeat_call_file_evidence() { + let (_dir, path, conn) = database(); + let mut task = proto::Task::decode(TASK).unwrap(); + task.messages[4].tool_call.as_mut().unwrap().run_shell_command.as_mut().unwrap().command = + "mv a.txt b.txt".into(); + task.messages[5] + .tool_call_result + .as_mut() + .unwrap() + .run_shell_command + .as_mut() + .unwrap() + .command = "mv a.txt b.txt".into(); + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!( + raw.events[2] + .files + .iter() + .map(|file| (&*file.path, file.operation.clone())) + .collect::>(), + [("a.txt", FileOperation::MoveFrom), ("b.txt", FileOperation::MoveTo)] + ); + assert!(raw.events[3].files.is_empty()); + assert_eq!(raw.events[3].target.as_deref(), Some("mv a.txt b.txt")); + assert_eq!(raw.events[3].status.as_deref(), Some("success")); + assert_eq!(raw.event_parser_version, Some(EVENT_PARSER_VERSION)); + task.messages[5].timestamp = Some(proto::Timestamp { seconds: 1790784001, nanos: 0 }); + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!( + raw.events + .iter() + .filter(|event| event.tool_call_id.as_deref() == Some("shell-1")) + .map(|event| event.files.len()) + .sum::(), + 2 + ); + task.messages[4].tool_call = None; + write_task(&conn, &task); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let result = raw + .events + .iter() + .find(|event| event.source_event_id.as_deref() == Some("shell-result")) + .unwrap(); + assert_eq!(result.files.len(), 2); +} + +#[test] +fn unreadable_database_does_not_block_another_database() { + let (_dir, path, _conn) = database(); + let bad = path.with_file_name("bad.sqlite"); + std::fs::write(&bad, b"not a database").unwrap(); + let scan = scan_paths(&[bad, path], None, true).unwrap(); + assert_eq!(scan.sessions.len(), 1); + assert_eq!(scan.stats.parsed, 1); + assert_eq!(scan.sessions[0].source_id, "conversation-1"); +} + +#[test] +fn database_error_isolation_does_not_hide_session_write_failures() { + let (_dir, path, _conn) = database(); + let mut writes = 0; + let mut sink = |_raw: RawSession| { + writes += 1; + Err(anyhow::anyhow!("index write failed")) + }; + let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); + let error = scan_paths(&[path.clone(), path], Some(&context), true).err().unwrap(); + assert!(error.is::()); + drop(context); + assert_eq!(writes, 1); +} + +#[test] +fn partial_database_read_retains_streamed_stats_and_scans_the_next_database() { + let (_dir, path, conn) = database(); + conn.execute("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('z-bad', x'ff')", []).unwrap(); + conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('z-bad', 'bad-metadata-task', ?1)", [TASK]).unwrap(); + let (_other_dir, other_path, conn) = database(); + conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data, last_modified_at) SELECT 'other', conversation_data, last_modified_at FROM agent_conversations; + UPDATE agent_tasks SET conversation_id = 'other'; + DELETE FROM agent_conversations WHERE conversation_id = 'conversation-1';").unwrap(); + let mut streamed = Vec::new(); + let mut sink = |raw: RawSession| { + streamed.push(raw); + Ok(()) + }; + let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); + let scan = scan_paths(&[path, other_path], Some(&context), true).unwrap(); + assert_eq!(scan.stats.parsed, 2); + assert_eq!(scan.stats.candidates, 3); + drop(context); + assert_eq!( + streamed.iter().map(|raw| raw.source_id.as_str()).collect::>(), + ["conversation-1", "other"] + ); +} diff --git a/src/usage.rs b/src/usage.rs index c993ddd1..d7260732 100644 --- a/src/usage.rs +++ b/src/usage.rs @@ -85,6 +85,7 @@ impl TokenTotals { pub(crate) struct UsageDedup { codex_seen: HashSet, claude_seen: HashSet, + warp_seen: HashSet, } impl UsageDedup { @@ -109,6 +110,9 @@ impl UsageDedup { { return self.claude_seen.insert(event.event_key.clone()); } + if event.source == "warp" { + return self.warp_seen.insert(event.event_key.clone()); + } true } } @@ -361,4 +365,19 @@ mod tests { assert_eq!(report.summary.events, 1); assert_eq!(report.summary.tokens.total_tokens, 14); } + + #[test] + fn aggregate_usage_dedupes_warp_events_across_forked_sessions() { + let first = + event("warp", "session-a", r#"["msg-1","primary_agent","warp","test-model"]"#, 1); + let mut duplicate = first.clone(); + duplicate.session_id = "session-b".into(); + let mut next = first.clone(); + next.event_key = r#"["msg-2","primary_agent","warp","test-model"]"#.into(); + let mut category = first.clone(); + category.event_key = r#"["msg-1","subagent","warp","test-model"]"#.into(); + let report = aggregate_usage_events(&[first, duplicate, next, category]); + assert_eq!(report.summary.events, 3); + assert_eq!(report.summary.tokens.total_tokens, 42); + } } diff --git a/tests/fixtures/warp/README.md b/tests/fixtures/warp/README.md new file mode 100644 index 00000000..d0afc3f8 --- /dev/null +++ b/tests/fixtures/warp/README.md @@ -0,0 +1,36 @@ +# Warp fixtures + +These are synthetic fixtures, not transcripts from an authenticated Warp session. + +`schema.sql` is a minimal synthetic schema based on the upstream migration that +introduced `agent_conversations` and `agent_tasks`. It retains the column types +read by the adapter, without copying the application's triggers or indexes. +Source: [Warp persistence migration](https://github.com/warpdotdev/warp/blob/d807d987652361ea970287748fe4f2ea002766cc/crates/persistence/migrations/2025-06-09-013710_create_agent_conversations_table/up.sql). + +`task.textproto` describes a user prompt, two assistant replies, file read, shell +command, file edit, tool results, ignored system/debug messages, and explicit +per-request token counts. `task.bin` was encoded with `libprotoc 36.1` using +[Warp's official task protocol](https://github.com/warpdotdev/warp-proto-apis/blob/00cdd6726f28fb6b32a6712f0ca3264dbd0956a1/apis/multi_agent/v1/task.proto). +It is independent of Recall's partial protobuf structs. Unknown protobuf fields +are deliberately present to exercise forward-compatible decoding. + +`origins.textproto` and `origins.bin` use the same official encoder and protocol +revision. They cover all eight `UserQueryOrigin` variants, absent attribution, +and an empty origin. Parent-agent, agent-message wake, schedule, automation, +and server-synthesized queries are excluded from user messages. Public API and +unknown origins remain readable because the origin records a delivery mechanism, +not proof of a non-human author: +[origin attribution](https://github.com/warpdotdev/warp-proto-apis/blob/00cdd6726f28fb6b32a6712f0ca3264dbd0956a1/apis/multi_agent/v1/attribution.proto). + +Warp persists each task with `encode_to_vec()`: +[persistence writer](https://github.com/warpdotdev/warp/blob/d807d987652361ea970287748fe4f2ea002766cc/app/src/persistence/agent.rs). +The input/output/cache token fields are separate counts; Warp adds all four to +compute total tokens: +[usage accounting](https://github.com/warpdotdev/warp/blob/d807d987652361ea970287748fe4f2ea002766cc/crates/persistence/src/model.rs). + +The installed Warp 0.2026.09.30.08.29.01 could not finish launching on the test +machine. Its embedded `CREATE TABLE agent_conversations` and `CREATE TABLE +agent_tasks` migration statements were inspected and match these read column +types. No local database was created by the app, and no locally produced agent +session was available. This fixture is not an installed database dump or a real +conversation. diff --git a/tests/fixtures/warp/origins.bin b/tests/fixtures/warp/origins.bin new file mode 100644 index 0000000000000000000000000000000000000000..5ce2b087b221aad504ee8950904028483be34603 GIT binary patch literal 501 zcmZvXOHKko5Qa@Jh$$nFG8)6mupnfDiK&GPmtMd#bcPZ-Gxm_4o&cW1_v2XJfFp1K z>9!$EX0U$vo$(U&Po=5%{{)nPvXDVXz@?Fx$4N0k=N2jW5UsTBPY^3GB2L!(3yYLl?XmJLu<>5eRN_^p^j0^OA~jYQWnABCbXX| zPg1OJc^2;?fIE;^8RH})?3`5-MzX4(oa!}bltL?q&9$!Gkwt!X$R{VK8+3+nr_9UA z!MY;MOv)Guo@XM7Q`BAI!V+z!#fpA$qW7St7n*S Date: Wed, 7 Oct 2026 10:37:55 -0400 Subject: [PATCH 2/6] fix(adapters): bound Warp scans and isolate conversation failures Signed-off-by: samzong --- src/adapters/warp.rs | 248 ++++++++++++++++++------------------- src/adapters/warp/tests.rs | 111 +++++++++++------ 2 files changed, 198 insertions(+), 161 deletions(-) diff --git a/src/adapters/warp.rs b/src/adapters/warp.rs index 1934791e..78b2e393 100644 --- a/src/adapters/warp.rs +++ b/src/adapters/warp.rs @@ -25,7 +25,7 @@ mod proto; mod tests; const METADATA_PARSER_VERSION: u32 = 1; -const EVENT_PARSER_VERSION: u32 = 2; +const EVENT_PARSER_VERSION: u32 = 3; const USAGE_PARSER_VERSION: u32 = 1; pub(crate) struct WarpAdapter; @@ -48,18 +48,23 @@ impl SourceAdapter for WarpAdapter { } fn scan(&self) -> Result> { - Ok(scan_paths(&default_db_paths(), None, true)?.sessions) + Ok(scan_paths(&default_db_paths(), None, None, true)?.sessions) } fn scan_for_sync_output( &self, context: &AdapterSyncContext, - _since_ts: Option, + since_ts: Option, include_events: bool, - _force: bool, + force: bool, ) -> Result> { Ok(Some(SyncScanOutput { - scan: scan_paths(&default_db_paths(), Some(context), include_events)?, + scan: scan_paths( + &default_db_paths(), + Some(context), + if force { None } else { since_ts }, + include_events, + )?, reconcile: None, })) } @@ -103,13 +108,15 @@ fn default_db_paths() -> Vec { fn scan_paths( paths: &[PathBuf], context: Option<&AdapterSyncContext>, + since_ts: Option, include_events: bool, ) -> Result { let mut result = SyncScanResult::default(); let mut seen = HashSet::new(); for path in paths { let mut scan = SyncScanResult::default(); - let outcome = scan_db_with_seen(Some(path), context, include_events, &mut seen, &mut scan); + let outcome = + scan_db_with_seen(Some(path), context, since_ts, include_events, &mut seen, &mut scan); result.absorb(scan); if let Err(error) = outcome { if error.is::() { @@ -147,13 +154,14 @@ fn scan_db( include_events: bool, ) -> Result { let mut result = SyncScanResult::default(); - scan_db_with_seen(path, context, include_events, &mut HashSet::new(), &mut result)?; + scan_db_with_seen(path, context, None, include_events, &mut HashSet::new(), &mut result)?; Ok(result) } fn scan_db_with_seen( path: Option<&Path>, context: Option<&AdapterSyncContext>, + since_ts: Option, include_events: bool, seen: &mut HashSet, result: &mut SyncScanResult, @@ -165,11 +173,9 @@ fn scan_db_with_seen( .context("opening Warp database read-only")?; conn.busy_timeout(Duration::from_secs(3))?; let snapshot = conn.unchecked_transaction()?; - if !has_table(&snapshot, "agent_conversations")? { + if !has_table(&snapshot, "agent_conversations")? || !has_table(&snapshot, "agent_tasks")? { return Ok(()); } - let has_tasks = has_table(&snapshot, "agent_tasks")?; - let has_queries = has_table(&snapshot, "ai_queries")?; let mut stmt = snapshot.prepare( "SELECT conversation_id, last_modified_at, conversation_data FROM agent_conversations WHERE (?1 IS NULL OR conversation_id = ?1) ORDER BY conversation_id", @@ -180,38 +186,41 @@ fn scan_db_with_seen( let parsed = (|| { let id: String = row.get(0)?; let modified: String = row.get(1)?; - parse_conversation( - &snapshot, - &id, - &modified, - path, - has_tasks, - has_queries, - include_events, - ) - })(); - match parsed { - Ok(Some(mut raw)) => { - let data: String = row.get(2)?; - match serde_json::from_str::(&data) { - Ok(data) => { - if let Some(parent) = data - .get("parent_conversation_id") - .and_then(Value::as_str) - .map(str::trim) - .filter(|parent| !parent.is_empty()) - { - raw.thread_role = Some(ThreadRole::Subagent); - raw.parent_links = vec![ParentLink { - relation: ParentRelation::Spawn, - source: "warp".into(), - source_id: parent.into(), - }]; - } + if let Some(cutoff) = since_ts + && before_cutoff(&snapshot, &id, &modified, cutoff)? + { + result.stats.rejected_before_parse += 1; + return Ok(None); + } + let data: String = row.get(2)?; + let Some(mut raw) = + parse_conversation(&snapshot, &id, &modified, path, include_events)? + else { + return Ok(None); + }; + match serde_json::from_str::(&data) { + Ok(data) => { + if let Some(parent) = data + .get("parent_conversation_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|parent| !parent.is_empty()) + { + raw.thread_role = Some(ThreadRole::Subagent); + raw.parent_links = vec![ParentLink { + relation: ParentRelation::Spawn, + source: "warp".into(), + source_id: parent.into(), + }]; } - Err(error) => warn!("failed to parse Warp conversation metadata: {error}"), } - raw.metadata_parser_version = Some(METADATA_PARSER_VERSION); + Err(error) => warn!("failed to parse Warp conversation metadata: {error}"), + } + raw.metadata_parser_version = Some(METADATA_PARSER_VERSION); + Ok::<_, anyhow::Error>(Some(raw)) + })(); + match parsed { + Ok(Some(raw)) => { if !seen.insert(raw.source_id.clone()) { result.stats.filtered_sessions += 1; continue; @@ -231,6 +240,22 @@ fn scan_db_with_seen( Ok(()) } +fn before_cutoff(conn: &Connection, id: &str, modified: &str, cutoff: i64) -> Result { + if parse_timestamp(modified).is_none_or(|timestamp| timestamp >= cutoff) { + return Ok(false); + } + let mut stmt = + conn.prepare("SELECT last_modified_at FROM agent_tasks WHERE conversation_id = ?1")?; + let mut rows = stmt.query([id])?; + while let Some(row) = rows.next()? { + let modified: String = row.get(0)?; + if parse_timestamp(&modified).is_none_or(|timestamp| timestamp >= cutoff) { + return Ok(false); + } + } + Ok(true) +} + fn parse_timestamp(text: &str) -> Option { DateTime::parse_from_rfc3339(text.trim()).map(|time| time.timestamp_millis()).ok().or_else( || { @@ -262,44 +287,39 @@ fn parse_conversation( id: &str, modified: &str, path: &Path, - has_tasks: bool, - has_queries: bool, include_events: bool, ) -> Result> { let mut updated_at = parse_timestamp(modified); let mut records = Vec::new(); let mut title = None; - if has_tasks { - let mut stmt = conn.prepare( - "SELECT task, last_modified_at FROM agent_tasks WHERE conversation_id = ?1 ORDER BY id", - )?; - let mut rows = stmt.query([id])?; - while let Some(row) = rows.next()? { - let bytes: Vec = row.get(0)?; - let task = - proto::Task::decode(bytes.as_slice()).context("decoding Warp task protobuf")?; - let modified: String = row.get(1)?; - updated_at = updated_at.max(parse_timestamp(&modified)); - if task.dependencies.as_ref().is_none_or(|deps| deps.parent_task_id.is_empty()) - && !task.description.trim().is_empty() - && title.is_none() - { - title = Some(task.description); - } - let mut ordering_time = task - .messages - .iter() - .find_map(|message| message.timestamp.as_ref().and_then(proto::Timestamp::millis)); - for (seq, message) in task.messages.into_iter().enumerate() { - let key = if message.id.is_empty() { - format!("{}:{seq}", task.id) - } else { - message.id.clone() - }; - ordering_time = - message.timestamp.as_ref().and_then(proto::Timestamp::millis).or(ordering_time); - records.push((ordering_time, key, task.id.clone(), message)); - } + let mut stmt = conn.prepare( + "SELECT task, last_modified_at FROM agent_tasks WHERE conversation_id = ?1 ORDER BY id", + )?; + let mut rows = stmt.query([id])?; + while let Some(row) = rows.next()? { + let bytes: Vec = row.get(0)?; + let task = proto::Task::decode(bytes.as_slice()).context("decoding Warp task protobuf")?; + let modified: String = row.get(1)?; + updated_at = updated_at.max(parse_timestamp(&modified)); + if task.dependencies.as_ref().is_none_or(|deps| deps.parent_task_id.is_empty()) + && !task.description.trim().is_empty() + && title.is_none() + { + title = Some(task.description); + } + let mut ordering_time = task + .messages + .iter() + .find_map(|message| message.timestamp.as_ref().and_then(proto::Timestamp::millis)); + for (seq, message) in task.messages.into_iter().enumerate() { + let key = if message.id.is_empty() { + format!("{}:{seq}", task.id) + } else { + message.id.clone() + }; + ordering_time = + message.timestamp.as_ref().and_then(proto::Timestamp::millis).or(ordering_time); + records.push((ordering_time, key, task.id.clone(), message)); } } records.sort_by_key(|(time, _, _, _)| *time); @@ -310,15 +330,18 @@ fn parse_conversation( let mut usage_events = Vec::new(); let mut directory = None; let mut task_directories = HashMap::new(); - let mut tool_names = HashMap::new(); - let shell_commands: HashMap<_, _> = records + let tool_calls: HashMap<_, _> = records .iter() .filter_map(|(_, _, _, message)| message.tool_call.as_ref()) .filter(|call| !call.tool_call_id.is_empty()) - .filter_map(|call| { - call.run_shell_command - .as_ref() - .map(|shell| (call.tool_call_id.clone(), shell.command.clone())) + .map(|call| { + ( + call.tool_call_id.clone(), + ( + call_name(call).to_string(), + call.run_shell_command.as_ref().map(|shell| shell.command.clone()), + ), + ) }) .collect(); let mut started_at = None; @@ -371,11 +394,11 @@ fn parse_conversation( }; if let Some(call) = message.tool_call { let event = call_event(&call, context, active_cwd); - tool_names.insert(call.tool_call_id, event.name.clone()); session_events.push(event); } else if let Some(result) = message.tool_call_result { - let name = tool_names.get(&result.tool_call_id).cloned().flatten(); - let command = shell_commands.get(&result.tool_call_id).map(String::as_str); + let call = tool_calls.get(&result.tool_call_id); + let name = call.map(|(name, _)| name.clone()); + let command = call.and_then(|(_, command)| command.as_deref()); session_events.push(result_event(&result, context, name, command, active_cwd)); } } @@ -385,36 +408,6 @@ fn parse_conversation( parse_usage(&metadata, &key, timestamp, &source_path, &mut usage_events)?; } } - if !has_tasks && has_queries { - let mut stmt = conn.prepare( - "SELECT input, working_directory, start_ts FROM ai_queries - WHERE conversation_id = ?1 ORDER BY start_ts, id", - )?; - let mut rows = stmt.query([id])?; - while let Some(row) = rows.next()? { - let input: Option = row.get(0)?; - let pwd: Option = row.get(1)?; - let time: String = row.get(2)?; - let timestamp = parse_timestamp(&time); - started_at = started_at.or(timestamp); - updated_at = updated_at.max(timestamp); - directory = directory.or(pwd.filter(|pwd| !pwd.trim().is_empty())); - if let Some(input) = input { - let input: Vec = serde_json::from_str(&input)?; - for item in input { - if let Some(text) = item.pointer("/Query/text").and_then(Value::as_str) - && !text.trim().is_empty() - { - messages.push(RawMessage { - role: Role::User, - content: text.into(), - timestamp, - }); - } - } - } - } - } if messages.is_empty() && session_events.is_empty() && usage_events.is_empty() { return Ok(None); } @@ -430,18 +423,28 @@ fn parse_conversation( Ok(Some(raw)) } +fn call_name(call: &proto::ToolCall) -> &str { + if call.run_shell_command.is_some() { + "run_shell_command" + } else if call.read_files.is_some() { + "read_files" + } else if call.apply_file_diffs.is_some() { + "apply_file_diffs" + } else { + call.other.as_ref().map(proto::OtherTool::name).unwrap_or("unknown") + } +} + fn call_event(call: &proto::ToolCall, context: EventContext, cwd: Option<&str>) -> RawSessionEvent { let mut files = Vec::new(); - let (name, args) = if let Some(shell) = &call.run_shell_command { - ("run_shell_command", Some(json!({"command": shell.command}))) + let name = call_name(call); + let args = if let Some(shell) = &call.run_shell_command { + Some(json!({"command": shell.command})) } else if let Some(read) = &call.read_files { files.extend(read.files.iter().filter(|file| !file.name.trim().is_empty()).map(|file| { FileEvidence::call(file.name.clone(), FileOperation::Read, cwd.map(str::to_string)) })); - ( - "read_files", - Some(json!({"paths": read.files.iter().map(|file| &file.name).collect::>()})), - ) + Some(json!({"paths": read.files.iter().map(|file| &file.name).collect::>()})) } else if let Some(diff) = &call.apply_file_diffs { for (paths, operation) in [ (&diff.diffs, FileOperation::Write), @@ -474,14 +477,11 @@ fn call_event(call: &proto::ToolCall, context: EventContext, cwd: Option<&str>) )); } } - ( - "apply_file_diffs", - Some( - json!({"summary": diff.summary, "paths": files.iter().map(|file| &file.path).collect::>()}), - ), + Some( + json!({"summary": diff.summary, "paths": files.iter().map(|file| &file.path).collect::>()}), ) } else { - (call.other.as_ref().map(proto::OtherTool::name).unwrap_or("unknown"), None) + None }; let mut event = events::tool_call_event(context, name.into(), args.as_ref()); event.tool_call_id = Some(call.tool_call_id.clone()).filter(|id| !id.is_empty()); diff --git a/src/adapters/warp/tests.rs b/src/adapters/warp/tests.rs index 2a383ba0..83d39bee 100644 --- a/src/adapters/warp/tests.rs +++ b/src/adapters/warp/tests.rs @@ -191,7 +191,7 @@ fn restricted_streaming_scan_and_usage_only_scan_obey_the_core_contract() { } #[test] -fn legacy_queries_do_not_replace_missing_tasks_in_a_modern_database() { +fn taskless_conversations_do_not_emit_user_only_sessions() { let (_dir, path, conn) = database(); conn.execute_batch("CREATE TABLE ai_queries (id INTEGER PRIMARY KEY, conversation_id TEXT, input TEXT, working_directory TEXT, start_ts TEXT); INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('legacy', '{}');").unwrap(); @@ -343,7 +343,7 @@ fn multiple_database_scan_streams_unique_conversations_across_channels() { Ok(()) }; let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); - let scan = scan_paths(&[path, other_path], Some(&context), true).unwrap(); + let scan = scan_paths(&[path, other_path], Some(&context), None, true).unwrap(); assert_eq!(scan.stats.candidates, 3); assert_eq!(scan.stats.parsed, 2); assert_eq!(scan.stats.filtered_sessions, 1); @@ -366,21 +366,6 @@ fn invalid_parent_metadata_does_not_discard_task_content() { } } -#[test] -fn legacy_queries_are_read_only_when_the_task_table_is_absent() { - let (_dir, path, conn) = database(); - conn.execute_batch("DROP TABLE agent_tasks; - CREATE TABLE ai_queries (id INTEGER PRIMARY KEY, conversation_id TEXT, input TEXT, working_directory TEXT, start_ts TEXT);").unwrap(); - let input = r#"[{"Context":{"text":"hidden context"}},{"Query":{"text":"Legacy prompt"}}]"#; - conn.execute("INSERT INTO ai_queries (conversation_id, input, working_directory, start_ts) VALUES ('conversation-1', ?1, '/tmp/legacy', '2026-09-30 15:00:00')", [input]).unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); - assert_eq!(raw.messages.len(), 1); - assert_eq!(raw.messages[0].content, "Legacy prompt"); - assert_eq!(raw.directory.as_deref(), Some("/tmp/legacy")); - assert!(raw.events.is_empty()); - assert!(raw.usage_events.is_empty()); -} - #[test] fn unchanged_shell_results_do_not_repeat_call_file_evidence() { let (_dir, path, conn) = database(); @@ -436,7 +421,7 @@ fn unreadable_database_does_not_block_another_database() { let (_dir, path, _conn) = database(); let bad = path.with_file_name("bad.sqlite"); std::fs::write(&bad, b"not a database").unwrap(); - let scan = scan_paths(&[bad, path], None, true).unwrap(); + let scan = scan_paths(&[bad, path], None, None, true).unwrap(); assert_eq!(scan.sessions.len(), 1); assert_eq!(scan.stats.parsed, 1); assert_eq!(scan.sessions[0].source_id, "conversation-1"); @@ -451,33 +436,85 @@ fn database_error_isolation_does_not_hide_session_write_failures() { Err(anyhow::anyhow!("index write failed")) }; let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); - let error = scan_paths(&[path.clone(), path], Some(&context), true).err().unwrap(); + let error = scan_paths(&[path.clone(), path], Some(&context), None, true).err().unwrap(); assert!(error.is::()); drop(context); assert_eq!(writes, 1); } #[test] -fn partial_database_read_retains_streamed_stats_and_scans_the_next_database() { +fn invalid_utf8_metadata_skips_only_its_conversation() { let (_dir, path, conn) = database(); - conn.execute("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('z-bad', x'ff')", []).unwrap(); - conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('z-bad', 'bad-metadata-task', ?1)", [TASK]).unwrap(); - let (_other_dir, other_path, conn) = database(); - conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data, last_modified_at) SELECT 'other', conversation_data, last_modified_at FROM agent_conversations; - UPDATE agent_tasks SET conversation_id = 'other'; - DELETE FROM agent_conversations WHERE conversation_id = 'conversation-1';").unwrap(); - let mut streamed = Vec::new(); - let mut sink = |raw: RawSession| { - streamed.push(raw); - Ok(()) - }; - let context = AdapterSyncContext::empty_for_test("warp").with_session_sink(&mut sink); - let scan = scan_paths(&[path, other_path], Some(&context), true).unwrap(); - assert_eq!(scan.stats.parsed, 2); + conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('a-bad', x'ff'), ('z-good', '{}');").unwrap(); + for id in ["a-bad", "z-good"] { + conn.execute( + "INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES (?1, ?1, ?2)", + params![id, TASK], + ) + .unwrap(); + } + let scan = scan_db(Some(&path), None, true).unwrap(); assert_eq!(scan.stats.candidates, 3); - drop(context); + assert_eq!(scan.stats.parsed, 2); assert_eq!( - streamed.iter().map(|raw| raw.source_id.as_str()).collect::>(), - ["conversation-1", "other"] + scan.sessions.iter().map(|raw| raw.source_id.as_str()).collect::>(), + ["conversation-1", "z-good"] + ); + conn.execute_batch("UPDATE agent_conversations SET conversation_data = CAST(x'ff' AS TEXT) WHERE conversation_id = 'a-bad';").unwrap(); + assert_eq!(scan_db(Some(&path), None, true).unwrap().sessions.len(), 2); +} + +#[test] +fn cutoff_rejects_old_tasks_before_decoding_and_keeps_recent_task_updates() { + let (_dir, path, conn) = database(); + let cutoff = 1790784009250; + conn.execute_batch("UPDATE agent_tasks SET task = x'ff', last_modified_at = '2026-09-29 00:00:00'; + UPDATE agent_conversations SET last_modified_at = '2026-09-29 00:00:00'; + INSERT INTO agent_conversations (conversation_id, conversation_data, last_modified_at) VALUES ('recent-task', '{}', '2026-09-29 00:00:00'), ('recent-conversation', '{}', '2026-09-30T18:00:09.250+02:00'), ('unknown-time', '{}', 'unknown');").unwrap(); + for (id, modified) in [ + ("recent-task", "2026-09-30T18:00:09.250+02:00"), + ("recent-conversation", "2026-09-29 00:00:00"), + ("unknown-time", "unknown"), + ] { + conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task, last_modified_at) VALUES (?1, ?1, ?2, ?3)", params![id, TASK, modified]).unwrap(); + } + let scan = scan_paths(&[path], None, Some(cutoff), true).unwrap(); + assert_eq!(scan.stats.candidates, 4); + assert_eq!(scan.stats.rejected_before_parse, 1); + assert_eq!(scan.stats.parsed, 3); + assert_eq!( + scan.sessions.iter().map(|raw| raw.source_id.as_str()).collect::>(), + ["recent-conversation", "recent-task", "unknown-time"] ); } + +#[test] +fn interleaved_result_names_do_not_depend_on_call_timestamp_order() { + let (_dir, path, conn) = database(); + let mut task = proto::Task::decode(TASK).unwrap(); + let results: Vec<_> = task + .messages + .iter() + .filter(|message| message.tool_call_result.is_some()) + .cloned() + .map(|mut message| { + message.timestamp = Some(proto::Timestamp { seconds: 1790784001, nanos: 0 }); + message + }) + .collect(); + task.messages.retain(|message| message.tool_call_result.is_none()); + write_task(&conn, &task); + let child = proto::Task { id: "results-task".into(), messages: results, ..Default::default() }; + conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('conversation-1', 'results-task', ?1)", [child.encode_to_vec()]).unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + for (id, name) in + [("read-1", "read_files"), ("shell-1", "run_shell_command"), ("edit-1", "apply_file_diffs")] + { + let result = raw + .events + .iter() + .find(|event| event.kind == "tool_result" && event.tool_call_id.as_deref() == Some(id)) + .unwrap(); + assert_eq!(result.name.as_deref(), Some(name)); + } +} From dd8f9dba60f60680011f9923267f7cd1d9c8401e Mon Sep 17 00:00:00 2001 From: samzong Date: Wed, 7 Oct 2026 10:49:00 -0400 Subject: [PATCH 3/6] docs(readme): list Warp in the adapter support table Signed-off-by: samzong --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 1f8b6b0d..e92036cd 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,7 @@ One index across every AI coding CLI. Sync once, search everywhere, resume right | TRAE CLI | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | JetBrains Junie | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | Zed | ✅ | ✅ | ✅ | ✅ | ✅ | — | ✅ | +| Warp | ✅ | ✅ | ✅ | ✅ | ✅ | — | ✅ | ## Acknowledgements From 5b44bbf521b70c00d2d92de3e9beff8281a9b955 Mon Sep 17 00:00:00 2001 From: samzong Date: Wed, 7 Oct 2026 12:01:19 -0400 Subject: [PATCH 4/6] test(adapters): validate Warp with a sanitized real conversation Signed-off-by: samzong --- src/adapters/warp/tests.rs | 61 ++++++++ tests/fixtures/warp/README.md | 43 ++++- tests/fixtures/warp/probe.bin | Bin 0 -> 1855 bytes tests/fixtures/warp/probe.textproto | 233 ++++++++++++++++++++++++++++ 4 files changed, 330 insertions(+), 7 deletions(-) create mode 100644 tests/fixtures/warp/probe.bin create mode 100644 tests/fixtures/warp/probe.textproto diff --git a/src/adapters/warp/tests.rs b/src/adapters/warp/tests.rs index 83d39bee..32dc7b75 100644 --- a/src/adapters/warp/tests.rs +++ b/src/adapters/warp/tests.rs @@ -2,6 +2,7 @@ use super::*; use rusqlite::params; const TASK: &[u8] = include_bytes!("../../../tests/fixtures/warp/task.bin"); +const PROBE: &[u8] = include_bytes!("../../../tests/fixtures/warp/probe.bin"); const SCHEMA: &str = include_str!("../../../tests/fixtures/warp/schema.sql"); fn database() -> (tempfile::TempDir, PathBuf, Connection) { @@ -518,3 +519,63 @@ fn interleaved_result_names_do_not_depend_on_call_timestamp_order() { assert_eq!(result.name.as_deref(), Some(name)); } } + +#[test] +fn real_probe_preserves_native_messages_events_and_missing_usage() { + let (_dir, path, conn) = database(); + conn.execute_batch("UPDATE agent_conversations SET conversation_data = '{}', last_modified_at = '2026-10-07 15:42:09'; + UPDATE agent_tasks SET last_modified_at = '2026-10-07 15:42:09';").unwrap(); + conn.execute("UPDATE agent_tasks SET task = ?1", [PROBE]).unwrap(); + let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + assert_eq!(raw.custom_title.as_deref(), Some("Count Lines And Append Text In File")); + assert_eq!(raw.directory.as_deref(), Some("/tmp/recall-warp-probe")); + assert_eq!(raw.started_at, 1791387715621); + assert_eq!(raw.updated_at, Some(1791387729000)); + assert_eq!(raw.messages.len(), 2); + assert_eq!(raw.messages[0].role, Role::User); + assert_eq!( + raw.messages[0].content, + "Read sample.txt, run wc -l sample.txt, then append a line gamma to sample.txt." + ); + assert_eq!(raw.messages[0].timestamp, Some(1791387715621)); + assert_eq!(raw.messages[1].role, Role::Assistant); + assert!(raw.messages[1].content.contains("alpha\nbeta\ngamma")); + assert_eq!(raw.messages[1].timestamp, Some(1791387728192)); + assert_eq!(raw.events.len(), 7); + let read = raw.events.iter().find(|event| event.kind == "file_read").unwrap(); + assert_eq!(read.timestamp, Some(1791387719101)); + assert_eq!(read.files.len(), 1); + assert_eq!(read.files[0].path, "/tmp/recall-warp-probe/sample.txt"); + assert_eq!(read.files[0].operation, FileOperation::Read); + assert_eq!(read.files[0].cwd.as_deref(), raw.directory.as_deref()); + let read_result = raw + .events + .iter() + .find(|event| event.kind == "tool_result" && event.name.as_deref() == Some("read_files")) + .unwrap(); + assert_eq!(read_result.status.as_deref(), Some("success")); + let commands: Vec<_> = raw.events.iter().filter(|event| event.kind == "command").collect(); + assert_eq!(commands.len(), 2); + assert_eq!(commands[0].target.as_deref(), Some("wc -l sample.txt")); + assert!(commands[1].target.as_ref().unwrap().contains("gamma >> sample.txt")); + let shell_results: Vec<_> = raw + .events + .iter() + .filter(|event| { + event.kind == "tool_result" && event.name.as_deref() == Some("run_shell_command") + }) + .collect(); + assert_eq!(shell_results.len(), 2); + assert!(shell_results[0].summary.as_ref().unwrap().contains("2 sample.txt")); + assert!( + shell_results[1].summary.as_ref().unwrap().contains("3 sample.txt\nalpha\nbeta\ngamma") + ); + for result in shell_results { + assert_eq!(result.status.as_deref(), Some("success")); + assert_eq!(result.attrs_json.as_deref(), Some(r#"{"exit_code":0}"#)); + assert!(result.files.is_empty()); + } + assert!(raw.usage_events.is_empty()); + assert!(raw.thread_role.is_none()); + assert!(raw.parent_links.is_empty()); +} diff --git a/tests/fixtures/warp/README.md b/tests/fixtures/warp/README.md index d0afc3f8..13321776 100644 --- a/tests/fixtures/warp/README.md +++ b/tests/fixtures/warp/README.md @@ -1,6 +1,34 @@ # Warp fixtures -These are synthetic fixtures, not transcripts from an authenticated Warp session. +`probe.textproto` and `probe.bin` are sanitized derivatives of a real local +conversation produced on 2026-10-07 with Warp 0.2026.09.30.08.29.01 and Grok 4.7 +(`grok-4-7-low`). The user requested reading `sample.txt`, running `wc -l`, and +appending `gamma` in `/tmp/recall-warp-probe`. The terminal output changed from +two lines to three, and the final file contained `alpha`, `beta`, and `gamma`. + +The user exported a consistent SQLite snapshot using `.backup` from Warp's own +terminal. Only the probe task was decoded using the official protocol revision +linked below. Message, task, request, and call IDs were replaced with fixture +aliases. Only allowlisted prompt/reply text, title, temporary-project paths, +commands, shell output/exit codes, timestamps, model markers, tool variants, +and query-origin markers were retained. Account/user identifiers, server data, +conversation tokens, credentials, project rules, reasoning text, attachments, +and result contexts were omitted. Read-success and ignored server/reasoning +payloads retain their original variants with empty bodies. The sanitized task +was serialized with libprotoc 36.1; it is not an unmodified original blob. + +This task contains one user message, one assistant message, four calls (including +a server call), and three results. It has no `RequestMetadata`, so Recall emits +no usage events. The source does not record a parent conversation. Both shell +commands and their successful results are preserved. The shared command-evidence +parser does not support `wc` or ordinary output redirection, so this conversation +has read-call file evidence but no inferred shell file evidence. + +The snapshot contained three conversations and two tasks. The other conversations +were checked only for indexing status and are absent from these fixtures. The +snapshot and temporary private indexes were deleted after verification. + +`task.*`, `origins.*`, and `schema.sql` remain synthetic fixtures. `schema.sql` is a minimal synthetic schema based on the upstream migration that introduced `agent_conversations` and `agent_tasks`. It retains the column types @@ -28,9 +56,10 @@ The input/output/cache token fields are separate counts; Warp adds all four to compute total tokens: [usage accounting](https://github.com/warpdotdev/warp/blob/d807d987652361ea970287748fe4f2ea002766cc/crates/persistence/src/model.rs). -The installed Warp 0.2026.09.30.08.29.01 could not finish launching on the test -machine. Its embedded `CREATE TABLE agent_conversations` and `CREATE TABLE -agent_tasks` migration statements were inspected and match these read column -types. No local database was created by the app, and no locally produced agent -session was available. This fixture is not an installed database dump or a real -conversation. +The application-created snapshot has the same adapter-read column types as +`schema.sql`, with an additional `summary` column in `agent_conversations`. +Real CLI verification used the macOS Application Support Stable fallback path +inside a sandbox HOME. Direct read-only access to the native group-container +database from the verification terminal failed under macOS access controls; +Recall warned and returned successfully. No privacy settings or Warp application +state were changed by the verifier. diff --git a/tests/fixtures/warp/probe.bin b/tests/fixtures/warp/probe.bin new file mode 100644 index 0000000000000000000000000000000000000000..f744c9cf710b984881c8306f35c69e426f227c3b GIT binary patch literal 1855 zcmd<$Dk#cNO4Th%EY22EcFr%&D^c*t%u6j+aLh|la4aZD1yUiY6(tIuc?xcsIjLHH zT>KC%xv9m)iRr1jhDrilJP@Je#GD);pCJm#)GVm-qSV6D)Z!9d^CBLO!xv^=V-sjU zxpM`RmL0lzMn**HUo!hJQ2#ncbOVfq3b_1&QWH}YiW74Sa#Hn5DoS(|ic0eo%99mz zb5KM}GE(yt5+UA9RLBAPTp>L%H#bqCBp*qgo{$um1ech8Np69DQ7YJ%<%vZFx?o=$ zF>w*;;CXXF4vs~4uL*I%HD$-qoy?~g<+ym#i}JH|O?1t5bMniD_&_`b6FqYU4Ioz& zGdxWR8Gj5M8$EaSO<}HL&?-WAo|%#sd6cEDPj){eMFGsx0ajZIM~`7N9P%6 zC7|1BuB3!wr@4?2mjHU~;kSSXYQe(BCC?eP0?=)+&=Nwi!Ay%u3N;!9AZZ$wZVio! zxH-;%-F#ruUvsTWbQ3MLtWiuf7t)r(aI=;;m#_i^7$GGv0b;G5G8Jfb92eAW@GNM6 z5*D~!{%`jI7s5%%0mFPl0_O8@oCAlpjBE=MOHM37QbnZmrf%9ci%IJUrsIsYS{Tt?S1d&> z*J){RsX!y$7|GFGi8%!siCjskC5c?%LWe*qFasvQ^Wb29`*EHrmQ;XRkf5gmg!aCl zyPOH90(*=oL@i0MnSTN7yyxF9L<1ASL(FhClUm5gHIFd?QA#8zJnftRw6){ejP(i$ z-~dT@+|ZZ+3X}weCp(tBSlIJ^Psg)4tDkPz_iRq*v))Buk> sample.txt && wc -l sample.txt && cat sample.txt" + } + } + task_id: "probe-task" + request_id: "probe-request-12" + timestamp { + seconds: 1791387726 + nanos: 592288021 + } +} +messages { + id: "probe-message-13" + tool_call_result { + tool_call_id: "probe-call-12" + run_shell_command { + command: "printf \'%s\\n\' gamma >> sample.txt && wc -l sample.txt && cat sample.txt" + command_finished { + output: " 3 sample.txt\nalpha\nbeta\ngamma" + exit_code: 0 + } + } + } + task_id: "probe-task" + request_id: "probe-request-16" + timestamp { + seconds: 1791387727 + nanos: 111655149 + } +} +messages { + id: "probe-message-14" + task_id: "probe-task" + request_id: "probe-request-16" + timestamp { + seconds: 1791387727 + nanos: 141474958 + } + model_used { + model_id: "grok-4-7-low" + model_display_name: "grok 4.7 (low)" + } +} +messages { + id: "probe-message-15" + task_id: "probe-task" + request_id: "probe-request-16" + timestamp { + seconds: 1791387728 + nanos: 192166887 + } + agent_reasoning { + } +} +messages { + id: "probe-message-16" + agent_output { + text: "`sample.txt` 原先是 `alpha`、`beta` 两行（末尾有换行），`wc -l` 是 2。\n\n已追加一行 `gamma`。现在是 3 行：\n\n```\nalpha\nbeta\ngamma\n```" + } + task_id: "probe-task" + request_id: "probe-request-16" + timestamp { + seconds: 1791387728 + nanos: 192474877 + } +} From 7c1a568176349425dd368b23652f75a6004f62e9 Mon Sep 17 00:00:00 2001 From: samzong Date: Wed, 7 Oct 2026 12:15:38 -0400 Subject: [PATCH 5/6] refactor(adapters): remove unused Warp protobuf fields Signed-off-by: samzong --- src/adapters/warp/proto.rs | 20 +------------------- src/adapters/warp/tests.rs | 3 --- tests/fixtures/warp/README.md | 5 +---- 3 files changed, 2 insertions(+), 26 deletions(-) diff --git a/src/adapters/warp/proto.rs b/src/adapters/warp/proto.rs index b22eb5b3..e06025cf 100644 --- a/src/adapters/warp/proto.rs +++ b/src/adapters/warp/proto.rs @@ -28,10 +28,6 @@ pub(super) struct Message { pub(super) tool_call: Option, #[prost(message, optional, tag = "5")] pub(super) tool_call_result: Option, - #[prost(string, tag = "11")] - pub(super) task_id: String, - #[prost(string, tag = "13")] - pub(super) request_id: String, #[prost(message, optional, tag = "14")] pub(super) timestamp: Option, #[prost(message, optional, tag = "30")] @@ -58,16 +54,12 @@ pub(super) struct UserQuery { #[derive(Clone, PartialEq, prost::Message)] pub(super) struct UserQueryOrigin { - #[prost(oneof = "QueryOrigin", tags = "1, 2, 3, 4, 5, 6, 7, 8")] + #[prost(oneof = "QueryOrigin", tags = "3, 4, 5, 6, 8")] pub(super) variant: Option, } #[derive(Clone, PartialEq, prost::Oneof)] pub(super) enum QueryOrigin { - #[prost(bytes, tag = "1")] - WarpClient(Vec), - #[prost(bytes, tag = "2")] - ExternalPlatform(Vec), #[prost(bytes, tag = "3")] ParentAgent(Vec), #[prost(bytes, tag = "4")] @@ -76,8 +68,6 @@ pub(super) enum QueryOrigin { Schedule(Vec), #[prost(bytes, tag = "6")] Automation(Vec), - #[prost(bytes, tag = "7")] - PublicApi(Vec), #[prost(bytes, tag = "8")] ServerSynthesized(Vec), } @@ -304,8 +294,6 @@ pub(super) struct ShellResult { pub(super) exit_code: Option, #[prost(string, tag = "3")] pub(super) command: String, - #[prost(bytes = "vec", optional, tag = "4")] - pub(super) long_running: Option>, #[prost(message, optional, tag = "5")] pub(super) command_finished: Option, #[prost(bytes = "vec", optional, tag = "6")] @@ -320,12 +308,6 @@ pub(super) struct ShellFinished { pub(super) output: String, #[prost(int32, tag = "2")] pub(super) exit_code: i32, - #[prost(string, tag = "3")] - pub(super) command_id: String, - #[prost(message, optional, tag = "4")] - pub(super) start_ts: Option, - #[prost(message, optional, tag = "5")] - pub(super) finish_ts: Option, } #[derive(Clone, PartialEq, prost::Message)] diff --git a/src/adapters/warp/tests.rs b/src/adapters/warp/tests.rs index 32dc7b75..5e143f80 100644 --- a/src/adapters/warp/tests.rs +++ b/src/adapters/warp/tests.rs @@ -297,9 +297,6 @@ fn forked_tasks_preserve_message_based_usage_keys_and_parent_links() { let (_dir, path, conn) = database(); let mut fork = proto::Task::decode(TASK).unwrap(); fork.id = "fork-task".into(); - for message in &mut fork.messages { - message.task_id = fork.id.clone(); - } conn.execute( "INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('fork', ?1)", [r#"{"parent_conversation_id":"conversation-1"}"#], diff --git a/tests/fixtures/warp/README.md b/tests/fixtures/warp/README.md index 13321776..877c783b 100644 --- a/tests/fixtures/warp/README.md +++ b/tests/fixtures/warp/README.md @@ -59,7 +59,4 @@ compute total tokens: The application-created snapshot has the same adapter-read column types as `schema.sql`, with an additional `summary` column in `agent_conversations`. Real CLI verification used the macOS Application Support Stable fallback path -inside a sandbox HOME. Direct read-only access to the native group-container -database from the verification terminal failed under macOS access controls; -Recall warned and returned successfully. No privacy settings or Warp application -state were changed by the verifier. +inside a sandbox HOME. From 8de10b6f580c22ddbec6dbdfe51c706fd39bb432 Mon Sep 17 00:00:00 2001 From: samzong Date: Wed, 7 Oct 2026 12:20:59 -0400 Subject: [PATCH 6/6] refactor(adapters): streamline Warp database scanning Signed-off-by: samzong --- src/adapters/warp.rs | 29 +++++++++++-------------- src/adapters/warp/tests.rs | 44 +++++++++++++++++++------------------- 2 files changed, 34 insertions(+), 39 deletions(-) diff --git a/src/adapters/warp.rs b/src/adapters/warp.rs index 78b2e393..da8f011f 100644 --- a/src/adapters/warp.rs +++ b/src/adapters/warp.rs @@ -114,11 +114,9 @@ fn scan_paths( let mut result = SyncScanResult::default(); let mut seen = HashSet::new(); for path in paths { - let mut scan = SyncScanResult::default(); - let outcome = - scan_db_with_seen(Some(path), context, since_ts, include_events, &mut seen, &mut scan); - result.absorb(scan); - if let Err(error) = outcome { + if let Err(error) = + scan_db_with_seen(path, context, since_ts, include_events, &mut seen, &mut result) + { if error.is::() { return Err(error); } @@ -149,7 +147,7 @@ fn has_table(conn: &Connection, name: &str) -> Result { #[cfg(test)] fn scan_db( - path: Option<&Path>, + path: &Path, context: Option<&AdapterSyncContext>, include_events: bool, ) -> Result { @@ -159,16 +157,16 @@ fn scan_db( } fn scan_db_with_seen( - path: Option<&Path>, + path: &Path, context: Option<&AdapterSyncContext>, since_ts: Option, include_events: bool, seen: &mut HashSet, result: &mut SyncScanResult, ) -> Result<()> { - let Some(path) = path.filter(|path| path.is_file()) else { + if !path.is_file() { return Ok(()); - }; + } let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY) .context("opening Warp database read-only")?; conn.busy_timeout(Duration::from_secs(3))?; @@ -275,13 +273,6 @@ impl proto::Timestamp { } } -fn cwd(context: Option<&proto::InputContext>) -> Option { - context - .and_then(|context| context.directory.as_ref()) - .map(|directory| directory.pwd.clone()) - .filter(|pwd| !pwd.trim().is_empty()) -} - fn parse_conversation( conn: &Connection, id: &str, @@ -354,7 +345,11 @@ fn parse_conversation( message.user_query.as_ref().and_then(|query| query.context.as_ref()).or_else(|| { message.tool_call_result.as_ref().and_then(|result| result.context.as_ref()) }); - if let Some(pwd) = cwd(context) { + if let Some(pwd) = context + .and_then(|context| context.directory.as_ref()) + .map(|directory| directory.pwd.clone()) + .filter(|pwd| !pwd.trim().is_empty()) + { directory = directory.or_else(|| Some(pwd.clone())); task_directories.insert(task_id.clone(), pwd); } diff --git a/src/adapters/warp/tests.rs b/src/adapters/warp/tests.rs index 5e143f80..b7917c43 100644 --- a/src/adapters/warp/tests.rs +++ b/src/adapters/warp/tests.rs @@ -37,7 +37,7 @@ fn write_task(conn: &Connection, task: &proto::Task) { fn official_wire_fixture_yields_messages_tools_and_observed_usage() { let (_dir, path, conn) = database(); let before = std::fs::read(&path).unwrap(); - let scan = scan_db(Some(&path), None, true).unwrap(); + let scan = scan_db(&path, None, true).unwrap(); assert_eq!(scan.sessions.len(), 1); assert_eq!(scan.stats.candidates, 1); assert_eq!(scan.stats.parsed, 1); @@ -93,10 +93,10 @@ fn live_wal_snapshot_excludes_uncommitted_changes_and_sees_committed_changes() { let mut task = proto::Task::decode(TASK).unwrap(); task.messages[0].user_query.as_mut().unwrap().query = "Updated prompt".into(); write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert!(raw.messages[0].content.starts_with("Read input.txt")); conn.execute_batch("COMMIT;").unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.messages[0].content, "Updated prompt"); } @@ -121,7 +121,7 @@ fn interleaved_subtasks_do_not_replace_the_root_tasks_tool_directory() { ..Default::default() }; conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('conversation-1', 'child-task', ?1)", [child.encode_to_vec()]).unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.directory.as_deref(), Some("/tmp/recall-warp-project")); assert_eq!(raw.events[0].files[0].cwd.as_deref(), Some("/tmp/recall-warp-project")); } @@ -131,7 +131,7 @@ fn corrupt_task_skips_its_conversation_without_losing_other_conversations() { let (_dir, path, conn) = database(); conn.execute_batch("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('bad', '{}'); INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('bad', 'bad-task', x'ff');").unwrap(); - let scan = scan_db(Some(&path), None, true).unwrap(); + let scan = scan_db(&path, None, true).unwrap(); assert_eq!(scan.stats.candidates, 2); assert_eq!(scan.sessions.len(), 1); assert_eq!(scan.sessions[0].source_id, "conversation-1"); @@ -147,7 +147,7 @@ fn aggregate_stats_and_absent_token_counts_do_not_invent_usage_or_tool_calls() { message.tool_call_result = None; } write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.messages.len(), 3); assert!(raw.events.is_empty()); assert!(raw.usage_events.is_empty()); @@ -161,7 +161,7 @@ fn missing_timestamps_preserve_message_order_and_duplicates_are_not_reindexed() let duplicate = task.messages[0].clone(); task.messages.push(duplicate); write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.messages.len(), 3); assert_eq!(raw.messages[0].role, Role::User); assert_eq!(raw.messages[1].content, "I will inspect the file first."); @@ -181,7 +181,7 @@ fn restricted_streaming_scan_and_usage_only_scan_obey_the_core_contract() { let context = AdapterSyncContext::empty_for_test("warp") .restricted_to("conversation-1") .with_session_sink(&mut sink); - let scan = scan_db(Some(&path), Some(&context), false).unwrap(); + let scan = scan_db(&path, Some(&context), false).unwrap(); assert_eq!(scan.stats.candidates, 1); assert!(scan.sessions.is_empty()); drop(context); @@ -200,7 +200,7 @@ fn taskless_conversations_do_not_emit_user_only_sessions() { for id in ["legacy", "conversation-1"] { conn.execute("INSERT INTO ai_queries (conversation_id, input, working_directory, start_ts) VALUES (?1, ?2, '/tmp/legacy', '2026-09-30 15:00:00')", params![id, input]).unwrap(); } - let scan = scan_db(Some(&path), None, true).unwrap(); + let scan = scan_db(&path, None, true).unwrap(); assert_eq!(scan.sessions.len(), 1); assert_eq!(scan.sessions[0].messages.len(), 3); assert_eq!(scan.sessions[0].source_id, "conversation-1"); @@ -216,7 +216,7 @@ fn shell_results_keep_executed_commands_and_failure_status() { shell.command = "mv input.txt changed.txt".into(); shell.command_finished.as_mut().unwrap().exit_code = 1; write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); let result = &raw.events[3]; assert_eq!(result.status.as_deref(), Some("error")); assert_eq!(result.target.as_deref(), Some("mv input.txt changed.txt")); @@ -261,11 +261,11 @@ fn file_delete_move_and_opaque_tool_variants_preserve_source_evidence() { fn missing_sources_are_empty_and_not_created() { let dir = tempfile::tempdir().unwrap(); let missing = dir.path().join("missing.sqlite"); - assert!(scan_db(Some(&missing), None, true).unwrap().sessions.is_empty()); + assert!(scan_db(&missing, None, true).unwrap().sessions.is_empty()); assert!(!missing.exists()); let path = dir.path().join("empty.sqlite"); let _conn = Connection::open(&path).unwrap(); - assert!(scan_db(Some(&path), None, true).unwrap().sessions.is_empty()); + assert!(scan_db(&path, None, true).unwrap().sessions.is_empty()); } #[test] @@ -284,7 +284,7 @@ fn official_origin_fixture_excludes_agent_and_automated_prompts() { proto::Task::decode(include_bytes!("../../../tests/fixtures/warp/origins.bin").as_slice()) .unwrap(); write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!( raw.messages.iter().map(|message| message.content.as_str()).collect::>(), ["origin-legacy", "origin-client", "origin-external", "origin-api", "origin-unknown"] @@ -307,7 +307,7 @@ fn forked_tasks_preserve_message_based_usage_keys_and_parent_links() { [fork.encode_to_vec()], ) .unwrap(); - let scan = scan_db(Some(&path), None, true).unwrap(); + let scan = scan_db(&path, None, true).unwrap(); let root = &scan.sessions[0]; let child = &scan.sessions[1]; assert_eq!(root.usage_events[0].event_key, child.usage_events[0].event_key); @@ -358,7 +358,7 @@ fn invalid_parent_metadata_does_not_discard_task_content() { let (_dir, path, conn) = database(); for data in [r#"{"parent_conversation_id":" "}"#, "malformed"] { conn.execute("UPDATE agent_conversations SET conversation_data = ?1", [data]).unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.messages.len(), 3); assert!(raw.parent_links.is_empty()); } @@ -379,7 +379,7 @@ fn unchanged_shell_results_do_not_repeat_call_file_evidence() { .unwrap() .command = "mv a.txt b.txt".into(); write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!( raw.events[2] .files @@ -394,7 +394,7 @@ fn unchanged_shell_results_do_not_repeat_call_file_evidence() { assert_eq!(raw.event_parser_version, Some(EVENT_PARSER_VERSION)); task.messages[5].timestamp = Some(proto::Timestamp { seconds: 1790784001, nanos: 0 }); write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!( raw.events .iter() @@ -405,7 +405,7 @@ fn unchanged_shell_results_do_not_repeat_call_file_evidence() { ); task.messages[4].tool_call = None; write_task(&conn, &task); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); let result = raw .events .iter() @@ -451,7 +451,7 @@ fn invalid_utf8_metadata_skips_only_its_conversation() { ) .unwrap(); } - let scan = scan_db(Some(&path), None, true).unwrap(); + let scan = scan_db(&path, None, true).unwrap(); assert_eq!(scan.stats.candidates, 3); assert_eq!(scan.stats.parsed, 2); assert_eq!( @@ -459,7 +459,7 @@ fn invalid_utf8_metadata_skips_only_its_conversation() { ["conversation-1", "z-good"] ); conn.execute_batch("UPDATE agent_conversations SET conversation_data = CAST(x'ff' AS TEXT) WHERE conversation_id = 'a-bad';").unwrap(); - assert_eq!(scan_db(Some(&path), None, true).unwrap().sessions.len(), 2); + assert_eq!(scan_db(&path, None, true).unwrap().sessions.len(), 2); } #[test] @@ -504,7 +504,7 @@ fn interleaved_result_names_do_not_depend_on_call_timestamp_order() { write_task(&conn, &task); let child = proto::Task { id: "results-task".into(), messages: results, ..Default::default() }; conn.execute("INSERT INTO agent_tasks (conversation_id, task_id, task) VALUES ('conversation-1', 'results-task', ?1)", [child.encode_to_vec()]).unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); for (id, name) in [("read-1", "read_files"), ("shell-1", "run_shell_command"), ("edit-1", "apply_file_diffs")] { @@ -523,7 +523,7 @@ fn real_probe_preserves_native_messages_events_and_missing_usage() { conn.execute_batch("UPDATE agent_conversations SET conversation_data = '{}', last_modified_at = '2026-10-07 15:42:09'; UPDATE agent_tasks SET last_modified_at = '2026-10-07 15:42:09';").unwrap(); conn.execute("UPDATE agent_tasks SET task = ?1", [PROBE]).unwrap(); - let raw = scan_db(Some(&path), None, true).unwrap().sessions.remove(0); + let raw = scan_db(&path, None, true).unwrap().sessions.remove(0); assert_eq!(raw.custom_title.as_deref(), Some("Count Lines And Append Text In File")); assert_eq!(raw.directory.as_deref(), Some("/tmp/recall-warp-probe")); assert_eq!(raw.started_at, 1791387715621);