diff --git a/docs/architecture/extensions/external-ai-work-sources-design.md b/docs/architecture/extensions/external-ai-work-sources-design.md index 78c0f3ade3..2d82c389e4 100644 --- a/docs/architecture/extensions/external-ai-work-sources-design.md +++ b/docs/architecture/extensions/external-ai-work-sources-design.md @@ -157,6 +157,14 @@ Zcode 继续使用来源 `session.title`。私有索引以可选字段保留来 Codex 恢复或重试可能留下多份相同线程 ID 的 rollout。目录在分页和计数前按线程去重， 优先使用 `threads.rollout_path`，旧数据库缺少该字段时按来源记录时间选择；旧文件候选 ID 在读取正文时也解析到当前记录，避免导入已中断的早期副本。不会拼接已放弃的重试分支。 +Codex 的 `threads.history_mode = paginated` 会话直接从同一状态快照发现，不要求旧 rollout +文件存在。正文在打开时以只读事务读取 `thread_history_1.sqlite` 的 `thread_turns` 和 +`thread_items`，包含已提交 WAL，按来源序号保留用户输入、助手输出、思考和非执行工具历史。 +同一线程同时存在 SQLite 与 JSONL 时使用分页历史；原有文件候选 ID 仍能通过状态库定位， +来源线程 ID 不变。分页来源中的显式 subagent 记录不作为独立主会话；源应用归档仍不覆盖本地选择。 +目录读取不访问历史正文，旧版缺少分页字段的数据库及 JSONL 继续可读。不可读、未知格式、孤立条目 +和超预算正文明确失败，不回退到残留 JSONL 或发布空历史；分页压缩标记没有替换上下文时要求完整历史。 +该格式兼容只在本地 adapter 和实际只读来源上验证，不增加远程工作区、远程控制、Peer 或 Detached Dispatch 入口。 对旧 `codex-rollout-jsonl-v1` 中只有用户输入、没有助手文本或工具历史的副本，显式读取时 检查全部原有轮次仍带来源标记、用户输入与新来源前缀相同且没有本地新增内容，再补齐历史。 修复保留原生名称、归属及用户元数据；存储服务先保留原副本和目标副本的恢复记录，再逐个 diff --git a/src/crates/adapters/codex-adapter/AGENTS.md b/src/crates/adapters/codex-adapter/AGENTS.md index 91dd477639..7159aa5410 100644 --- a/src/crates/adapters/codex-adapter/AGENTS.md +++ b/src/crates/adapters/codex-adapter/AGENTS.md @@ -9,11 +9,14 @@ ## Session discovery and import verification ```bash -cargo test -p openbitfun-codex-adapter --lib session::tests +cargo test -p openbitfun-codex-adapter --lib session:: # Optional read-only installed-source verification (prints counts only): cargo test -p openbitfun-codex-adapter --lib inspect_installed_codex_session_titles -- --ignored --nocapture # Set OPENBITFUN_TEST_CODEX_THREAD to inspect one logical thread's full history: cargo test -p openbitfun-codex-adapter --lib inspect_installed_codex_thread_history -- --ignored --nocapture +# Verify all paginated source IDs and histories. Optionally set +# OPENBITFUN_TEST_CODEX_WORKSPACE to restrict verification to one workspace: +cargo test -p openbitfun-codex-adapter --lib inspect_installed_codex_paginated_workspace -- --ignored --nocapture ``` ## Pet source verification diff --git a/src/crates/adapters/codex-adapter/src/session.rs b/src/crates/adapters/codex-adapter/src/session.rs index b98586afa3..7ea0a99659 100644 --- a/src/crates/adapters/codex-adapter/src/session.rs +++ b/src/crates/adapters/codex-adapter/src/session.rs @@ -1,4 +1,5 @@ -//! Bounded, read-only Codex rollout import. Discovery reads headers, never full histories. +//! Bounded, read-only Codex rollout and paginated SQLite history import. +//! Discovery reads metadata, never full histories. use openbitfun_product_domains::session_import::*; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; @@ -9,10 +10,13 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; +mod sqlite_history; +#[cfg(test)] +mod sqlite_history_tests; mod titles; use titles::{TitleCache, TitleSnapshot}; -pub const SOURCE_REVISION: &str = "codex-rollout-jsonl-v2"; +pub const SOURCE_REVISION: &str = "codex-rollout-and-sqlite-v3"; const MAX_FILE_BYTES: u64 = 256 * 1024 * 1024; const MAX_LINE_BYTES: usize = 8 * 1024 * 1024; const MAX_DOCUMENT_BYTES: usize = 128 * 1024 * 1024; @@ -108,20 +112,39 @@ impl CodexSessionReader { let titles = self.title_snapshot(); let mut diagnostics = titles.diagnostics.clone(); let paths = self.logical_session_paths(&titles, &mut diagnostics)?; + let mut available = titles + .paginated + .values() + .filter(|thread| !thread.subagent) + .map(|thread| { + let mut candidate = thread.candidate.clone(); + titles.apply(&mut candidate); + candidate + }) + .collect::>(); + for path in paths { + match self.inspect_path_with_titles(&path, &titles) { + Ok(candidate) if !titles.paginated.contains_key(&candidate.source_session_id) => { + available.push(candidate) + } + Ok(_) => {} + Err(error) if diagnostics.len() < 20 => { + diagnostics.push(warning("source_unreadable", &error)) + } + Err(_) => {} + } + } + available.sort_by(|left, right| { + right + .updated_at + .cmp(&left.updated_at) + .then_with(|| left.candidate_id.cmp(&right.candidate_id)) + }); let mut candidates = Vec::new(); let mut matched = 0usize; let mut has_more = false; let query = query.to_lowercase(); - for path in paths { - let candidate = match self.inspect_path_with_titles(&path, &titles) { - Ok(candidate) => candidate, - Err(error) => { - if diagnostics.len() < 20 { - diagnostics.push(warning("source_unreadable", &error)); - } - continue; - } - }; + for candidate in available { if !query.is_empty() && !format!( "{} {}", @@ -158,6 +181,17 @@ impl CodexSessionReader { id: &str, mode: ExternalSessionImportMode, ) -> Result { + if !self.codex_home.is_absolute() { + return Err("Codex home must be an absolute host path".into()); + } + let titles = self.title_snapshot(); + if let Some(thread) = titles + .paginated + .values() + .find(|thread| thread.matches_locator(id)) + { + return sqlite_history::read(thread, mode, &titles); + } let cached = self .headers .lock() @@ -233,6 +267,9 @@ impl CodexSessionReader { ) -> Result { let candidate = self.inspect_header(path)?; let titles = self.title_snapshot(); + if let Some(thread) = titles.paginated.get(&candidate.source_session_id) { + return sqlite_history::read(thread, mode, &titles); + } // A resume can publish a new rollout after the last catalog scan. Read // only this thread's indexed file, after checking the source boundary. if let Some(current) = titles.current_rollout(&candidate.source_session_id) { diff --git a/src/crates/adapters/codex-adapter/src/session/sqlite_history.rs b/src/crates/adapters/codex-adapter/src/session/sqlite_history.rs new file mode 100644 index 0000000000..ef6b63df80 --- /dev/null +++ b/src/crates/adapters/codex-adapter/src/session/sqlite_history.rs @@ -0,0 +1,445 @@ +//! Read Codex's projected ThreadItem history without starting its runtime or +//! modifying its databases. The state catalog and history store are separate. +use super::*; +use rusqlite::{params, Connection, OpenFlags}; +use std::time::Duration; + +pub(super) struct Thread { + pub candidate: ExternalSessionCandidate, + pub rollout_path: Option, + pub directory: PathBuf, + pub subagent: bool, +} + +impl Thread { + pub(super) fn matches_locator(&self, locator: &str) -> bool { + self.candidate.candidate_id == locator + || self.rollout_path.as_deref().is_some_and(|path| { + // Keep already-persisted file locators usable after Codex has + // migrated their bodies and removed the old JSONL files. + candidate_id(Path::new(path)) == locator + || candidate_id(Path::new(path.strip_prefix("\\\\?\\").unwrap_or(path))) + == locator + }) + } +} + +fn database_path(directory: &Path) -> Result { + for entry in fs::read_dir(directory).map_err(|error| error.to_string())? { + let entry = entry.map_err(|error| error.to_string())?; + let filename = entry.file_name(); + if filename + .to_string_lossy() + .strip_prefix("thread_history_") + .and_then(|name| name.strip_suffix(".sqlite")) + .and_then(|version| version.parse::().ok()) + .is_some_and(|version| version > 1) + { + return Err("Unsupported Codex paginated history database version; no partial import was created".into()); + } + } + let path = directory.join("thread_history_1.sqlite"); + if !path.is_file() { + return Err( + "Codex paginated history database is unavailable; the source session was preserved" + .into(), + ); + } + Ok(path) +} + +fn read_error(error: rusqlite::Error) -> String { + format!("Codex paginated history cannot be read: {error}") +} + +fn hash_value(hasher: &mut Sha256, value: &[u8]) { + hasher.update((value.len() as u64).to_le_bytes()); + hasher.update(value); +} + +pub(super) fn read( + thread: &Thread, + mode: ExternalSessionImportMode, + titles: &TitleSnapshot, +) -> Result { + let connection = Connection::open_with_flags( + database_path(&thread.directory)?, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, + ) + .map_err(read_error)?; + connection + .busy_timeout(Duration::from_secs(2)) + .map_err(read_error)?; + // A read transaction sees one committed snapshot, including WAL. immutable + // would miss recent records, and reading a whole DB would expose other chats. + connection + .execute_batch("PRAGMA query_only = ON; BEGIN DEFERRED TRANSACTION;") + .map_err(read_error)?; + let id = &thread.candidate.source_session_id; + let (item_count, source_bytes, max_item): (u64, u64, u64) = connection + .query_row( + "SELECT COUNT(*), COALESCE(SUM(length(CAST(item_json AS BLOB))), 0), + COALESCE(MAX(length(CAST(item_json AS BLOB))), 0) + FROM thread_items WHERE thread_id = ?1", + [id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .map_err(read_error)?; + if source_bytes > MAX_FILE_BYTES { + return Err("Source exceeds the 256 MiB import limit".into()); + } + if max_item > MAX_LINE_BYTES as u64 { + return Err("Codex history item exceeds the 8 MiB import limit".into()); + } + let mut turn_query = connection + .prepare( + "SELECT turn_id, status, started_at, error_json FROM thread_turns + WHERE thread_id = ?1 ORDER BY rollout_ordinal, turn_id LIMIT ?2", + ) + .map_err(read_error)?; + let stored_turns = turn_query + .query_map(params![id, MAX_TURNS as i64 + 1], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, Option>(2)?, + row.get::<_, Option>(3)?, + )) + }) + .map_err(read_error)? + .collect::, _>>() + .map_err(read_error)?; + if stored_turns.len() > MAX_TURNS { + return Err("Conversation exceeds the 4096 turn import limit".into()); + } + let mut item_query = connection + .prepare( + "SELECT item_id, created_at_ms, item_json FROM thread_items + WHERE thread_id = ?1 AND turn_id = ?2 ORDER BY rollout_ordinal, item_id", + ) + .map_err(read_error)?; + let mut turns = Vec::new(); + let mut warnings = BTreeSet::new(); + let mut diagnostics = titles.diagnostics.clone(); + let mut hasher = Sha256::new(); + let mut read_items = 0u64; + let mut converted_bytes = 0usize; + for (turn_id, status, started_at, error) in stored_turns { + hash_value( + &mut hasher, + serde_json::to_string(&(&turn_id, &status, started_at, &error)) + .map_err(|error| error.to_string())? + .as_bytes(), + ); + let status = match status.as_str() { + "completed" => ExternalSessionStatus::Complete, + "inProgress" | "interrupted" | "failed" => ExternalSessionStatus::Interrupted, + _ => return Err(format!("Unsupported Codex turn status: {status}")), + }; + if let Some(error) = error.filter(|error| error != "null") { + if diagnostics.len() < 20 { + diagnostics.push(warning("source_turn_error", &error)); + } + } + let time = started_at + .and_then(|time| u64::try_from(time).ok()) + .and_then(|time| time.checked_mul(1000)) + .or(thread.candidate.created_at) + .unwrap_or(0); + let mut turn = empty_turn(turn_id.clone(), time, status); + let mut rows = item_query.query(params![id, turn_id]).map_err(read_error)?; + while let Some(row) = rows.next().map_err(read_error)? { + let item_id: String = row.get(0).map_err(read_error)?; + let created_at: Option = row.get(1).map_err(read_error)?; + let raw: String = row.get(2).map_err(read_error)?; + hash_value(&mut hasher, item_id.as_bytes()); + hash_value(&mut hasher, &created_at.unwrap_or(time).to_le_bytes()); + hash_value(&mut hasher, raw.as_bytes()); + read_items += 1; + let item: Value = serde_json::from_str(&raw) + .map_err(|_| "Malformed Codex history item; no partial import was created")?; + let kind = item["type"] + .as_str() + .ok_or("Codex history item type is missing")?; + if item["id"].as_str() != Some(item_id.as_str()) { + return Err("Codex history item id does not match its database record".into()); + } + if kind == "userMessage" { + let text = user_text(&item["content"], &mut warnings)?; + if is_injected(&text) { + warnings.insert("source_instructions_not_imported"); + continue; + } + if !turn.user_text.is_empty() || !turn.items.is_empty() { + push_turn(&mut turns, turn, &mut converted_bytes)?; + turn = empty_turn( + format!("{turn_id}:{item_id}"), + created_at.unwrap_or(time), + status, + ); + } + turn.user_text = text; + turn.timestamp = created_at.unwrap_or(time); + } else if let Some(projected) = project_item(&item, mode, &mut warnings)? { + turn.items.push(projected); + } + turn.source_event_count += 1; + } + if !turn.user_text.is_empty() || !turn.items.is_empty() { + push_turn(&mut turns, turn, &mut converted_bytes)?; + } + } + if read_items != item_count { + return Err("Codex paginated history contains items without a readable turn; no partial import was created".into()); + } + if turns.is_empty() { + return Err("No usable human conversation was found in Codex paginated history".into()); + } + if mode == ExternalSessionImportMode::LatestCompacted + && !warnings.contains("compaction_present") + { + warnings.insert("no_compaction"); + } + let mut candidate = thread.candidate.clone(); + titles.apply(&mut candidate); + candidate.file_size_bytes = source_bytes; + candidate.fingerprint = format!("sha256:{}", hex::encode(hasher.finalize())); + candidate.turn_count = Some(turns.len()); + candidate.status = turns.last().unwrap().status; + diagnostics.extend(warnings.into_iter().map(|code| warning(code, code))); + Ok(CodexSessionDocument { + candidate, + turns, + warnings: diagnostics, + }) +} + +fn empty_turn(id: String, timestamp: u64, status: ExternalSessionStatus) -> ExternalSessionTurn { + ExternalSessionTurn { + source_turn_id: id, + timestamp, + user_text: String::new(), + items: Vec::new(), + source_event_count: 0, + status, + } +} + +fn push_turn( + turns: &mut Vec, + turn: ExternalSessionTurn, + bytes: &mut usize, +) -> Result<(), String> { + if turns.len() >= MAX_TURNS { + return Err("Conversation exceeds the 4096 turn import limit".into()); + } + *bytes += serde_json::to_vec(&turn) + .map_err(|error| error.to_string())? + .len(); + if *bytes > MAX_DOCUMENT_BYTES { + return Err("Converted history exceeds the 128 MiB storage budget".into()); + } + turns.push(turn); + Ok(()) +} + +fn user_text(content: &Value, warnings: &mut BTreeSet<&'static str>) -> Result { + let mut parts = Vec::new(); + for part in content + .as_array() + .ok_or("Codex user message content is not an array")? + { + match part["type"].as_str().unwrap_or("") { + "text" => parts.push( + part["text"] + .as_str() + .ok_or("Codex user text is missing")? + .to_owned(), + ), + "image" | "localImage" | "file" | "skill" | "mention" => { + warnings.insert("attachments_not_imported"); + // Keep the reference visible without reading arbitrary source + // files, copying credentials, or activating skills/mentions. + parts.push(format!("[Codex attachment: {}]", part)); + } + kind => return Err(format!("Unsupported Codex user content type: {kind}")), + } + } + Ok(parts.join("\n")) +} + +fn tool_success(item: &Value) -> Result, String> { + match item["status"].as_str() { + Some("completed") => Ok(Some( + item["success"] != false + && item["error"].is_null() + && item["failure"].is_null() + && item["exitCode"].as_i64().is_none_or(|code| code == 0), + )), + Some("failed" | "declined" | "interrupted") => Ok(Some(false)), + Some("inProgress") | None => Ok(None), + Some("generating") if item["type"] == "imageGeneration" => Ok(None), + Some(status) => Err(format!("Unsupported Codex tool status: {status}")), + } +} + +fn project_item( + item: &Value, + mode: ExternalSessionImportMode, + warnings: &mut BTreeSet<&'static str>, +) -> Result, String> { + let kind = item["type"].as_str().unwrap_or(""); + match kind { + "agentMessage" | "plan" => { + return Ok(Some(ExternalSessionItem::Text { + text: item["text"] + .as_str() + .ok_or("Codex agent text is missing")? + .to_owned(), + })) + } + "reasoning" => { + let mut parts = Vec::new(); + for key in ["summary", "content"] { + if let Some(items) = item[key].as_array() { + for text in items { + parts.push(text.as_str().ok_or("Unsupported Codex reasoning content")?); + } + } + } + return Ok((!parts.is_empty()).then(|| ExternalSessionItem::Thinking { + text: parts.join("\n"), + })); + } + "contextCompaction" => { + warnings.insert("compaction_present"); + if mode == ExternalSessionImportMode::LatestCompacted { + return Err("Codex paginated history has no readable compaction replacement; select full history to preserve readable conversation".into()); + } + return Ok(None); + } + _ => {} + } + let success = tool_success(item)?; + let (name, input, result) = match kind { + "commandExecution" => ( + "exec_command".into(), + json!({"cmd":item["command"],"workdir":item["cwd"]}), + (success.is_some() || !item["aggregatedOutput"].is_null()).then(|| { + json!({ + "output":item["aggregatedOutput"],"exit_code":item["exitCode"], + "wall_time_seconds":item["durationMs"].as_f64().map(|ms| ms / 1000.0), + "session_id":item["processId"],"status":item["status"]}) + }), + ), + "fileChange" => ( + "apply_patch".into(), + json!({"changes":item["changes"]}), + success.map(|_| json!({"status":item["status"]})), + ), + "mcpToolCall" => ( + format!( + "mcp__{}__{}", + item["server"] + .as_str() + .ok_or("Codex MCP server is missing")?, + item["tool"].as_str().ok_or("Codex MCP tool is missing")? + ), + item["arguments"].clone(), + if !item["error"].is_null() { + Some(json!({"error":item["error"],"result":item["result"],"is_error":true})) + } else { + item.get("result") + .filter(|result| !result.is_null()) + .cloned() + }, + ), + "dynamicToolCall" => { + let tool = item["tool"] + .as_str() + .ok_or("Codex dynamic tool name is missing")?; + let name = item["namespace"] + .as_str() + .filter(|ns| !ns.is_empty()) + .map_or_else(|| tool.to_owned(), |ns| format!("{ns}.{tool}")); + let output = item + .get("contentItems") + .filter(|value| !value.is_null()) + .map(|content| { + // ThreadItem uses inputText for its display blocks; the neutral + // tool renderer already understands the standard text blocks. + let mut content = content.clone(); + if let Some(parts) = content.as_array_mut() { + for part in parts { + if part["type"] == "inputText" { + part["type"] = Value::String("text".into()); + } + } + } + content + }); + (name, item["arguments"].clone(), output) + } + "webSearch" => { + let input = if item["action"]["type"] == "search" || item["action"].is_null() { + let queries = item["action"]["queries"] + .as_array() + .cloned() + .unwrap_or_else(|| vec![item["query"].clone()]); + json!({"search_query":queries.into_iter().map(|q| json!({"q":q})).collect::>()}) + } else { + json!({"query":item["query"],"action":item["action"]}) + }; + ("web.run".into(), input, item.get("results").cloned()) + } + "imageView" => ( + "view_image".into(), + json!({"path":item["path"]}), + Some(json!({"path":item["path"]})), + ), + "imageGeneration" => { + // result can contain megabytes of inline image data. Separate the + // request from its output so one image is stored only once. + let mut input = item + .as_object() + .cloned() + .ok_or("Codex image generation item is invalid")?; + let mut result = serde_json::Map::new(); + for key in ["result", "failure", "savedPath", "status"] { + if let Some(value) = input.remove(key) { + result.insert(key.into(), value); + } + } + ( + "codex.imageGeneration".into(), + Value::Object(input), + Some(Value::Object(result)), + ) + } + "sleep" + | "subAgentActivity" + | "collabAgentToolCall" + | "hookPrompt" + | "enteredReviewMode" + | "exitedReviewMode" => { + // These source records have no native runtime or permission state. + // Preserve their complete payload as a non-executable history card. + (format!("codex.{kind}"), item.clone(), Some(item.clone())) + } + _ => { + return Err(format!( + "Unsupported Codex history item type: {kind}; no partial import was created" + )) + } + }; + Ok(Some(ExternalSessionItem::Tool { + id: item["id"] + .as_str() + .ok_or("Codex tool id is missing")? + .to_owned(), + name, + input, + result, + success, + })) +} diff --git a/src/crates/adapters/codex-adapter/src/session/sqlite_history_tests.rs b/src/crates/adapters/codex-adapter/src/session/sqlite_history_tests.rs new file mode 100644 index 0000000000..957d64f99d --- /dev/null +++ b/src/crates/adapters/codex-adapter/src/session/sqlite_history_tests.rs @@ -0,0 +1,568 @@ +use super::*; +use rusqlite::{params, Connection}; + +struct Fixture { + dir: tempfile::TempDir, + state: Connection, + history: Connection, +} + +impl Fixture { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let state = Connection::open(dir.path().join("state_5.sqlite")).unwrap(); + state + .execute_batch( + "PRAGMA journal_mode=WAL; + CREATE TABLE threads (id TEXT PRIMARY KEY, title TEXT, name TEXT, cwd TEXT, + created_at INTEGER, updated_at INTEGER, rollout_path TEXT, + history_mode TEXT, source TEXT, archived INTEGER, first_user_message TEXT, + model_provider TEXT, model TEXT, cli_version TEXT);", + ) + .unwrap(); + let history = Connection::open(dir.path().join("thread_history_1.sqlite")).unwrap(); + history + .execute_batch( + "PRAGMA journal_mode=WAL; + CREATE TABLE thread_turns (thread_id TEXT, turn_id TEXT, rollout_ordinal INTEGER, + status TEXT, started_at INTEGER, error_json TEXT, + PRIMARY KEY(thread_id, turn_id)); + CREATE TABLE thread_items (thread_id TEXT, turn_id TEXT, item_id TEXT, + rollout_ordinal INTEGER, created_at_ms INTEGER, item_json TEXT, + PRIMARY KEY(thread_id, turn_id, item_id));", + ) + .unwrap(); + Self { + dir, + state, + history, + } + } + + fn reader(&self) -> CodexSessionReader { + CodexSessionReader::new(self.dir.path().into()) + } + + fn thread(&self, id: &str, source: &str, archived: bool) { + self.state + .execute( + "INSERT INTO threads VALUES (?1, ?2, NULL, '/project', + 1700000000, 1700000001, ?3, 'paginated', ?4, ?5, 'Original request', + 'openai', 'test-model', 'test-version')", + params![ + id, + format!("Saved {id}"), + self.dir + .path() + .join("sessions") + .join(format!("rollout-{id}.jsonl")) + .to_string_lossy(), + source, + archived + ], + ) + .unwrap(); + } + + fn turn(&self, thread: &str, id: &str, ordinal: i64, status: &str) { + self.history + .execute( + "INSERT INTO thread_turns VALUES (?1, ?2, ?3, ?4, 1700000000, NULL)", + params![thread, id, ordinal, status], + ) + .unwrap(); + } + + fn item(&self, thread: &str, turn: &str, ordinal: i64, item: Value) { + self.history + .execute( + "INSERT INTO thread_items VALUES (?1, ?2, ?3, ?4, 1700000000123, ?5)", + params![ + thread, + turn, + item["id"].as_str().unwrap(), + ordinal, + item.to_string() + ], + ) + .unwrap(); + } + + fn conversation(&self, id: &str) { + self.thread(id, "vscode", false); + self.turn(id, "turn-1", 0, "completed"); + self.item( + id, + "turn-1", + 1, + json!({"type":"userMessage", "id":"user-1", + "content":[{"type":"text","text":"Original request","text_elements":[]}]}), + ); + self.item( + id, + "turn-1", + 2, + json!({"type":"agentMessage", "id":"answer-1", "text":"Saved answer"}), + ); + } + + fn read(&self, reader: &CodexSessionReader, id: &str) -> CodexSessionDocument { + let catalog = reader.discover_all().unwrap(); + let candidate = catalog + .candidates + .iter() + .find(|candidate| candidate.source_session_id == id) + .expect("The SQLite-only thread must be discoverable"); + reader + .read_candidate(&candidate.candidate_id, ExternalSessionImportMode::Full) + .unwrap() + } +} + +#[test] +fn sqlite_only_threads_are_discovered_across_pages_without_reading_bodies() { + let fixture = Fixture::new(); + for index in 0..63 { + fixture.thread(&format!("thread-{index:03}"), "vscode", index == 0); + } + fixture.thread( + "child", + r#"{"subagent":{"thread_spawn":{"parent_thread_id":"thread-000"}}}"#, + false, + ); + // Discovery must not need a history table, including while a source is offline. + fixture + .history + .execute_batch("DROP TABLE thread_items; DROP TABLE thread_turns;") + .unwrap(); + let reader = fixture.reader(); + let catalog = reader.discover_all().unwrap(); + assert!(catalog.diagnostics.is_empty()); + assert_eq!(catalog.candidates.len(), 63); + assert_eq!(reader.discover(0, 50, "").unwrap().next_cursor, Some(50)); + assert_eq!(reader.discover(50, 50, "").unwrap().candidates.len(), 13); + let candidate = &reader + .discover(0, 50, "Saved thread-000") + .unwrap() + .candidates[0]; + assert_eq!(candidate.title.as_deref(), Some("Saved thread-000")); + assert!(candidate.title_is_source_name); + assert_eq!(candidate.cwd.as_deref(), Some("/project")); + assert_eq!(candidate.created_at, Some(1700000000000)); + assert_eq!(candidate.updated_at, Some(1700000001000)); + assert_eq!( + reader + .header_reads + .load(std::sync::atomic::Ordering::Relaxed), + 0 + ); +} + +#[test] +fn sqlite_history_reads_committed_wal_and_keeps_explicit_turn_status() { + let fixture = Fixture::new(); + fixture.conversation("one"); + fixture + .history + .execute("UPDATE thread_turns SET status = 'interrupted'", []) + .unwrap(); + let reader = fixture.reader(); + let document = fixture.read(&reader, "one"); + assert_eq!(document.turns.len(), 1); + assert_eq!(document.turns[0].user_text, "Original request"); + assert_eq!(document.turns[0].status, ExternalSessionStatus::Interrupted); + assert!( + matches!(&document.turns[0].items[0], ExternalSessionItem::Text { text } if text == "Saved answer") + ); + assert_eq!(document.candidate.model_name.as_deref(), Some("test-model")); + assert_eq!(document.candidate.title.as_deref(), Some("Saved one")); + let original = document.candidate.fingerprint; + fixture + .history + .execute_batch("BEGIN IMMEDIATE; UPDATE thread_turns SET status = 'completed';") + .unwrap(); + assert_eq!(fixture.read(&reader, "one").candidate.fingerprint, original); + fixture.history.execute_batch("COMMIT;").unwrap(); + let updated = fixture.read(&reader, "one"); + assert_eq!(updated.turns[0].status, ExternalSessionStatus::Complete); + assert_ne!(updated.candidate.fingerprint, original); + assert_eq!( + fixture.read(&reader, "one").candidate.fingerprint, + updated.candidate.fingerprint + ); +} + +#[test] +fn paginated_threads_replace_stale_rollouts_and_keep_deleted_file_locators_usable() { + let fixture = Fixture::new(); + fixture.conversation("one"); + let directory = fixture.dir.path().join("sessions"); + fs::create_dir(&directory).unwrap(); + let rollout = directory.join("rollout-one.jsonl"); + fs::write(&rollout, format!("{}\n{}\n", json!({"type":"session_meta","payload":{"id":"one","cwd":"/project"}}), + json!({"type":"response_item","payload":{"type":"message","role":"user","content":"Stale prompt"}}))).unwrap(); + let legacy_id = candidate_id(&rollout); + for present in [true, false] { + if !present { + fs::remove_file(&rollout).unwrap(); + } + let reader = fixture.reader(); + let catalog = reader.discover_all().unwrap(); + assert_eq!(catalog.candidates.len(), 1); + let document = reader + .read_candidate(&legacy_id, ExternalSessionImportMode::Full) + .unwrap(); + assert_eq!( + document.candidate.candidate_id, + catalog.candidates[0].candidate_id + ); + assert_eq!(document.turns[0].user_text, "Original request"); + assert_eq!(document.turns[0].items.len(), 1); + } +} + +#[test] +fn state_wal_refreshes_pagination_metadata_without_changing_source_files() { + let fixture = Fixture::new(); + fixture.conversation("one"); + let reader = fixture.reader(); + let original = reader.discover_all().unwrap(); + let metadata_reads = reader.titles.lock().unwrap().reads; + fixture + .state + .execute( + "UPDATE threads SET name = 'Renamed thread in WAL', updated_at = updated_at + 1", + [], + ) + .unwrap(); + let updated = reader.discover_all().unwrap(); + assert_eq!( + updated.candidates[0].title.as_deref(), + Some("Renamed thread in WAL") + ); + assert_ne!( + updated.candidates[0].fingerprint, + original.candidates[0].fingerprint + ); + assert_eq!(reader.titles.lock().unwrap().reads, metadata_reads + 1); + let snapshot = |path: &Path| fs::read(path).unwrap(); + let paths = [ + "state_5.sqlite", + "state_5.sqlite-wal", + "thread_history_1.sqlite", + "thread_history_1.sqlite-wal", + ] + .map(|name| fixture.dir.path().join(name)); + let before = paths.iter().map(|path| snapshot(path)).collect::>(); + fixture.read(&reader, "one"); + let after = paths.iter().map(|path| snapshot(path)).collect::>(); + assert_eq!( + before, after, + "Import must not write source database or WAL content" + ); +} + +#[test] +fn paginated_projection_preserves_order_thinking_tools_steering_and_attachments() { + let fixture = Fixture::new(); + fixture.thread("one", "cli", false); + // Insert out of order; source ordinals, not row insertion or ids, own order. + fixture.turn("one", "turn-last", 100, "inProgress"); + fixture.turn("one", "turn-first", 0, "completed"); + let items = vec![ + json!({"type":"userMessage","id":"u1","content":[{"type":"text","text":"Read the file"}]}), + json!({"type":"reasoning","id":"think","summary":["Check source"],"content":["Inspect the file first"]}), + json!({"type":"commandExecution","id":"cmd","command":"cat README.md","cwd":"/project", + "status":"completed","aggregatedOutput":"Contents","exitCode":0,"durationMs":50}), + json!({"type":"fileChange","id":"patch","status":"completed","changes":[{"path":"a.rs","kind":{"type":"update"},"diff":"-old\n+new"}]}), + json!({"type":"mcpToolCall","id":"mcp","server":"sample","tool":"lookup","arguments":{"q":"test"}, + "status":"failed","result":null,"error":{"message":"offline"}}), + json!({"type":"dynamicToolCall","id":"dynamic","namespace":"functions","tool":"exec","arguments":{"code":"1 + 1"}, + "status":"completed","success":true,"contentItems":[{"type":"inputText","text":"2"}]}), + json!({"type":"webSearch","id":"web","query":"example","action":{"type":"search"},"results":[{"title":"Example"}]}), + json!({"type":"imageView","id":"view","path":"/unread/image.png"}), + json!({"type":"sleep","id":"sleep","durationMs":500}), + json!({"type":"imageGeneration","id":"image","status":"completed","result":"/unread/generated.png"}), + json!({"type":"subAgentActivity","id":"child","kind":"spawn","agentThreadId":"child-thread"}), + json!({"type":"collabAgentToolCall","id":"collab","tool":"wait","status":"completed","agentsStates":{}}), + json!({"type":"agentMessage","id":"a1","text":"First answer","futureField":true}), + json!({"type":"userMessage","id":"u2","content":[{"type":"text","text":"Also look at this"},{"type":"localImage","path":"/unread/image.png"}]}), + json!({"type":"agentMessage","id":"a2","text":"Second answer"}), + ]; + for (index, item) in items.into_iter().enumerate().rev() { + fixture.item("one", "turn-first", index as i64, item); + } + fixture.item( + "one", + "turn-last", + 101, + json!({"type":"userMessage","id":"u3","content":[{"type":"text","text":"Next request"}]}), + ); + fixture.item( + "one", + "turn-last", + 102, + json!({"type":"agentMessage","id":"a3","text":"Still working"}), + ); + fixture.item( + "one", + "turn-last", + 103, + json!({"type":"imageGeneration","id":"image-pending", + "status":"generating","result":"","revisedPrompt":"In progress"}), + ); + let document = fixture.read(&fixture.reader(), "one"); + assert_eq!(document.turns.len(), 3); + assert_eq!(document.turns[0].source_turn_id, "turn-first"); + assert_eq!(document.turns[1].source_turn_id, "turn-first:u2"); + assert_eq!(document.turns[2].source_turn_id, "turn-last"); + assert_eq!(document.turns[2].status, ExternalSessionStatus::Interrupted); + assert!(matches!( + &document.turns[2].items[1], + ExternalSessionItem::Tool { success: None, .. } + )); + assert!( + matches!(&document.turns[0].items[0], ExternalSessionItem::Thinking { text } if text.contains("Check source") && text.contains("Inspect the file first")) + ); + assert!( + matches!(&document.turns[0].items[1], ExternalSessionItem::Tool { name, input, result:Some(result), success:Some(true), .. } + if name == "exec_command" && input["cmd"] == "cat README.md" && result["output"] == "Contents" && result["exit_code"] == 0) + ); + assert!( + matches!(&document.turns[0].items[3], ExternalSessionItem::Tool { name, result:Some(result), success:Some(false), .. } + if name == "mcp__sample__lookup" && result["error"]["message"] == "offline") + ); + assert!( + matches!(&document.turns[0].items[4], ExternalSessionItem::Tool { name, result:Some(result), .. } + if name == "functions.exec" && result[0]["type"] == "text" && result[0]["text"] == "2") + ); + assert!( + matches!(&document.turns[0].items[8], ExternalSessionItem::Tool { input, result:Some(result), .. } + if input.get("result").is_none() && result["result"] == "/unread/generated.png") + ); + assert!(document.turns[1].user_text.contains("Also look at this")); + assert!(document.turns[1].user_text.contains("/unread/image.png")); + assert!(document + .warnings + .iter() + .any(|warning| warning.code == "attachments_not_imported")); +} + +#[test] +fn unknown_corrupt_or_unavailable_paginated_history_never_falls_back_to_partial_jsonl() { + let fixture = Fixture::new(); + fixture.conversation("one"); + let reader = fixture.reader(); + let id = reader.discover_all().unwrap().candidates[0] + .candidate_id + .clone(); + fixture + .history + .execute( + "UPDATE thread_items SET item_json = 'broken json' WHERE item_id = 'answer-1'", + [], + ) + .unwrap(); + assert!(reader + .read_candidate(&id, ExternalSessionImportMode::Full) + .err() + .unwrap() + .contains("Malformed")); + fixture + .history + .execute( + "UPDATE thread_items SET item_json = ?1 WHERE item_id = 'answer-1'", + [json!({"id":"answer-1","type":"futureBehavior"}).to_string()], + ) + .unwrap(); + assert!(reader + .read_candidate(&id, ExternalSessionImportMode::Full) + .err() + .unwrap() + .contains("Unsupported Codex history item")); + fs::write(fixture.dir.path().join("thread_history_2.sqlite"), []).unwrap(); + assert!(reader + .read_candidate(&id, ExternalSessionImportMode::Full) + .err() + .unwrap() + .contains("database version")); + assert_eq!(reader.discover_all().unwrap().candidates.len(), 1); + + let missing = Fixture::new(); + missing.conversation("missing"); + let missing_id = missing.reader().discover_all().unwrap().candidates[0] + .candidate_id + .clone(); + drop(missing.history); + fs::remove_file(missing.dir.path().join("thread_history_1.sqlite")).unwrap(); + let reader = CodexSessionReader::new(missing.dir.path().into()); + assert!(reader + .read_candidate(&missing_id, ExternalSessionImportMode::Full) + .err() + .unwrap() + .contains("unavailable")); + assert_eq!(reader.discover_all().unwrap().candidates.len(), 1); +} + +#[test] +fn compaction_requires_full_history_and_orphan_items_are_reported() { + let fixture = Fixture::new(); + fixture.conversation("one"); + fixture.item( + "one", + "turn-1", + 3, + json!({"type":"contextCompaction","id":"compact"}), + ); + let reader = fixture.reader(); + let document = fixture.read(&reader, "one"); + assert!(document + .warnings + .iter() + .any(|warning| warning.code == "compaction_present")); + assert!(reader + .read_candidate( + &document.candidate.candidate_id, + ExternalSessionImportMode::LatestCompacted + ) + .err() + .unwrap() + .contains("select full history")); + fixture.item( + "one", + "missing-turn", + 10, + json!({"type":"agentMessage","id":"orphan","text":"Must not disappear"}), + ); + assert!(reader + .read_candidate( + &document.candidate.candidate_id, + ExternalSessionImportMode::Full + ) + .err() + .unwrap() + .contains("without a readable turn")); +} + +#[test] +#[ignore = "Read-only installed Codex catalog and history inspection, printing only aggregate counts"] +fn inspect_installed_codex_paginated_workspace() { + let workspace = std::env::var("OPENBITFUN_TEST_CODEX_WORKSPACE").ok(); + let normalize = |path: &str| { + let path = path.strip_prefix("\\\\?\\").unwrap_or(path); + if cfg!(windows) { + path.replace('\\', "/").trim_end_matches('/').to_lowercase() + } else { + path.trim_end_matches('/').to_owned() + } + }; + let reader = CodexSessionReader::default(); + let catalog = reader.discover_all().unwrap(); + assert!(catalog.diagnostics.is_empty()); + let candidates = catalog + .candidates + .iter() + .filter(|candidate| { + workspace.as_ref().is_none_or(|workspace| { + candidate + .cwd + .as_deref() + .is_some_and(|cwd| normalize(cwd) == normalize(workspace)) + }) + }) + .collect::>(); + // Compare the real state catalog to discovery independently of its cached + // metadata projection. A count alone cannot detect a missing/repeated ID. + let state_path = fs::read_dir(&reader.codex_home) + .unwrap() + .filter_map(|entry| { + let entry = entry.ok()?; + let name = entry.file_name(); + let name = name.to_str()?; + let version = name + .strip_prefix("state_")? + .strip_suffix(".sqlite")? + .parse::() + .ok()?; + Some((version, entry.path())) + }) + .max_by_key(|(version, _)| *version) + .expect("No installed Codex state database") + .1; + let connection = + Connection::open_with_flags(state_path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .unwrap(); + connection + .execute_batch("PRAGMA query_only = ON; BEGIN DEFERRED TRANSACTION;") + .unwrap(); + let source_rows = connection + .prepare("SELECT id, cwd, source FROM threads WHERE history_mode = 'paginated'") + .unwrap() + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + }) + .unwrap() + .collect::, _>>() + .unwrap(); + let source_ids = source_rows + .into_iter() + .filter(|(_, cwd, source)| { + let child = serde_json::from_str::(source) + .ok() + .is_some_and(|source| source.get("subagent").is_some()); + !child + && workspace + .as_ref() + .is_none_or(|workspace| normalize(cwd) == normalize(workspace)) + }) + .map(|(id, _, _)| id) + .collect::>(); + let discovered_ids = candidates + .iter() + .map(|candidate| candidate.source_session_id.clone()) + .collect::>(); + assert_eq!( + discovered_ids, source_ids, + "Every source main session must be discovered" + ); + assert_eq!( + candidates.len(), + source_ids.len(), + "No duplicate source sessions" + ); + let mut loaded = 0; + let mut empty = 0; + let mut failed = 0; + let mut turns = 0; + let mut tools = 0; + let mut warning_counts = std::collections::BTreeMap::::new(); + for candidate in &candidates { + match reader.read_candidate(&candidate.candidate_id, ExternalSessionImportMode::Full) { + Ok(document) => { + loaded += 1; + turns += document.turns.len(); + tools += document + .turns + .iter() + .flat_map(|turn| &turn.items) + .filter(|item| matches!(item, ExternalSessionItem::Tool { .. })) + .count(); + for warning in document.warnings { + *warning_counts.entry(warning.code).or_default() += 1; + } + } + Err(error) if error.starts_with("No usable human conversation") => empty += 1, + Err(error) => { + failed += 1; + eprintln!("Codex source read error: {error}"); + } + } + } + println!("Codex source inspection: {} source main sessions matched by ID, {loaded} readable, {empty} empty, {failed} failed, {turns} turns, {tools} tool items; warnings: {warning_counts:?}", candidates.len()); + assert!(!candidates.is_empty()); + assert_eq!(failed, 0); +} diff --git a/src/crates/adapters/codex-adapter/src/session/titles.rs b/src/crates/adapters/codex-adapter/src/session/titles.rs index b0c3e5acb5..2deb6bb1f9 100644 --- a/src/crates/adapters/codex-adapter/src/session/titles.rs +++ b/src/crates/adapters/codex-adapter/src/session/titles.rs @@ -1,5 +1,5 @@ -//! Codex keeps display names outside its rollout files. Read one small, -//! cached metadata snapshot per discovery, including committed WAL records. +//! Read one cached state/index metadata snapshot, including committed WAL +//! records. Paginated threads have no rollout file to discover on disk. use super::*; use rusqlite::{Connection, OpenFlags}; use std::time::Duration; @@ -12,6 +12,7 @@ pub(super) struct TitleSnapshot { indexed_names: HashMap, titles: HashMap, rollout_paths: HashMap, + pub(super) paginated: HashMap, pub(super) diagnostics: Vec, } @@ -237,9 +238,45 @@ fn read_database(path: &Path, snapshot: &mut TitleSnapshot) -> Result<(), String } else { "NULL" }; + let optional = |name: &str| { + if columns.contains(name) { + name.to_owned() + } else { + "NULL".into() + } + }; + let millis = |name: &str| { + let ms = format!("{name}_ms"); + if columns.contains(&ms) { + format!( + "COALESCE({ms}, {})", + if columns.contains(name) { + format!("{name} * 1000") + } else { + "NULL".into() + } + ) + } else if columns.contains(name) { + format!("{name} * 1000") + } else { + "NULL".into() + } + }; + let metadata_columns = [ + optional("history_mode"), + optional("cwd"), + millis("created_at"), + millis("updated_at"), + optional("source"), + optional("first_user_message"), + optional("model_provider"), + optional("model"), + optional("cli_version"), + ] + .join(", "); let mut query = connection .prepare(&format!( - "SELECT id, {name}, {title}, {rollout_path} FROM threads ORDER BY id LIMIT ?1", + "SELECT id, {name}, {title}, {rollout_path}, {metadata_columns} FROM threads ORDER BY id LIMIT ?1", )) .map_err(read_error)?; let rows = query @@ -249,6 +286,15 @@ fn read_database(path: &Path, snapshot: &mut TitleSnapshot) -> Result<(), String row.get::<_, Option>(1)?, row.get::<_, Option>(2)?, row.get::<_, Option>(3)?, + row.get::<_, Option>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, Option>(6)?, + row.get::<_, Option>(7)?, + row.get::<_, Option>(8)?, + row.get::<_, Option>(9)?, + row.get::<_, Option>(10)?, + row.get::<_, Option>(11)?, + row.get::<_, Option>(12)?, )) }) .map_err(read_error)?; @@ -256,15 +302,70 @@ fn read_database(path: &Path, snapshot: &mut TitleSnapshot) -> Result<(), String if index >= MAX_FILES { return Err("Codex title catalog exceeds the discovery limit".into()); } - let (id, name, title, rollout_path) = row.map_err(read_error)?; - if let Some(path) = rollout_path.filter(|path| !path.trim().is_empty()) { - snapshot.rollout_paths.entry(id.clone()).or_insert(path); + let ( + id, + name, + title, + rollout_path, + history_mode, + cwd, + created_at, + updated_at, + source, + first_user_message, + model_provider, + model_name, + format_version, + ) = row.map_err(read_error)?; + if let Some(path) = rollout_path.as_ref().filter(|path| !path.trim().is_empty()) { + snapshot + .rollout_paths + .entry(id.clone()) + .or_insert_with(|| path.clone()); } if let Some(name) = name.filter(|name| !name.trim().is_empty()) { snapshot.names.entry(id.clone()).or_insert(name); } if let Some(title) = title.filter(|title| !title.trim().is_empty()) { - snapshot.titles.entry(id).or_insert(title); + snapshot.titles.entry(id.clone()).or_insert(title); + } + if history_mode.as_deref() == Some("paginated") { + let updated_at = updated_at.and_then(|time| u64::try_from(time).ok()); + let candidate = ExternalSessionCandidate { + candidate_id: format!("codex:thread:{id}"), + ecosystem_id: "codex".into(), + source_session_id: id.clone(), + title: first_user_message.as_deref().and_then(user_title_preview), + fallback_title: first_user_message.as_deref().map(|text| preview(text, 120)), + title_is_source_name: false, + cwd, + created_at: created_at.and_then(|time| u64::try_from(time).ok()), + updated_at, + model_provider, + model_name, + turn_count: None, + file_size_bytes: 0, + format_version, + status: ExternalSessionStatus::Complete, + fingerprint: format!("metadata:paginated:{}", updated_at.unwrap_or(0)), + source_location: Some(format!("$CODEX_HOME/thread_history_1.sqlite#{id}")), + target_workspace_id: None, + target_workspace_name: None, + imported_session_id: None, + }; + let subagent = source + .as_deref() + .and_then(|source| serde_json::from_str::(source).ok()) + .is_some_and(|source| source.get("subagent").is_some()); + snapshot + .paginated + .entry(id) + .or_insert(super::sqlite_history::Thread { + candidate, + rollout_path, + subagent, + directory: path.parent().unwrap_or_else(|| Path::new("")).to_path_buf(), + }); } } Ok(()) diff --git a/src/crates/adapters/zcode-adapter/src/session/tests.rs b/src/crates/adapters/zcode-adapter/src/session/tests.rs index 0794f1ad05..591972ba16 100644 --- a/src/crates/adapters/zcode-adapter/src/session/tests.rs +++ b/src/crates/adapters/zcode-adapter/src/session/tests.rs @@ -407,11 +407,48 @@ fn unsupported_schema_and_candidate_paths_do_not_return_empty_success() { #[ignore = "Reads the installed Zcode session database without changing it"] fn inspect_installed_zcode() { let reader = ZcodeSessionReader::default(); - let catalog = reader.discover(0, 50, "").unwrap(); + let catalog = reader.discover_all().unwrap(); assert!( !catalog.candidates.is_empty(), "No installed Zcode sessions found" ); + let (connection, _) = reader.open().unwrap().unwrap(); + let source_ids = connection + .prepare("SELECT id FROM session WHERE parent_id IS NULL") + .unwrap() + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::, _>>() + .unwrap(); + let discovered_ids = catalog + .candidates + .iter() + .map(|candidate| candidate.source_session_id.clone()) + .collect::>(); + assert_eq!( + discovered_ids, source_ids, + "Every source main session must be discovered" + ); + assert_eq!( + catalog.candidates.len(), + source_ids.len(), + "No duplicate source sessions" + ); + let mut paged_ids = Vec::new(); + let mut cursor = 0; + loop { + let page = reader.discover(cursor, 50, "").unwrap(); + paged_ids.extend( + page.candidates + .into_iter() + .map(|candidate| candidate.source_session_id), + ); + let Some(next) = page.next_cursor else { break }; + assert!(next > cursor); + cursor = next; + } + assert_eq!(paged_ids.len(), source_ids.len()); + assert_eq!(paged_ids.into_iter().collect::>(), source_ids); let mut turns = 0; let mut tools = 0; for candidate in &catalog.candidates { @@ -427,7 +464,7 @@ fn inspect_installed_zcode() { .count(); } println!( - "Zcode read-only import: {} sessions, {turns} user turns, {tools} tool calls", + "Zcode read-only import: {} source main sessions matched by ID and pagination, {turns} user turns, {tools} tool calls", catalog.candidates.len() ); }