agent_sessions/import/
database.rs

1use super::*;
2use rusqlite::{Connection, OpenFlags};
3
4#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
5pub struct DatabaseSession {
6    pub id: String,
7    pub title: Option<String>,
8    pub cwd: Option<String>,
9}
10fn open(agent: Agent, path: &Path, opts: &ReadOptions) -> Result<Connection, ImportError> {
11    if !matches!(
12        agent,
13        Agent::OpenCode
14            | Agent::Cursor
15            | Agent::Goose
16            | Agent::Hermes
17            | Agent::ZCode
18            | Agent::CursorCli
19            | Agent::Zed
20            | Agent::Warp
21    ) {
22        return Err(ReadError::InvalidOptions("no database adapter for this agent").into());
23    }
24    if opts.stop_at_byte.is_some() {
25        return Err(ReadError::InvalidOptions(
26            "database snapshots do not have byte-prefix boundaries",
27        )
28        .into());
29    }
30    let size = std::fs::metadata(path).map_err(ReadError::Io)?.len();
31    if let Some(limit) = opts.max_file_bytes
32        && size > limit
33    {
34        return Err(ReadError::TooLarge { limit }.into());
35    }
36    Connection::open_with_flags(
37        path,
38        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
39    )
40    .map_err(|e| ReadError::Database(e).into())
41}
42/// Enumerate native IDs; never creates or migrates the database.
43pub fn list_database_sessions(
44    agent: Agent,
45    path: &Path,
46    opts: &ReadOptions,
47) -> Result<Vec<DatabaseSession>, ImportError> {
48    let conn = open(agent, path, opts)?;
49    let mut result = Vec::new();
50    if matches!(agent, Agent::OpenCode | Agent::ZCode) {
51        let mut statement = conn
52            .prepare("SELECT id,title,directory FROM session ORDER BY time_created,id")
53            .map_err(ReadError::Database)?;
54        let rows = statement
55            .query_map([], |r| {
56                Ok(DatabaseSession {
57                    id: r.get(0)?,
58                    title: r.get(1)?,
59                    cwd: r.get(2)?,
60                })
61            })
62            .map_err(ReadError::Database)?;
63        for row in rows {
64            result.push(row.map_err(ReadError::Database)?);
65        }
66    } else if agent == Agent::Goose {
67        let mut statement = conn
68            .prepare("SELECT id,name,working_dir FROM sessions ORDER BY created_at,id")
69            .map_err(ReadError::Database)?;
70        let rows = statement
71            .query_map([], |r| {
72                Ok(DatabaseSession {
73                    id: r.get(0)?,
74                    title: r.get(1)?,
75                    cwd: r.get(2)?,
76                })
77            })
78            .map_err(ReadError::Database)?;
79        for row in rows {
80            result.push(row.map_err(ReadError::Database)?);
81        }
82    } else if agent == Agent::Hermes {
83        let mut statement = conn
84            .prepare("SELECT id,title,cwd FROM sessions ORDER BY started_at,id")
85            .map_err(ReadError::Database)?;
86        for row in statement
87            .query_map([], |r| {
88                Ok(DatabaseSession {
89                    id: r.get(0)?,
90                    title: r.get(1)?,
91                    cwd: r.get(2)?,
92                })
93            })
94            .map_err(ReadError::Database)?
95        {
96            result.push(row.map_err(ReadError::Database)?);
97        }
98    } else if agent == Agent::Warp {
99        let mut statement=conn.prepare("SELECT conversation_id,summary FROM agent_conversations ORDER BY last_modified_at,id").map_err(ReadError::Database)?;
100        for row in statement
101            .query_map([], |r| {
102                Ok(DatabaseSession {
103                    id: r.get(0)?,
104                    title: r.get(1)?,
105                    cwd: None,
106                })
107            })
108            .map_err(ReadError::Database)?
109        {
110            result.push(row.map_err(ReadError::Database)?);
111        }
112    } else if agent == Agent::CursorCli {
113        let meta = super::cursor_cli::metadata(&conn, opts)?;
114        result.push(DatabaseSession {
115            id: crate::parser::required(&meta, "agentId")?.into(),
116            title: crate::parser::string(&meta, "name"),
117            cwd: None,
118        });
119    } else if agent == Agent::Zed {
120        let mut statement = conn
121            .prepare("SELECT id,summary FROM threads ORDER BY updated_at,id")
122            .map_err(ReadError::Database)?;
123        for row in statement
124            .query_map([], |r| {
125                Ok(DatabaseSession {
126                    id: r.get(0)?,
127                    title: r.get(1)?,
128                    cwd: None,
129                })
130            })
131            .map_err(ReadError::Database)?
132        {
133            result.push(row.map_err(ReadError::Database)?);
134        }
135    } else {
136        let mut statement = conn
137            .prepare("SELECT key FROM cursorDiskKV WHERE key LIKE 'composerData:%' ORDER BY key")
138            .map_err(ReadError::Database)?;
139        let rows = statement
140            .query_map([], |r| r.get::<_, String>(0))
141            .map_err(ReadError::Database)?;
142        for row in rows {
143            let key = row.map_err(ReadError::Database)?;
144            result.push(DatabaseSession {
145                id: key["composerData:".len()..].into(),
146                title: None,
147                cwd: None,
148            });
149        }
150    }
151    Ok(result)
152}
153fn json(conn: &Connection, sql: &str, key: &str, opts: &ReadOptions) -> Result<Value, ImportError> {
154    // Check serialized row size in SQL before copying the value into Rust.
155    let mut stmt = conn.prepare(sql).map_err(ReadError::Database)?;
156    let mut rows = stmt.query([key]).map_err(ReadError::Database)?;
157    let row = rows
158        .next()
159        .map_err(ReadError::Database)?
160        .ok_or(LineErrorKind::MissingField("database record"))?;
161    let size = row.get::<_, u64>(0).map_err(ReadError::Database)?;
162    if opts.max_line_bytes.is_some_and(|limit| size > limit as u64) {
163        return Err(LineErrorKind::TooLong.into());
164    }
165    let bytes = row.get::<_, Vec<u8>>(1).map_err(ReadError::Database)?;
166    serde_json::from_slice(&bytes).map_err(|_| LineErrorKind::InvalidJson.into())
167}
168fn source(table: &str, key: &str) -> ImportSource {
169    ImportSource::DatabaseRow {
170        table: table.into(),
171        key: key.into(),
172    }
173}
174/// Read one session within a SQLite read transaction. No byte offsets are invented.
175/// opts.max_line_bytes bounds each serialized row; max_file_bytes bounds the DB file.
176pub fn import_database(
177    agent: Agent,
178    path: &Path,
179    id: &str,
180    opts: &ReadOptions,
181) -> Result<SessionImport, ImportError> {
182    let mut conn = open(agent, path, opts)?;
183    let tx = conn.transaction().map_err(ReadError::Database)?;
184    let mut result = SessionImport::default();
185    let mut state = State {
186        session_id: Some(id.into()),
187        ..Default::default()
188    };
189    if agent == Agent::Warp {
190        super::warp::import(&tx, id, opts, &mut result)?;
191    } else if agent == Agent::CursorCli {
192        super::cursor_cli::import(&tx, id, opts, &mut result)?;
193    } else if agent == Agent::Zed {
194        let (data_type, parent, length): (String, Option<String>, u64) = tx
195            .query_row(
196                "SELECT data_type,parent_id,length(data) FROM threads WHERE id=?1",
197                [id],
198                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
199            )
200            .map_err(ReadError::Database)?;
201        if opts.max_line_bytes.is_some_and(|n| length > n as u64) {
202            return Err(LineErrorKind::TooLong.into());
203        }
204        let data: Vec<u8> = tx
205            .query_row("SELECT data FROM threads WHERE id=?1", [id], |r| r.get(0))
206            .map_err(ReadError::Database)?;
207        let bytes = match data_type.as_str() {
208            "json" => data,
209            "zstd" => {
210                let decoder =
211                    zstd::stream::read::Decoder::new(data.as_slice()).map_err(ReadError::Io)?;
212                let limit = opts.max_line_bytes.map(|n| n as u64).unwrap_or(u64::MAX);
213                let mut decoded = Vec::new();
214                std::io::Read::read_to_end(
215                    &mut std::io::Read::take(decoder, limit.saturating_add(1)),
216                    &mut decoded,
217                )
218                .map_err(ReadError::Io)?;
219                if decoded.len() as u64 > limit {
220                    return Err(LineErrorKind::TooLong.into());
221                }
222                decoded
223            }
224            _ => return Err(LineErrorKind::InvalidField("threads.data_type".into()).into()),
225        };
226        let value: Value =
227            serde_json::from_slice(&bytes).map_err(|_| LineErrorKind::InvalidJson)?;
228        super::document(agent, &value, opts, &mut result)?;
229        for event in &mut result.events {
230            event.sources = vec![source("threads", id)];
231            event.session_id = Some(id.into());
232            if let Event::Message(m) = &mut event.value {
233                m.is_sidechain |= parent.is_some();
234            }
235        }
236    } else if matches!(agent, Agent::OpenCode | Agent::ZCode) {
237        let (directory, parent): (String, Option<String>) = tx
238            .query_row(
239                "SELECT directory,parent_id FROM session WHERE id=?1",
240                [id],
241                |r| Ok((r.get(0)?, r.get(1)?)),
242            )
243            .map_err(ReadError::Database)?;
244        state.sidechain = parent.is_some();
245        emit_meta(
246            MetaUpdate {
247                session_id: Some(id.into()),
248                cwd: Some(directory),
249                ..Default::default()
250            },
251            vec![source("session", id)],
252            opts,
253            &mut result,
254        );
255        let ids = {
256            let mut stmt = tx
257                .prepare(if agent == Agent::ZCode {
258                    "SELECT id FROM message WHERE session_id=?1 ORDER BY sequence,id"
259                } else {
260                    "SELECT id FROM message WHERE session_id=?1 ORDER BY time_created,id"
261                })
262                .map_err(ReadError::Database)?;
263            stmt.query_map([id], |r| r.get::<_, String>(0))
264                .map_err(ReadError::Database)?
265                .collect::<Result<Vec<_>, _>>()
266                .map_err(ReadError::Database)?
267        };
268        for message_id in ids {
269            let mut info = json(
270                &tx,
271                "SELECT length(CAST(data AS BLOB)),CAST(data AS BLOB) FROM message WHERE id=?1",
272                &message_id,
273                opts,
274            )?;
275            if !info.is_object() {
276                return Err(LineErrorKind::InvalidField("message.data".into()).into());
277            }
278            info["id"] = Value::String(message_id.clone());
279            info["sessionID"] = Value::String(id.into());
280            let ids = {
281                let mut stmt=tx.prepare(if agent == Agent::ZCode { "SELECT id FROM part WHERE message_id=?1 AND session_id=?2 ORDER BY sequence,id" } else { "SELECT id FROM part WHERE message_id=?1 AND session_id=?2 ORDER BY time_created,id" }).map_err(ReadError::Database)?;
282                stmt.query_map([&message_id, id], |r| r.get::<_, String>(0))
283                    .map_err(ReadError::Database)?
284                    .collect::<Result<Vec<_>, _>>()
285                    .map_err(ReadError::Database)?
286            };
287            let mut parts = Vec::new();
288            let mut sources = vec![source("message", &message_id)];
289            for part_id in ids {
290                parts.push(json(
291                    &tx,
292                    "SELECT length(CAST(data AS BLOB)),CAST(data AS BLOB) FROM part WHERE id=?1",
293                    &part_id,
294                    opts,
295                )?);
296                sources.push(source("part", &part_id));
297            }
298            let record = serde_json::json!({"info":info,"parts":parts});
299            parse_record(agent, &record, &mut state, sources, opts, &mut result)?;
300            result.summary.lines += 1;
301        }
302    } else if agent == Agent::Goose {
303        let directory: String = tx
304            .query_row("SELECT working_dir FROM sessions WHERE id=?1", [id], |r| {
305                r.get(0)
306            })
307            .map_err(ReadError::Database)?;
308        emit_meta(
309            MetaUpdate {
310                session_id: Some(id.into()),
311                cwd: Some(directory),
312                ..Default::default()
313            },
314            vec![source("sessions", id)],
315            opts,
316            &mut result,
317        );
318        let mut stmt=tx.prepare("SELECT id,message_id,role,created_timestamp,length(CAST(content_json AS BLOB)),CAST(content_json AS BLOB),length(CAST(metadata_json AS BLOB)),CAST(metadata_json AS BLOB) FROM messages WHERE session_id=?1 ORDER BY created_timestamp,id").map_err(ReadError::Database)?;
319        let mut rows = stmt.query([id]).map_err(ReadError::Database)?;
320        while let Some(row) = rows.next().map_err(ReadError::Database)? {
321            let content_size: u64 = row.get(4).map_err(ReadError::Database)?;
322            let metadata_size: Option<u64> = row.get(6).map_err(ReadError::Database)?;
323            if opts.max_line_bytes.is_some_and(|limit| {
324                content_size.saturating_add(metadata_size.unwrap_or(0)) > limit as u64
325            }) {
326                return Err(LineErrorKind::TooLong.into());
327            }
328            let content: Vec<u8> = row.get(5).map_err(ReadError::Database)?;
329            let metadata: Option<Vec<u8>> = row.get(7).map_err(ReadError::Database)?;
330            let content: Value =
331                serde_json::from_slice(&content).map_err(|_| LineErrorKind::InvalidJson)?;
332            let metadata: Value = metadata
333                .map(|v| serde_json::from_slice(&v))
334                .transpose()
335                .map_err(|_| LineErrorKind::InvalidJson)?
336                .unwrap_or(Value::Null);
337            let row_id: i64 = row.get(0).map_err(ReadError::Database)?;
338            let message_id: Option<String> = row.get(1).map_err(ReadError::Database)?;
339            let role: String = row.get(2).map_err(ReadError::Database)?;
340            let created: i64 = row.get(3).map_err(ReadError::Database)?;
341            let record = serde_json::json!({"id":message_id,"role":role,"created":created,"content":content,"metadata":metadata});
342            parse_record(
343                agent,
344                &record,
345                &mut state,
346                vec![source("messages", &row_id.to_string())],
347                opts,
348                &mut result,
349            )?;
350            result.summary.lines += 1;
351        }
352    } else if agent == Agent::Hermes {
353        let (model, cwd, parent): (Option<String>, Option<String>, Option<String>) = tx
354            .query_row(
355                "SELECT model,cwd,parent_session_id FROM sessions WHERE id=?1",
356                [id],
357                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
358            )
359            .map_err(ReadError::Database)?;
360        state.sidechain = parent.is_some();
361        emit_meta(
362            MetaUpdate {
363                session_id: Some(id.into()),
364                model,
365                cwd,
366                ..Default::default()
367            },
368            vec![source("sessions", id)],
369            opts,
370            &mut result,
371        );
372        let mut stmt = tx.prepare("SELECT id,role,length(CAST(content AS BLOB)),content,tool_call_id,length(CAST(tool_calls AS BLOB)),tool_calls,timestamp FROM messages WHERE session_id=?1 ORDER BY timestamp,id").map_err(ReadError::Database)?;
373        let mut rows = stmt.query([id]).map_err(ReadError::Database)?;
374        while let Some(row) = rows.next().map_err(ReadError::Database)? {
375            let content_size: Option<u64> = row.get(2).map_err(ReadError::Database)?;
376            let calls_size: Option<u64> = row.get(5).map_err(ReadError::Database)?;
377            if opts.max_line_bytes.is_some_and(|limit| {
378                content_size
379                    .unwrap_or(0)
380                    .saturating_add(calls_size.unwrap_or(0))
381                    > limit as u64
382            }) {
383                return Err(LineErrorKind::TooLong.into());
384            }
385            let row_id: i64 = row.get(0).map_err(ReadError::Database)?;
386            let role: String = row.get(1).map_err(ReadError::Database)?;
387            let content: Option<String> = row.get(3).map_err(ReadError::Database)?;
388            // Hermes explicitly tags structured content. Plain JSON-looking text stays text.
389            let content = match content {
390                Some(s) if s.starts_with("\0json:") => {
391                    serde_json::from_str(&s[6..]).map_err(|_| LineErrorKind::InvalidJson)?
392                }
393                Some(s) => Value::String(s),
394                None => Value::Null,
395            };
396            let call_id: Option<String> = row.get(4).map_err(ReadError::Database)?;
397            let calls: Option<String> = row.get(6).map_err(ReadError::Database)?;
398            let calls: Value = calls
399                .map(|s| serde_json::from_str(&s))
400                .transpose()
401                .map_err(|_| LineErrorKind::InvalidJson)?
402                .unwrap_or(Value::Null);
403            let timestamp: Option<f64> = row.get(7).map_err(ReadError::Database)?;
404            let mut record = serde_json::json!({"role":role,"content":content,"tool_call_id":call_id,"timestamp":timestamp});
405            if !calls.is_null() {
406                record["tool_calls"] = calls;
407            }
408            parse_record(
409                agent,
410                &record,
411                &mut state,
412                vec![source("messages", &row_id.to_string())],
413                opts,
414                &mut result,
415            )?;
416            result.summary.lines += 1;
417        }
418        tally(
419            &mut result.summary.ignored_types,
420            "hermes:session-totals-not-message-usage",
421        );
422    } else {
423        let key = format!("composerData:{id}");
424        let composer = json(
425            &tx,
426            "SELECT length(CAST(value AS BLOB)),CAST(value AS BLOB) FROM cursorDiskKV WHERE key=?1",
427            &key,
428            opts,
429        )?;
430        emit_meta(
431            MetaUpdate {
432                session_id: Some(id.into()),
433                model: composer
434                    .pointer("/modelConfig/modelName")
435                    .and_then(Value::as_str)
436                    .map(str::to_owned),
437                ..Default::default()
438            },
439            vec![source("cursorDiskKV", &key)],
440            opts,
441            &mut result,
442        );
443        let headers = composer
444            .get("fullConversationHeadersOnly")
445            .and_then(Value::as_array)
446            .ok_or(LineErrorKind::MissingField("fullConversationHeadersOnly"))?;
447        for header in headers {
448            let bubble = crate::parser::required(header, "bubbleId")?;
449            let key = format!("bubbleId:{id}:{bubble}");
450            let record = json(
451                &tx,
452                "SELECT length(CAST(value AS BLOB)),CAST(value AS BLOB) FROM cursorDiskKV WHERE key=?1",
453                &key,
454                opts,
455            )?;
456            parse_record(
457                agent,
458                &record,
459                &mut state,
460                vec![source("cursorDiskKV", &key)],
461                opts,
462                &mut result,
463            )?;
464            result.summary.lines += 1;
465        }
466    }
467    // Dropping a read transaction rolls back only its read snapshot; no DB writes.
468    result.summary.status = ReadStatus::Complete;
469    Ok(result)
470}
agent_sessions/import/database.rs

agent_sessions/import/
database.rs
