use crate::{Error, Result};
use ccql::datasources::transcript::{
discover_transcript_files, flattened_usage_fields, SessionAggregate, TranscriptFile,
};
use chrono::DateTime;
use rusqlite::{params, Connection, ErrorCode, OptionalExtension, TransactionBehavior};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs::{self, File, Metadata};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::time::{Duration, UNIX_EPOCH};
pub struct UnifiedEngine {
conn: Connection,
claude_data_dir: PathBuf,
codex_data_dir: PathBuf,
git_repo_path: PathBuf,
claude_cache_attached: bool,
claude_sessions_loaded: bool,
claude_tool_calls_loaded: bool,
codex_loaded: bool,
grok_data_dir: PathBuf,
grok_loaded: bool,
macos_log_stats: Option<crate::providers::macos_logs::MacosLogStats>,
}
impl UnifiedEngine {
pub fn new(claude_data_dir: PathBuf, git_repo_path: PathBuf) -> Result<Self> {
Self::new_with_codex_data_dir(claude_data_dir, git_repo_path, default_codex_data_dir())
}
pub fn new_with_codex_data_dir(
claude_data_dir: PathBuf,
git_repo_path: PathBuf,
codex_data_dir: PathBuf,
) -> Result<Self> {
let conn = Connection::open_in_memory()?;
conn.pragma_update(None, "temp_store", "MEMORY")?;
conn.create_scalar_function(
"DATE",
1,
rusqlite::functions::FunctionFlags::SQLITE_DETERMINISTIC,
|ctx| {
let value: String = ctx.get(0)?;
Ok(normalize_date(&value))
},
)?;
Ok(Self {
conn,
claude_data_dir,
codex_data_dir,
git_repo_path,
claude_cache_attached: false,
claude_sessions_loaded: false,
claude_tool_calls_loaded: false,
codex_loaded: false,
grok_data_dir: default_grok_data_dir(),
grok_loaded: false,
macos_log_stats: None,
})
}
pub fn load_claude_tables(&mut self, tables: &[&str]) -> Result<()> {
for table in tables {
match *table {
"history" => self.load_history()?,
"jhistory" | "codex_history" => self.load_jhistory()?,
"transcripts" => self.load_transcripts()?,
"sessions" => self.load_sessions()?,
"todos" => self.load_todos()?,
"tool_calls" => self.load_tool_calls()?,
"codex_tool_calls" => self.load_codex_tool_calls()?,
"codex_threads"
| "codex_events"
| "codex_messages"
| "codex_tool_executions"
| "codex_compactions"
| "codex_ingest_errors" => self.load_codex_tables()?,
_ => {}
}
}
Ok(())
}
pub fn load_git_tables(&mut self, tables: &[&str]) -> Result<()> {
for table in tables {
match *table {
"commits" => self.load_commits()?,
"diffs" => self.load_diffs()?,
"diff_files" => self.load_diff_files()?,
"branches" => self.load_branches()?,
_ => {}
}
}
Ok(())
}
pub fn load_code_tables(&mut self, tables: &[&str]) -> Result<()> {
crate::providers::load_all_code_tables(&self.conn, &self.git_repo_path, tables)
}
pub fn load_shell_history(&mut self) -> Result<()> {
crate::providers::shell_history::load(&mut self.conn)
}
pub fn load_macos_logs(
&mut self,
config: crate::providers::macos_logs::MacosLogConfig,
) -> Result<()> {
self.macos_log_stats = Some(crate::providers::macos_logs::register(&self.conn, config)?);
Ok(())
}
pub fn macos_log_truncation(&self) -> Option<(usize, String)> {
self.macos_log_stats.as_ref()?.truncation()
}
pub fn load_command_events(&mut self) -> Result<()> {
self.load_shell_history()?;
if !self.claude_tool_calls_loaded {
self.load_tool_calls()?;
}
if !self.codex_loaded {
self.load_codex_tables()?;
}
self.conn.execute_batch(
"CREATE TEMP VIEW IF NOT EXISTS command_events (
source, channel, actor, provenance_quality, provenance_reason,
source_id, source_order, session_id, parent_session_id, agent_id,
agent_role, originator, tool_name, timestamp, duration_ms,
exit_code, command, cwd, hostname, source_path
) AS
SELECT source, 'shell', 'unknown', 'unattributed',
'unattributed_shell_history', source_id, source_order,
session_id, NULL, NULL, NULL, NULL, NULL, timestamp,
duration_ms, exit_code, command, cwd, hostname, history_path
FROM shell_history
UNION ALL
SELECT 'claude', 'agent_tool', 'agent', 'exact', NULL,
source_id, rowid, session_id, parent_session_id, agent_id,
agent_role, originator, tool_name, timestamp, NULL, NULL,
command, cwd, NULL, source_path
FROM tool_calls
WHERE tool_name = 'Bash' AND command IS NOT NULL
UNION ALL
SELECT 'codex', 'agent_tool', 'agent', 'exact', NULL,
execution.call_id, execution.call_record_index,
execution.thread_id, thread.parent_thread_id, thread.agent_path,
thread.agent_role, thread.originator, execution.tool_name,
execution.called_at, NULL, execution.exit_code, execution.cmd,
COALESCE(execution.cwd, thread.cwd), NULL, execution.source_path
FROM codex_tool_executions AS execution
LEFT JOIN codex_threads AS thread
ON thread.thread_id = execution.thread_id
WHERE execution.tool_name IN ('exec_command', 'shell')
AND execution.cmd IS NOT NULL;",
)?;
Ok(())
}
pub fn load_work_tables(&mut self, tables: &[&str]) -> Result<()> {
if tables.is_empty() {
return Ok(());
}
let wl = crate::worklog::Worklog::open()?;
wl.materialize_into(&self.conn, tables)
}
pub(crate) fn conn(&self) -> &Connection {
&self.conn
}
pub fn query(&self, sql: &str) -> Result<Vec<Value>> {
let mut stmt = self.conn.prepare(sql)?;
let column_names: Vec<String> = stmt
.column_names()
.into_iter()
.map(|s| s.to_string())
.collect();
let rows = stmt.query_map([], |row| {
let mut obj = serde_json::Map::new();
for (i, name) in column_names.iter().enumerate() {
let value: Value = if let Ok(v) = row.get::<_, i64>(i) {
Value::Number(v.into())
} else if let Ok(v) = row.get::<_, f64>(i) {
serde_json::Number::from_f64(v)
.map(Value::Number)
.unwrap_or(Value::Null)
} else if let Ok(v) = row.get::<_, String>(i) {
Value::String(v)
} else {
Value::Null
};
obj.insert(name.clone(), value);
}
Ok(Value::Object(obj))
})?;
Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}
fn load_history(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS history (
rowid INTEGER PRIMARY KEY,
display TEXT,
timestamp TEXT,
project TEXT
)",
[],
)?;
let history_path = self.claude_data_dir.join("history.jsonl");
if history_path.exists() {
let content = std::fs::read_to_string(&history_path)?;
for line in content.lines() {
if let Ok(entry) = serde_json::from_str::<Value>(line) {
let display = entry.get("display").and_then(|v| v.as_str()).unwrap_or("");
let timestamp = entry
.get("timestamp")
.map(|v| v.to_string())
.unwrap_or_default();
let project = entry.get("project").and_then(|v| v.as_str()).unwrap_or("");
self.conn.execute(
"INSERT INTO history (display, timestamp, project) VALUES (?1, ?2, ?3)",
params![display, timestamp, project],
)?;
}
}
}
self.conn.execute_batch(
"CREATE INDEX IF NOT EXISTS idx_history_timestamp ON history(timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_history_project_timestamp
ON history(project, timestamp DESC);",
)?;
Ok(())
}
fn ccql_config(&self) -> Option<ccql::Config> {
ccql::Config::new_with_codex_data_dir(
self.claude_data_dir.clone(),
self.codex_data_dir.clone(),
)
.ok()
}
fn load_transcripts(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS transcripts (
rowid INTEGER PRIMARY KEY,
type TEXT,
content TEXT,
tool_name TEXT,
session_id TEXT,
_source_file TEXT,
_session_id TEXT,
_project TEXT,
_agent_id TEXT,
timestamp TEXT,
model TEXT,
usage_input_tokens INTEGER,
usage_output_tokens INTEGER,
usage_cache_read_input_tokens INTEGER,
usage_cache_creation_input_tokens INTEGER,
usage_ephemeral_5m_input_tokens INTEGER,
usage_ephemeral_1h_input_tokens INTEGER,
usage_service_tier TEXT
)",
[],
)?;
let Some(config) = self.ccql_config() else {
return Ok(());
};
let tx = self.conn.transaction()?;
{
let mut stmt = tx.prepare(
"INSERT INTO transcripts (type, content, tool_name, session_id,
_source_file, _session_id, _project, _agent_id, timestamp,
model, usage_input_tokens, usage_output_tokens,
usage_cache_read_input_tokens, usage_cache_creation_input_tokens,
usage_ephemeral_5m_input_tokens, usage_ephemeral_1h_input_tokens,
usage_service_tier)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)",
)?;
for file in discover_transcript_files(&config) {
let content = match std::fs::read_to_string(&file.path) {
Ok(c) => c,
Err(_) => continue,
};
for line in content.lines() {
let Ok(entry) = serde_json::from_str::<Value>(line) else {
continue;
};
let msg_type = entry.get("type").and_then(|v| v.as_str()).unwrap_or("");
let msg_content = entry
.get("content")
.and_then(|v| v.as_str())
.or_else(|| entry.get("message").and_then(|v| v.as_str()))
.unwrap_or("");
let tool_name = entry
.get("tool_name")
.and_then(|v| v.as_str())
.unwrap_or("");
let timestamp = entry.get("timestamp").and_then(|v| v.as_str());
let usage: HashMap<&str, &Value> =
flattened_usage_fields(&entry).into_iter().collect();
let usage_int = |key: &str| usage.get(key).and_then(|v| v.as_i64());
stmt.execute(params![
msg_type,
msg_content,
tool_name,
file.session_id,
file.source_file,
file.session_id,
file.project,
file.agent_id,
timestamp,
usage.get("model").and_then(|v| v.as_str()),
usage_int("usage_input_tokens"),
usage_int("usage_output_tokens"),
usage_int("usage_cache_read_input_tokens"),
usage_int("usage_cache_creation_input_tokens"),
usage_int("usage_ephemeral_5m_input_tokens"),
usage_int("usage_ephemeral_1h_input_tokens"),
usage.get("usage_service_tier").and_then(|v| v.as_str()),
])?;
}
}
}
tx.commit()?;
Ok(())
}
fn load_tool_calls(&mut self) -> Result<()> {
if self.claude_tool_calls_loaded {
return Ok(());
}
let Some(config) = self.ccql_config() else {
create_empty_tool_calls_table(&self.conn)?;
self.claude_tool_calls_loaded = true;
return Ok(());
};
let files = discover_transcript_files(&config);
self.ensure_claude_cache(&files, false)?;
self.create_tool_calls_view()
}
pub(crate) fn load_current_claude_tool_calls(&mut self) -> Result<()> {
if self.claude_tool_calls_loaded {
return Ok(());
}
if self.ccql_config().is_none() {
create_empty_tool_calls_table(&self.conn)?;
self.claude_tool_calls_loaded = true;
return Ok(());
}
let cache_path = claude_tool_cache_path(&self.claude_data_dir);
self.attach_claude_cache(&cache_path)?;
self.create_tool_calls_view()
}
fn create_tool_calls_view(&mut self) -> Result<()> {
self.conn.execute_batch(
"CREATE TEMP VIEW tool_calls AS
SELECT rowid, tool_name, input_json, target, source_id, command,
session_id, parent_session_id, agent_id, agent_role,
originator, cwd, source_path, project AS _project, timestamp
FROM claude_tool_index.tool_calls;",
)?;
self.claude_tool_calls_loaded = true;
Ok(())
}
fn ensure_claude_cache(
&mut self,
files: &[TranscriptFile],
include_sessions: bool,
) -> Result<()> {
let cache_path = claude_tool_cache_path(&self.claude_data_dir);
let mut cache = open_claude_tool_cache(&cache_path)?;
sync_claude_tool_cache(&mut cache, files, include_sessions)?;
drop(cache);
self.attach_claude_cache(&cache_path)
}
fn attach_claude_cache(&mut self, cache_path: &Path) -> Result<()> {
if !self.claude_cache_attached {
self.conn.execute(
"ATTACH DATABASE ?1 AS claude_tool_index",
[cache_path.to_string_lossy().as_ref()],
)?;
self.claude_cache_attached = true;
}
Ok(())
}
fn load_codex_tool_calls(&mut self) -> Result<()> {
self.load_codex_tables()
}
fn load_codex_tables(&mut self) -> Result<()> {
if self.codex_loaded {
return Ok(());
}
let mut index = crate::codex_index::CodexIndex::open(&self.codex_data_dir)?;
index.sync()?;
let cache_path = index.cache_path().to_string_lossy().into_owned();
drop(index);
self.attach_codex_tables(cache_path)
}
pub(crate) fn load_current_codex_tables(&mut self) -> Result<()> {
if self.codex_loaded {
return Ok(());
}
let index = crate::codex_index::CodexIndex::open(&self.codex_data_dir)?;
let cache_path = index.cache_path().to_string_lossy().into_owned();
drop(index);
self.attach_codex_tables(cache_path)
}
fn attach_codex_tables(&mut self, cache_path: String) -> Result<()> {
self.conn
.execute("ATTACH DATABASE ?1 AS codex_index", [cache_path])?;
self.conn.execute_batch(
"
CREATE TEMP VIEW codex_threads AS
SELECT * FROM codex_index.codex_threads;
CREATE TEMP VIEW codex_events AS
SELECT * FROM codex_index.codex_events;
CREATE TEMP VIEW codex_messages AS
SELECT * FROM codex_index.codex_messages;
CREATE TEMP VIEW codex_tool_executions AS
SELECT * FROM codex_index.codex_tool_executions;
CREATE TEMP VIEW codex_compactions AS
SELECT * FROM codex_index.codex_compactions;
CREATE TEMP VIEW codex_ingest_errors AS
SELECT * FROM codex_index.codex_ingest_errors;
CREATE TEMP VIEW codex_tool_calls AS
SELECT
row_number() OVER (
ORDER BY execution.thread_id, execution.call_record_index
) AS rowid,
execution.tool_name,
execution.arguments_json,
execution.cmd,
execution.call_id AS source_id,
execution.thread_id AS session_id,
thread.parent_thread_id AS parent_session_id,
thread.agent_path AS agent_id,
thread.agent_role,
thread.originator,
COALESCE(execution.cwd, thread.cwd) AS cwd,
execution.source_path,
execution.called_at AS timestamp
FROM codex_index.codex_tool_executions AS execution
LEFT JOIN codex_index.codex_threads AS thread
ON thread.thread_id = execution.thread_id
WHERE execution.call_record_index IS NOT NULL;
",
)?;
self.codex_loaded = true;
Ok(())
}
pub fn load_grok_tables(&mut self, tables: &[&str]) -> Result<()> {
if tables.is_empty() {
return Ok(());
}
if self.grok_loaded {
return Ok(());
}
let mut index = crate::grok_index::GrokIndex::open(&self.grok_data_dir)?;
index.sync()?;
let cache_path = index.cache_path().to_string_lossy().into_owned();
drop(index);
self.attach_grok_tables(cache_path)
}
fn attach_grok_tables(&mut self, cache_path: String) -> Result<()> {
self.conn
.execute("ATTACH DATABASE ?1 AS grok_index", [cache_path])?;
self.conn.execute_batch(
"
CREATE TEMP VIEW grok_bots AS
SELECT * FROM grok_index.grok_bots;
CREATE TEMP VIEW grok_entries AS
SELECT * FROM grok_index.grok_entries;
CREATE TEMP VIEW grok_ingest_errors AS
SELECT * FROM grok_index.grok_ingest_errors;
CREATE TEMP VIEW grok_messages AS
SELECT
entry.bot_id,
bot.name AS bot_name,
entry.entry_id,
entry.kind,
entry.role,
entry.direction,
entry.message_type,
entry.text,
entry.from_agent,
entry.to_agent,
entry.author,
entry.request_id,
entry.timestamp,
entry.timestamp_ms,
entry.provenance,
entry.source_order,
entry.source_path
FROM grok_index.grok_entries AS entry
LEFT JOIN grok_index.grok_bots AS bot
ON bot.bot_id = entry.bot_id
WHERE entry.text IS NOT NULL AND entry.text != '';
",
)?;
self.grok_loaded = true;
Ok(())
}
fn load_sessions(&mut self) -> Result<()> {
if self.claude_sessions_loaded {
return Ok(());
}
let Some(config) = self.ccql_config() else {
create_empty_sessions_table(&self.conn)?;
self.claude_sessions_loaded = true;
return Ok(());
};
let files = discover_transcript_files(&config);
self.ensure_claude_cache(&files, true)?;
self.conn.execute_batch(
"CREATE TEMP VIEW sessions AS
SELECT session_id, project, cwd, git_branch, version, title,
first_timestamp, last_timestamp, user_message_count,
assistant_message_count, subagent_count, total_input_tokens,
total_output_tokens, total_cache_read_input_tokens,
total_cache_creation_input_tokens, pr_url, pr_number
FROM claude_tool_index.sessions;",
)?;
self.claude_sessions_loaded = true;
Ok(())
}
fn load_jhistory(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS jhistory (
rowid INTEGER PRIMARY KEY,
session_id TEXT,
ts INTEGER,
text TEXT,
display TEXT,
timestamp INTEGER
)",
[],
)?;
self.conn.execute(
"CREATE VIEW IF NOT EXISTS codex_history AS SELECT * FROM jhistory",
[],
)?;
let jhistory_path = self.codex_data_dir.join("history.jsonl");
if !jhistory_path.exists() {
return Ok(());
}
let content = std::fs::read_to_string(&jhistory_path)?;
for line in content.lines() {
if let Ok(entry) = serde_json::from_str::<Value>(line) {
let text = entry
.get("text")
.or_else(|| entry.get("display"))
.and_then(json_value_as_string)
.unwrap_or_default();
let session_id = entry
.get("session_id")
.or_else(|| entry.get("sessionId"))
.and_then(json_value_as_string)
.unwrap_or_default();
let ts = entry
.get("ts")
.and_then(json_number_as_i64)
.or_else(|| {
entry
.get("timestamp")
.and_then(json_number_as_i64)
.map(normalize_ts_seconds)
})
.unwrap_or(0);
let timestamp = ts.saturating_mul(1000);
self.conn.execute(
"INSERT INTO jhistory (session_id, ts, text, display, timestamp)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![session_id, ts, text.clone(), text, timestamp],
)?;
}
}
self.conn.execute_batch(
"CREATE INDEX IF NOT EXISTS idx_jhistory_timestamp ON jhistory(timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_jhistory_session_timestamp
ON jhistory(session_id, timestamp DESC);",
)?;
Ok(())
}
fn load_todos(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS todos (
rowid INTEGER PRIMARY KEY,
content TEXT,
status TEXT
)",
[],
)?;
let todos_dir = self.claude_data_dir.join("todos");
if !todos_dir.is_dir() {
return Ok(());
}
let entries = match std::fs::read_dir(&todos_dir) {
Ok(e) => e,
Err(_) => return Ok(()),
};
for entry in entries.filter_map(|e| e.ok()) {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let content = match std::fs::read_to_string(&path) {
Ok(c) => c,
Err(_) => continue,
};
if let Ok(items) = serde_json::from_str::<Vec<Value>>(&content) {
for item in &items {
let todo_content = item.get("content").and_then(|v| v.as_str()).unwrap_or("");
let status = item.get("status").and_then(|v| v.as_str()).unwrap_or("");
self.conn.execute(
"INSERT INTO todos (content, status) VALUES (?1, ?2)",
params![todo_content, status],
)?;
}
}
}
self.conn.execute(
"CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status)",
[],
)?;
Ok(())
}
fn load_commits(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS commits (
id TEXT PRIMARY KEY,
short_id TEXT,
author_name TEXT,
author_email TEXT,
authored_at TEXT,
summary TEXT,
message TEXT,
is_merge INTEGER
)",
[],
)?;
if let Ok(repo) = git2::Repository::open(&self.git_repo_path) {
let mut revwalk = repo.revwalk().map_err(|e| Error::Vcsql(e.to_string()))?;
revwalk.push_head().ok();
for oid in revwalk.filter_map(|r| r.ok()) {
if let Ok(commit) = repo.find_commit(oid) {
let id = commit.id().to_string();
let short_id = &id[..7.min(id.len())];
let author = commit.author();
let author_name = author.name().unwrap_or("");
let author_email = author.email().unwrap_or("");
let time = commit.time();
let authored_at = format_git_time(time.seconds());
let summary = commit.summary().unwrap_or("");
let message = commit.message().unwrap_or("");
let is_merge = if commit.parent_count() > 1 { 1 } else { 0 };
self.conn.execute(
"INSERT OR IGNORE INTO commits VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
id,
short_id,
author_name,
author_email,
authored_at,
summary,
message,
is_merge
],
)?;
}
}
}
self.conn.execute(
"CREATE INDEX IF NOT EXISTS idx_commits_authored_at ON commits(authored_at DESC)",
[],
)?;
Ok(())
}
fn load_diffs(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS diffs (
commit_id TEXT PRIMARY KEY,
files_changed INTEGER,
insertions INTEGER,
deletions INTEGER
)",
[],
)?;
if let Ok(repo) = git2::Repository::open(&self.git_repo_path) {
let mut revwalk = repo.revwalk().map_err(|e| Error::Vcsql(e.to_string()))?;
revwalk.push_head().ok();
for oid in revwalk.filter_map(|r| r.ok()) {
let Ok(commit) = repo.find_commit(oid) else {
continue;
};
let commit_tree = match commit.tree() {
Ok(t) => t,
Err(_) => continue,
};
let parent_tree = if commit.parent_count() > 0 {
commit.parent(0).ok().and_then(|p| p.tree().ok())
} else {
None
};
let diff =
match repo.diff_tree_to_tree(parent_tree.as_ref(), Some(&commit_tree), None) {
Ok(d) => d,
Err(_) => continue,
};
let stats = match diff.stats() {
Ok(s) => s,
Err(_) => continue,
};
let commit_id = commit.id().to_string();
self.conn.execute(
"INSERT OR IGNORE INTO diffs VALUES (?1, ?2, ?3, ?4)",
params![
commit_id,
stats.files_changed() as i64,
stats.insertions() as i64,
stats.deletions() as i64,
],
)?;
}
}
Ok(())
}
fn load_diff_files(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS diff_files (
commit_id TEXT,
path TEXT,
status TEXT,
insertions INTEGER,
deletions INTEGER
)",
[],
)?;
if let Ok(repo) = git2::Repository::open(&self.git_repo_path) {
let mut revwalk = repo.revwalk().map_err(|e| Error::Vcsql(e.to_string()))?;
revwalk.push_head().ok();
for oid in revwalk.filter_map(|r| r.ok()) {
let Ok(commit) = repo.find_commit(oid) else {
continue;
};
let commit_tree = match commit.tree() {
Ok(t) => t,
Err(_) => continue,
};
let parent_tree = if commit.parent_count() > 0 {
commit.parent(0).ok().and_then(|p| p.tree().ok())
} else {
None
};
let diff =
match repo.diff_tree_to_tree(parent_tree.as_ref(), Some(&commit_tree), None) {
Ok(d) => d,
Err(_) => continue,
};
let commit_id = commit.id().to_string();
for delta_idx in 0..diff.deltas().len() {
let delta = diff.deltas().nth(delta_idx).unwrap();
let path = delta
.new_file()
.path()
.or_else(|| delta.old_file().path())
.map(|p| p.to_string_lossy().to_string())
.unwrap_or_default();
let status = match delta.status() {
git2::Delta::Added => "A",
git2::Delta::Deleted => "D",
git2::Delta::Modified => "M",
git2::Delta::Renamed => "R",
git2::Delta::Copied => "C",
_ => "?",
};
let (insertions, deletions) =
if let Ok(Some(ref p)) = git2::Patch::from_diff(&diff, delta_idx) {
let (_, adds, dels) = p.line_stats().unwrap_or((0, 0, 0));
(adds as i64, dels as i64)
} else {
(0i64, 0i64)
};
self.conn.execute(
"INSERT INTO diff_files VALUES (?1, ?2, ?3, ?4, ?5)",
params![commit_id, path, status, insertions, deletions],
)?;
}
}
}
Ok(())
}
fn load_branches(&mut self) -> Result<()> {
self.conn.execute(
"CREATE TABLE IF NOT EXISTS branches (
name TEXT PRIMARY KEY,
target TEXT,
is_head INTEGER,
is_remote INTEGER
)",
[],
)?;
if let Ok(repo) = git2::Repository::open(&self.git_repo_path) {
if let Ok(branches) = repo.branches(None) {
for branch in branches.filter_map(|b| b.ok()) {
let (branch, branch_type) = branch;
let name = branch.name().ok().flatten().unwrap_or("");
let target = branch
.get()
.target()
.map(|t| t.to_string())
.unwrap_or_default();
let is_head = if branch.is_head() { 1 } else { 0 };
let is_remote = if branch_type == git2::BranchType::Remote {
1
} else {
0
};
self.conn.execute(
"INSERT OR IGNORE INTO branches VALUES (?1, ?2, ?3, ?4)",
params![name, target, is_head, is_remote],
)?;
}
}
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ClaudeCacheFileState {
size: u64,
modified_ns: i64,
}
fn create_empty_tool_calls_table(conn: &Connection) -> Result<()> {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS tool_calls (
rowid INTEGER PRIMARY KEY,
tool_name TEXT,
input_json TEXT,
target TEXT,
source_id TEXT,
command TEXT,
session_id TEXT,
parent_session_id TEXT,
agent_id TEXT,
agent_role TEXT,
originator TEXT,
cwd TEXT,
source_path TEXT,
_project TEXT,
timestamp TEXT
);",
)?;
Ok(())
}
fn create_empty_sessions_table(conn: &Connection) -> Result<()> {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT,
project TEXT,
cwd TEXT,
git_branch TEXT,
version TEXT,
title TEXT,
first_timestamp TEXT,
last_timestamp TEXT,
user_message_count INTEGER,
assistant_message_count INTEGER,
subagent_count INTEGER,
total_input_tokens INTEGER,
total_output_tokens INTEGER,
total_cache_read_input_tokens INTEGER,
total_cache_creation_input_tokens INTEGER,
pr_url TEXT,
pr_number INTEGER
);",
)?;
Ok(())
}
fn claude_tool_cache_path(claude_data_dir: &Path) -> PathBuf {
let cache_root = dirs::cache_dir()
.unwrap_or_else(|| std::env::temp_dir().join("devsql-cache"))
.join("devsql")
.join("claude-tool-index");
let canonical_dir = claude_data_dir
.canonicalize()
.unwrap_or_else(|_| claude_data_dir.to_path_buf());
let digest = Sha256::digest(canonical_dir.to_string_lossy().as_bytes());
cache_root.join(format!("{digest:x}.sqlite"))
}
fn open_claude_tool_cache(path: &Path) -> Result<Connection> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
set_private_directory_permissions(parent)?;
}
let mut conn = Connection::open(path)?;
set_private_file_permissions(path)?;
conn.busy_timeout(Duration::from_secs(30))?;
if !claude_cache_schema_ready(&conn)? {
conn.execute_batch(
"PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
CREATE TABLE IF NOT EXISTS source_files (
source_path TEXT PRIMARY KEY,
size INTEGER NOT NULL,
modified_ns INTEGER NOT NULL,
tail_hash TEXT
);
CREATE TABLE IF NOT EXISTS session_source_files (
source_path TEXT PRIMARY KEY,
size INTEGER NOT NULL,
modified_ns INTEGER NOT NULL,
tail_hash TEXT
);
CREATE TABLE IF NOT EXISTS tool_calls (
rowid INTEGER PRIMARY KEY,
tool_name TEXT,
input_json TEXT,
target TEXT,
source_id TEXT,
command TEXT,
session_id TEXT,
parent_session_id TEXT,
agent_id TEXT,
agent_role TEXT,
originator TEXT,
cwd TEXT,
source_path TEXT NOT NULL,
project TEXT,
timestamp TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
source_path TEXT PRIMARY KEY,
session_id TEXT,
project TEXT,
cwd TEXT,
git_branch TEXT,
version TEXT,
title TEXT,
first_timestamp TEXT,
last_timestamp TEXT,
user_message_count INTEGER,
assistant_message_count INTEGER,
subagent_count INTEGER,
total_input_tokens INTEGER,
total_output_tokens INTEGER,
total_cache_read_input_tokens INTEGER,
total_cache_creation_input_tokens INTEGER,
pr_url TEXT,
pr_number INTEGER
);
CREATE INDEX IF NOT EXISTS idx_claude_tool_calls_source
ON tool_calls(source_path);
CREATE INDEX IF NOT EXISTS idx_claude_tool_calls_tool_timestamp
ON tool_calls(tool_name, timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_claude_tool_calls_session_timestamp
ON tool_calls(session_id, timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_claude_tool_calls_timestamp
ON tool_calls(timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_claude_sessions_last_timestamp
ON sessions(last_timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_claude_sessions_project_last_timestamp
ON sessions(project, last_timestamp DESC);",
)?;
ensure_cache_tail_hash_columns(&conn)?;
backfill_cache_tail_hashes(&mut conn)?;
}
set_private_cache_sidecar_permissions(path)?;
Ok(conn)
}
fn claude_cache_schema_ready(conn: &Connection) -> Result<bool> {
let tables: i64 = conn.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type = 'table'
AND name IN ('source_files', 'session_source_files', 'tool_calls', 'sessions')",
[],
|row| row.get(0),
)?;
let indexes: i64 = conn.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type = 'index'
AND name IN (
'idx_claude_tool_calls_source',
'idx_claude_tool_calls_tool_timestamp',
'idx_claude_tool_calls_session_timestamp',
'idx_claude_tool_calls_timestamp',
'idx_claude_sessions_last_timestamp',
'idx_claude_sessions_project_last_timestamp'
)",
[],
|row| row.get(0),
)?;
Ok(tables == 4
&& indexes == 6
&& table_has_column(conn, "source_files", "tail_hash")?
&& table_has_column(conn, "session_source_files", "tail_hash")?)
}
fn table_has_column(conn: &Connection, table: &str, column: &str) -> Result<bool> {
let mut statement = conn.prepare(&format!("PRAGMA table_info({table})"))?;
let columns = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(columns.iter().any(|candidate| candidate == column))
}
const CACHE_TAIL_BYTES: u64 = 64 * 1024;
fn ensure_cache_tail_hash_columns(conn: &Connection) -> Result<()> {
for table in ["source_files", "session_source_files"] {
let mut statement = conn.prepare(&format!("PRAGMA table_info({table})"))?;
let columns = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<std::result::Result<Vec<_>, _>>()?;
if !columns.iter().any(|column| column == "tail_hash") {
conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN tail_hash TEXT"))?;
}
}
Ok(())
}
fn backfill_cache_tail_hashes(conn: &mut Connection) -> Result<()> {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
for table in ["source_files", "session_source_files"] {
let rows = {
let mut statement = tx.prepare(&format!(
"SELECT source_path, size FROM {table} WHERE tail_hash IS NULL"
))?;
let rows = statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64))
})?
.collect::<std::result::Result<Vec<_>, _>>()?;
rows
};
let mut update = tx.prepare(&format!(
"UPDATE {table} SET tail_hash = ?1 WHERE source_path = ?2"
))?;
for (source_path, size) in rows {
if let Some(tail_hash) = file_tail_hash(Path::new(&source_path), size) {
update.execute(params![tail_hash, source_path])?;
}
}
}
tx.commit()?;
Ok(())
}
fn file_tail_hash(path: &Path, end: u64) -> Option<String> {
let mut file = File::open(path).ok()?;
if file.metadata().ok()?.len() < end {
return None;
}
let start = end.saturating_sub(CACHE_TAIL_BYTES);
file.seek(SeekFrom::Start(start)).ok()?;
let mut bytes = Vec::with_capacity((end - start) as usize);
file.take(end - start).read_to_end(&mut bytes).ok()?;
if bytes.len() as u64 != end - start {
return None;
}
Some(format!("{:x}", Sha256::digest(&bytes)))
}
fn cached_tail_hash(conn: &Connection, table: &str, source_path: &str) -> Result<Option<String>> {
let sql = match table {
"source_files" => "SELECT tail_hash FROM source_files WHERE source_path = ?1",
"session_source_files" => {
"SELECT tail_hash FROM session_source_files WHERE source_path = ?1"
}
_ => unreachable!("fixed cache metadata table"),
};
Ok(conn
.query_row(sql, [source_path], |row| row.get(0))
.optional()?
.flatten())
}
fn append_offset(
conn: &Connection,
table: &str,
path: &Path,
source_path: &str,
cached: Option<&ClaudeCacheFileState>,
current: &ClaudeCacheFileState,
) -> Result<Option<u64>> {
let Some(cached) = cached.filter(|cached| cached.size < current.size) else {
return Ok(None);
};
if cached.size > 0 {
let mut file = File::open(path)?;
file.seek(SeekFrom::Start(cached.size - 1))?;
let mut last_byte = [0_u8; 1];
file.read_exact(&mut last_byte)?;
if last_byte[0] != b'\n' {
return Ok(None);
}
}
let Some(cached_hash) = cached_tail_hash(conn, table, source_path)? else {
return Ok(None);
};
Ok(
(file_tail_hash(path, cached.size).as_deref() == Some(cached_hash.as_str()))
.then_some(cached.size),
)
}
fn cached_session_aggregate(
conn: &Connection,
source_path: &str,
) -> Result<Option<SessionAggregate>> {
Ok(conn
.query_row(
"SELECT cwd, git_branch, version, title, first_timestamp, last_timestamp,
user_message_count, assistant_message_count, total_input_tokens,
total_output_tokens, total_cache_read_input_tokens,
total_cache_creation_input_tokens, pr_url, pr_number
FROM sessions WHERE source_path = ?1",
[source_path],
|row| {
Ok(SessionAggregate {
cwd: row.get(0)?,
git_branch: row.get(1)?,
version: row.get(2)?,
title: row.get(3)?,
first_timestamp: row.get(4)?,
last_timestamp: row.get(5)?,
user_message_count: row.get(6)?,
assistant_message_count: row.get(7)?,
total_input_tokens: row.get(8)?,
total_output_tokens: row.get(9)?,
total_cache_read_input_tokens: row.get(10)?,
total_cache_creation_input_tokens: row.get(11)?,
pr_url: row.get(12)?,
pr_number: row.get(13)?,
})
},
)
.optional()?)
}
fn sync_claude_tool_cache(
cache: &mut Connection,
files: &[TranscriptFile],
include_sessions: bool,
) -> Result<()> {
let current = claude_file_states(files);
let current_sessions = claude_session_file_states(files);
if claude_cache_matches(cache, ¤t, ¤t_sessions, include_sessions)? {
return Ok(());
}
cache.busy_timeout(Duration::from_millis(100))?;
let tx = match cache.transaction_with_behavior(TransactionBehavior::Immediate) {
Ok(tx) => tx,
Err(error) if is_busy_sql_error(&error) => return Ok(()),
Err(error) => return Err(error.into()),
};
if claude_cache_matches(&tx, ¤t, ¤t_sessions, include_sessions)? {
tx.commit()?;
return Ok(());
}
let cached = claude_cached_file_states(&tx)?;
let cached_sessions = claude_cached_session_file_states(&tx)?;
for file in files {
let source_path = file.path.to_string_lossy().into_owned();
let Some(state) = current.get(&source_path) else {
continue;
};
if cached.get(&source_path) == Some(state) {
continue;
}
let offset = append_offset(
&tx,
"source_files",
&file.path,
&source_path,
cached.get(&source_path),
state,
)?;
if offset.is_none() {
tx.execute(
"DELETE FROM tool_calls WHERE source_path = ?1",
[&source_path],
)?;
}
{
let mut insert = tx.prepare(
"INSERT INTO tool_calls (
tool_name, input_json, target, source_id, command, session_id,
parent_session_id, agent_id, agent_role, originator, cwd,
source_path, project, timestamp
) VALUES (
?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
)",
)?;
visit_jsonl_candidates_from_offset(
&file.path,
&[b"\"tool_use\""],
offset.unwrap_or(0),
|entry| {
let record_session_id = string_field(&entry, "sessionId")
.or_else(|| string_field(&entry, "session_id"))
.unwrap_or_else(|| file.session_id.clone());
let parent_session_id = file.agent_id.as_ref().map(|_| file.session_id.clone());
let agent_role = string_field(&entry, "agentName");
let originator = string_field(&entry, "originator")
.or_else(|| nested_string_field(&entry, "origin", "kind"))
.or_else(|| string_field(&entry, "entrypoint"));
let cwd = string_field(&entry, "cwd");
for call in ccql::datasources::tool_calls::extract_tool_calls(&entry) {
insert.execute(params![
call.tool_name,
call.input_json,
call.target,
call.source_id,
call.command,
record_session_id,
parent_session_id,
file.agent_id,
agent_role,
originator,
cwd,
source_path,
file.project,
call.timestamp,
])?;
}
Ok(())
},
)?;
}
let tail_hash = file_tail_hash(&file.path, state.size);
tx.execute(
"INSERT OR REPLACE INTO source_files (source_path, size, modified_ns, tail_hash)
VALUES (?1, ?2, ?3, ?4)",
params![source_path, state.size as i64, state.modified_ns, tail_hash],
)?;
}
for file in files
.iter()
.filter(|file| include_sessions && file.agent_id.is_none())
{
let source_path = file.path.to_string_lossy().into_owned();
let Some(state) = current_sessions.get(&source_path) else {
continue;
};
if cached_sessions.get(&source_path) == Some(state) {
continue;
}
let offset = append_offset(
&tx,
"session_source_files",
&file.path,
&source_path,
cached_sessions.get(&source_path),
state,
)?;
let mut aggregate = if offset.is_some() {
cached_session_aggregate(&tx, &source_path)?.unwrap_or_default()
} else {
SessionAggregate::default()
};
visit_jsonl_candidates_from_offset(&file.path, &[], offset.unwrap_or(0), |entry| {
aggregate.observe(&entry);
Ok(())
})?;
tx.execute(
"INSERT OR REPLACE INTO sessions (
source_path, session_id, project, cwd, git_branch, version, title,
first_timestamp, last_timestamp, user_message_count,
assistant_message_count, subagent_count, total_input_tokens,
total_output_tokens, total_cache_read_input_tokens,
total_cache_creation_input_tokens, pr_url, pr_number
) VALUES (
?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, 0,
?12, ?13, ?14, ?15, ?16, ?17
)",
params![
source_path,
file.session_id,
file.project,
aggregate.cwd,
aggregate.git_branch,
aggregate.version,
aggregate.title,
aggregate.first_timestamp,
aggregate.last_timestamp,
aggregate.user_message_count,
aggregate.assistant_message_count,
aggregate.total_input_tokens,
aggregate.total_output_tokens,
aggregate.total_cache_read_input_tokens,
aggregate.total_cache_creation_input_tokens,
aggregate.pr_url,
aggregate.pr_number,
],
)?;
let tail_hash = file_tail_hash(&file.path, state.size);
tx.execute(
"INSERT OR REPLACE INTO session_source_files
(source_path, size, modified_ns, tail_hash)
VALUES (?1, ?2, ?3, ?4)",
params![source_path, state.size as i64, state.modified_ns, tail_hash],
)?;
}
for source_path in cached.keys() {
if current.contains_key(source_path) {
continue;
}
tx.execute(
"DELETE FROM tool_calls WHERE source_path = ?1",
[source_path],
)?;
tx.execute(
"DELETE FROM source_files WHERE source_path = ?1",
[source_path],
)?;
}
if include_sessions {
for source_path in cached_sessions.keys() {
if current_sessions.contains_key(source_path) {
continue;
}
tx.execute("DELETE FROM sessions WHERE source_path = ?1", [source_path])?;
tx.execute(
"DELETE FROM session_source_files WHERE source_path = ?1",
[source_path],
)?;
}
let mut subagent_counts: HashMap<(Option<String>, String), i64> = HashMap::new();
for file in files.iter().filter(|file| file.agent_id.is_some()) {
*subagent_counts
.entry((file.project.clone(), file.session_id.clone()))
.or_insert(0) += 1;
}
tx.execute("UPDATE sessions SET subagent_count = 0", [])?;
let mut update_subagents = tx.prepare(
"UPDATE sessions
SET subagent_count = ?1
WHERE project IS ?2 AND session_id = ?3",
)?;
for ((project, session_id), count) in subagent_counts {
update_subagents.execute(params![count, project, session_id])?;
}
}
tx.commit()?;
Ok(())
}
fn is_busy_sql_error(error: &rusqlite::Error) -> bool {
matches!(
error,
rusqlite::Error::SqliteFailure(code, _)
if matches!(code.code, ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked)
)
}
fn claude_file_states(files: &[TranscriptFile]) -> HashMap<String, ClaudeCacheFileState> {
files
.iter()
.filter_map(|file| {
let metadata = fs::metadata(&file.path).ok()?;
Some((
file.path.to_string_lossy().into_owned(),
ClaudeCacheFileState {
size: metadata.len(),
modified_ns: metadata_modified_ns(&metadata),
},
))
})
.collect()
}
fn claude_session_file_states(files: &[TranscriptFile]) -> HashMap<String, ClaudeCacheFileState> {
claude_file_states(
&files
.iter()
.filter(|file| file.agent_id.is_none())
.cloned()
.collect::<Vec<_>>(),
)
}
fn claude_cached_file_states(conn: &Connection) -> Result<HashMap<String, ClaudeCacheFileState>> {
let mut statement = conn.prepare("SELECT source_path, size, modified_ns FROM source_files")?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
ClaudeCacheFileState {
size: row.get::<_, i64>(1)? as u64,
modified_ns: row.get(2)?,
},
))
})?;
Ok(rows.collect::<std::result::Result<HashMap<_, _>, _>>()?)
}
fn claude_cached_session_file_states(
conn: &Connection,
) -> Result<HashMap<String, ClaudeCacheFileState>> {
let mut statement =
conn.prepare("SELECT source_path, size, modified_ns FROM session_source_files")?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
ClaudeCacheFileState {
size: row.get::<_, i64>(1)? as u64,
modified_ns: row.get(2)?,
},
))
})?;
Ok(rows.collect::<std::result::Result<HashMap<_, _>, _>>()?)
}
fn claude_cache_matches(
conn: &Connection,
current: &HashMap<String, ClaudeCacheFileState>,
current_sessions: &HashMap<String, ClaudeCacheFileState>,
include_sessions: bool,
) -> Result<bool> {
Ok(claude_cached_file_states(conn)? == *current
&& (!include_sessions || claude_cached_session_file_states(conn)? == *current_sessions))
}
fn metadata_modified_ns(metadata: &Metadata) -> i64 {
metadata
.modified()
.ok()
.and_then(|modified| modified.duration_since(UNIX_EPOCH).ok())
.map(|duration| i64::try_from(duration.as_nanos()).unwrap_or(i64::MAX))
.unwrap_or_default()
}
#[cfg(unix)]
fn set_private_directory_permissions(path: &Path) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
Ok(())
}
#[cfg(not(unix))]
fn set_private_directory_permissions(_path: &Path) -> Result<()> {
Ok(())
}
#[cfg(unix)]
fn set_private_file_permissions(path: &Path) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o600))?;
Ok(())
}
#[cfg(not(unix))]
fn set_private_file_permissions(_path: &Path) -> Result<()> {
Ok(())
}
fn set_private_cache_sidecar_permissions(path: &Path) -> Result<()> {
for sidecar in [
PathBuf::from(format!("{}-wal", path.to_string_lossy())),
PathBuf::from(format!("{}-shm", path.to_string_lossy())),
] {
if sidecar.exists() {
set_private_file_permissions(&sidecar)?;
}
}
Ok(())
}
fn normalize_date(value: &str) -> String {
if value.chars().all(|c| c.is_ascii_digit()) && value.len() >= 13 {
if let Ok(ms) = value.parse::<i64>() {
let secs = ms / 1000;
if let Some(dt) = DateTime::from_timestamp(secs, 0) {
return dt.format("%Y-%m-%d").to_string();
}
}
}
if value.chars().all(|c| c.is_ascii_digit()) && value.len() >= 10 {
if let Ok(secs) = value.parse::<i64>() {
if let Some(dt) = DateTime::from_timestamp(secs, 0) {
return dt.format("%Y-%m-%d").to_string();
}
}
}
if value.len() >= 10 {
return value[..10].to_string();
}
value.to_string()
}
fn format_git_time(secs: i64) -> String {
DateTime::from_timestamp(secs, 0)
.map(|dt| dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
.unwrap_or_default()
}
fn default_codex_data_dir() -> PathBuf {
std::env::var_os("CODEX_HOME")
.map(PathBuf::from)
.or_else(|| dirs::home_dir().map(|p| p.join(".codex")))
.unwrap_or_else(|| PathBuf::from(".codex"))
}
pub(crate) fn default_grok_data_dir() -> PathBuf {
if let Some(dir) = std::env::var_os("DEVSQL_GROK_DIR") {
return PathBuf::from(dir);
}
let sand_data = PathBuf::from("/home/box/sand-data");
if sand_data.is_dir() {
return sand_data;
}
let Some(home) = dirs::home_dir() else {
return PathBuf::from("Grok Bot");
};
if cfg!(target_os = "macos") {
home.join("Library")
.join("Application Support")
.join("Grok Bot")
} else {
std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".config"))
.join("Grok Bot")
}
}
fn json_number_as_i64(value: &Value) -> Option<i64> {
value.as_i64().or_else(|| {
value
.as_u64()
.and_then(|n| i64::try_from(n).ok())
.or_else(|| value.as_str().and_then(|s| s.parse::<i64>().ok()))
})
}
fn json_value_as_string(value: &Value) -> Option<String> {
match value {
Value::String(s) => Some(s.clone()),
Value::Null => None,
other => Some(other.to_string()),
}
}
fn normalize_ts_seconds(raw_ts: i64) -> i64 {
if raw_ts > 10_000_000_000 {
raw_ts / 1000
} else {
raw_ts
}
}
fn visit_jsonl_candidates_from_offset<F>(
path: &Path,
markers: &[&[u8]],
offset: u64,
mut visit: F,
) -> Result<()>
where
F: FnMut(Value) -> Result<()>,
{
let Ok(mut file) = File::open(path) else {
return Ok(());
};
if file.seek(SeekFrom::Start(offset)).is_err() {
return Ok(());
}
let mut reader = BufReader::with_capacity(256 * 1024, file);
let mut line = Vec::new();
loop {
line.clear();
let bytes_read = match reader.read_until(b'\n', &mut line) {
Ok(bytes_read) => bytes_read,
Err(_) => return Ok(()),
};
if bytes_read == 0 {
return Ok(());
}
if !markers.is_empty()
&& !markers
.iter()
.any(|marker| memchr::memmem::find(&line, marker).is_some())
{
continue;
}
let Ok(entry) = serde_json::from_slice::<Value>(&line) else {
continue;
};
visit(entry)?;
}
}
fn string_field(json: &Value, field: &str) -> Option<String> {
json.get(field).and_then(Value::as_str).map(String::from)
}
fn nested_string_field(json: &Value, parent: &str, field: &str) -> Option<String> {
json.get(parent)
.and_then(|value| value.get(field))
.and_then(Value::as_str)
.map(String::from)
}
fn query_mentions_table(query_upper: &str, table_name: &str) -> bool {
let table_upper = table_name.to_uppercase();
query_upper
.split(|c: char| !c.is_ascii_alphanumeric() && c != '_')
.any(|token| token == table_upper)
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TableRequirements {
pub claude: Vec<String>,
pub git: Vec<String>,
pub code: Vec<String>,
pub shell: Vec<String>,
pub work: Vec<String>,
pub system: Vec<String>,
pub grok: Vec<String>,
}
pub fn detect_tables(query: &str) -> TableRequirements {
let query_upper = query.to_uppercase();
let claude_tables = [
"history",
"jhistory",
"codex_history",
"transcripts",
"sessions",
"todos",
"stats",
"tool_calls",
"codex_tool_calls",
"codex_threads",
"codex_events",
"codex_messages",
"codex_tool_executions",
"codex_compactions",
"codex_ingest_errors",
];
let git_tables = [
"commits",
"commit_parents",
"branches",
"tags",
"refs",
"stashes",
"reflog",
"diffs",
"diff_files",
"blame",
"config",
"remotes",
"submodules",
"status",
"worktrees",
"hooks",
"notes",
];
let code_tables = [
"source_files",
"source_lines",
"symbols",
"imports",
"ast_nodes",
];
let shell_tables = ["shell_history", "command_events"];
let work_tables = ["work_tasks", "work_events"];
let system_tables = ["macos_logs"];
let grok_tables = [
"grok_bots",
"grok_entries",
"grok_messages",
"grok_ingest_errors",
];
let needed_claude: Vec<String> = claude_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_git: Vec<String> = git_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_code: Vec<String> = code_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_shell: Vec<String> = shell_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_work: Vec<String> = work_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_system: Vec<String> = system_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
let needed_grok: Vec<String> = grok_tables
.iter()
.filter(|t| query_mentions_table(&query_upper, t))
.map(|s| s.to_string())
.collect();
TableRequirements {
claude: needed_claude,
git: needed_git,
code: needed_code,
shell: needed_shell,
work: needed_work,
system: needed_system,
grok: needed_grok,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
#[test]
fn detect_tables_handles_jhistory_without_history_false_positive() {
let needed = detect_tables("SELECT session_id, text FROM jhistory LIMIT 5");
let claude = needed.claude;
assert!(claude.contains(&"jhistory".to_string()));
assert!(!claude.contains(&"history".to_string()));
}
#[test]
fn detect_tables_handles_codex_history_without_history_false_positive() {
let needed = detect_tables("SELECT session_id, text FROM codex_history LIMIT 5");
let claude = needed.claude;
assert!(claude.contains(&"codex_history".to_string()));
assert!(!claude.contains(&"history".to_string()));
}
#[test]
fn detect_tables_finds_code_tables() {
let code = detect_tables(
"SELECT * FROM source_files JOIN symbols ON source_files.path = symbols.file_path",
)
.code;
assert!(code.contains(&"source_files".to_string()));
assert!(code.contains(&"symbols".to_string()));
assert!(!code.contains(&"source_lines".to_string()));
}
#[test]
fn detect_tables_finds_work_tables() {
let work = detect_tables(
"SELECT * FROM work_events JOIN work_tasks ON work_events.task_id = work_tasks.id",
)
.work;
assert!(work.contains(&"work_events".to_string()));
assert!(work.contains(&"work_tasks".to_string()));
}
#[test]
fn detect_tables_finds_shell_history_without_history_false_positive() {
let needed = detect_tables("SELECT command FROM shell_history LIMIT 5");
let (claude, shell) = (needed.claude, needed.shell);
assert_eq!(shell, vec!["shell_history".to_string()]);
assert!(!claude.contains(&"history".to_string()));
}
#[test]
fn detect_tables_finds_command_events() {
let shell = detect_tables("SELECT actor, command FROM command_events").shell;
assert_eq!(shell, vec!["command_events".to_string()]);
}
#[test]
fn detect_tables_finds_macos_logs() {
let needed = detect_tables("SELECT subsystem, message FROM macos_logs");
let (shell, system) = (needed.shell, needed.system);
assert!(shell.is_empty());
assert_eq!(system, vec!["macos_logs".to_string()]);
}
#[test]
fn detect_tables_finds_grok_tables_without_history_false_positive() {
let needed = detect_tables("SELECT name, text FROM grok_messages LIMIT 5");
assert_eq!(needed.grok, vec!["grok_messages".to_string()]);
assert!(!needed.claude.contains(&"history".to_string()));
assert!(needed.shell.is_empty());
}
#[test]
fn detect_tables_finds_grok_bots_and_entries() {
let needed = detect_tables(
"SELECT * FROM grok_bots JOIN grok_entries ON grok_bots.bot_id = grok_entries.bot_id",
);
assert!(needed.grok.contains(&"grok_bots".to_string()));
assert!(needed.grok.contains(&"grok_entries".to_string()));
}
#[test]
fn normalize_ts_seconds_converts_millis() {
assert_eq!(normalize_ts_seconds(1_754_402_102), 1_754_402_102);
assert_eq!(normalize_ts_seconds(1_754_402_102_000), 1_754_402_102);
}
fn write(path: &std::path::Path, contents: &str) {
std::fs::create_dir_all(path.parent().unwrap()).expect("mkdir");
std::fs::write(path, contents).expect("write");
}
#[test]
fn claude_tool_cache_is_incremental_lock_free_and_indexed() {
let temp = tempfile::tempdir().expect("temp");
let journal = temp.path().join("claude-session.jsonl");
write(
&journal,
concat!(
r#"{"type":"assistant","timestamp":"2026-07-12T10:00:00.000Z","sessionId":"claude-session","message":{"content":[{"type":"tool_use","id":"toolu_bash_1","name":"Bash","input":{"command":"first"}}]}}"#,
"\n",
),
);
let files = vec![TranscriptFile {
path: journal.clone(),
source_file: "claude-session.jsonl".into(),
session_id: "claude-session".into(),
project: Some("-work-claude".into()),
agent_id: None,
}];
let cache_path = temp.path().join("cache/tool-calls.sqlite");
let mut cache = open_claude_tool_cache(&cache_path).expect("open cache");
sync_claude_tool_cache(&mut cache, &files, true).expect("initial sync");
let count: i64 = cache
.query_row("SELECT COUNT(*) FROM tool_calls", [], |row| row.get(0))
.expect("count");
assert_eq!(count, 1);
let session_count: i64 = cache
.query_row("SELECT COUNT(*) FROM sessions", [], |row| row.get(0))
.expect("session count");
assert_eq!(session_count, 1);
cache
.busy_timeout(Duration::from_millis(50))
.expect("short test timeout");
let writer = Connection::open(&cache_path).expect("writer connection");
writer
.execute_batch("PRAGMA journal_mode = WAL; BEGIN IMMEDIATE")
.expect("hold writer lock");
let started = std::time::Instant::now();
sync_claude_tool_cache(&mut cache, &files, true).expect("warm sync remains read-only");
let elapsed = started.elapsed();
writer
.execute_batch("ROLLBACK")
.expect("release writer lock");
assert!(
elapsed < Duration::from_millis(50),
"warm cache waited for the writer lock: {elapsed:?}"
);
let plan: String = cache
.query_row(
"EXPLAIN QUERY PLAN
SELECT source_id, timestamp FROM tool_calls
WHERE tool_name = 'Bash'
ORDER BY timestamp DESC LIMIT 10",
[],
|row| row.get(3),
)
.expect("query plan");
assert!(
plan.contains("idx_claude_tool_calls_tool_timestamp"),
"unexpected tool-call plan: {plan}"
);
let session_plan: String = cache
.query_row(
"EXPLAIN QUERY PLAN
SELECT session_id, last_timestamp FROM sessions
WHERE project = '-work-claude'
ORDER BY last_timestamp DESC LIMIT 10",
[],
|row| row.get(3),
)
.expect("session query plan");
assert!(
session_plan.contains("idx_claude_sessions_project_last_timestamp"),
"unexpected session plan: {session_plan}"
);
let mut contents = std::fs::read_to_string(&journal).expect("read journal");
contents.push_str(concat!(
r#"{"type":"assistant","timestamp":"2026-07-12T10:00:01.000Z","sessionId":"claude-session","message":{"content":[{"type":"tool_use","id":"toolu_read_1","name":"Read","input":{"file_path":"src/lib.rs"}}]}}"#,
"\n",
));
std::fs::write(&journal, contents).expect("append tool call");
let writer = Connection::open(&cache_path).expect("second writer connection");
writer
.execute_batch("PRAGMA journal_mode = WAL; BEGIN IMMEDIATE")
.expect("hold writer lock for changed journal");
let started = std::time::Instant::now();
sync_claude_tool_cache(&mut cache, &files, true).expect("serve last good cache");
let elapsed = started.elapsed();
writer
.execute_batch("ROLLBACK")
.expect("release changed-journal writer lock");
assert!(
elapsed < Duration::from_millis(250),
"changed cache waited for the writer lock: {elapsed:?}"
);
let stale_count: i64 = cache
.query_row("SELECT COUNT(*) FROM tool_calls", [], |row| row.get(0))
.expect("stale count");
assert_eq!(stale_count, 1);
sync_claude_tool_cache(&mut cache, &files, true).expect("incremental sync");
let count: i64 = cache
.query_row("SELECT COUNT(*) FROM tool_calls", [], |row| row.get(0))
.expect("updated count");
assert_eq!(count, 2);
let last_timestamp: String = cache
.query_row(
"SELECT last_timestamp FROM sessions WHERE session_id = 'claude-session'",
[],
|row| row.get(0),
)
.expect("updated session");
assert_eq!(last_timestamp, "2026-07-12T10:00:01.000Z");
}
#[test]
fn opening_claude_cache_restores_missing_consumer_index() {
let temp = tempfile::tempdir().expect("temp");
let cache_path = temp.path().join("cache/tool-calls.sqlite");
let cache = open_claude_tool_cache(&cache_path).expect("open cache");
cache
.execute_batch("DROP INDEX idx_claude_tool_calls_tool_timestamp")
.expect("drop index");
drop(cache);
let cache = open_claude_tool_cache(&cache_path).expect("reopen cache");
let restored: i64 = cache
.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type = 'index'
AND name = 'idx_claude_tool_calls_tool_timestamp'",
[],
|row| row.get(0),
)
.expect("restored index");
assert_eq!(restored, 1);
}
#[test]
fn loads_modern_projects_layout_with_usage_and_sessions() {
let temp = tempfile::tempdir().expect("temp");
let root = temp.path();
let slug = "-Users-doug-Developer-app";
let records = [
r#"{"type":"user","content":"hi","timestamp":"2026-06-01T10:00:00.000Z","cwd":"/Users/doug/Developer/app","gitBranch":"main","version":"2.1.100"}"#,
r#"{"type":"assistant","timestamp":"2026-06-01T10:00:05.000Z","message":{"model":"claude-opus-4-7","usage":{"input_tokens":6,"output_tokens":127,"cache_read_input_tokens":100,"cache_creation_input_tokens":200}}}"#,
r#"{"type":"ai-title","aiTitle":"Fix the widget","sessionId":"sess-rich"}"#,
r#"{"type":"pr-link","sessionId":"sess-rich","prNumber":42,"prUrl":"https://github.com/org/repo/pull/42"}"#,
]
.join("\n");
write(
&root.join("projects").join(slug).join("sess-rich.jsonl"),
&records,
);
write(
&root
.join("projects")
.join(slug)
.join("sess-rich")
.join("subagents")
.join("agent-one.jsonl"),
r#"{"type":"assistant","message":{"usage":{"input_tokens":999,"output_tokens":999}}}"#,
);
write(
&root.join("transcripts").join("ses_old.jsonl"),
r#"{"type":"user","content":"legacy"}"#,
);
let mut engine =
UnifiedEngine::new(root.to_path_buf(), root.to_path_buf()).expect("engine");
engine
.load_claude_tables(&["transcripts", "sessions"])
.expect("load");
let count = engine
.query("SELECT COUNT(*) AS n FROM transcripts")
.expect("count");
assert_eq!(count[0]["n"], serde_json::json!(6));
let sub = engine
.query("SELECT _project, _agent_id, _session_id FROM transcripts WHERE _agent_id IS NOT NULL")
.expect("subagent row");
assert_eq!(sub.len(), 1);
assert_eq!(sub[0]["_project"], serde_json::json!(slug));
assert_eq!(sub[0]["_agent_id"], serde_json::json!("agent-one"));
assert_eq!(sub[0]["_session_id"], serde_json::json!("sess-rich"));
let usage = engine
.query(
"SELECT model, usage_input_tokens, usage_output_tokens, \
usage_cache_read_input_tokens, usage_cache_creation_input_tokens \
FROM transcripts WHERE _agent_id IS NULL AND type = 'assistant'",
)
.expect("usage row");
assert_eq!(usage.len(), 1);
assert_eq!(usage[0]["model"], serde_json::json!("claude-opus-4-7"));
assert_eq!(usage[0]["usage_input_tokens"], serde_json::json!(6));
assert_eq!(usage[0]["usage_output_tokens"], serde_json::json!(127));
assert_eq!(
usage[0]["usage_cache_read_input_tokens"],
serde_json::json!(100)
);
assert_eq!(
usage[0]["usage_cache_creation_input_tokens"],
serde_json::json!(200)
);
let sessions = engine
.query(
"SELECT session_id, project, title, pr_number, pr_url, subagent_count, \
user_message_count, assistant_message_count, total_input_tokens, \
total_output_tokens FROM sessions ORDER BY session_id",
)
.expect("sessions");
assert_eq!(sessions.len(), 2);
let legacy = &sessions[0];
assert_eq!(legacy["session_id"], serde_json::json!("old"));
assert_eq!(legacy["project"], serde_json::Value::Null);
assert_eq!(legacy["subagent_count"], serde_json::json!(0));
let rich = &sessions[1];
assert_eq!(rich["session_id"], serde_json::json!("sess-rich"));
assert_eq!(rich["project"], serde_json::json!(slug));
assert_eq!(rich["title"], serde_json::json!("Fix the widget"));
assert_eq!(rich["pr_number"], serde_json::json!(42));
assert_eq!(
rich["pr_url"],
serde_json::json!("https://github.com/org/repo/pull/42")
);
assert_eq!(rich["subagent_count"], serde_json::json!(1));
assert_eq!(rich["user_message_count"], serde_json::json!(1));
assert_eq!(rich["assistant_message_count"], serde_json::json!(1));
assert_eq!(rich["total_input_tokens"], serde_json::json!(6));
assert_eq!(rich["total_output_tokens"], serde_json::json!(127));
}
#[test]
fn codex_tool_calls_include_archived_compressed_journals() {
let temp = tempfile::tempdir().expect("temp");
let codex_home = temp.path().join("codex");
let journal = codex_home
.join("archived_sessions")
.join("rollout-thread-z.jsonl.zst");
std::fs::create_dir_all(journal.parent().expect("parent")).expect("mkdir");
let output = std::fs::File::create(&journal).expect("create");
let mut encoder = zstd::stream::write::Encoder::new(output, 0).expect("encoder");
encoder
.write_all(
concat!(
"{\"type\":\"session_meta\",\"payload\":{\"id\":\"thread-z\",\"cwd\":\"/repo\"}}\n",
"{\"timestamp\":\"2026-07-27T10:00:00Z\",\"type\":\"response_item\",\"payload\":{\"type\":\"message\",\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"zstd history\"}]}}\n",
"{\"timestamp\":\"2026-07-27T10:00:01Z\",\"type\":\"response_item\",\"payload\":{\"type\":\"function_call\",\"name\":\"exec_command\",\"call_id\":\"call-z\",\"arguments\":\"{\\\"cmd\\\":\\\"cargo test\\\"}\"}}\n",
"{\"timestamp\":\"2026-07-27T10:00:02Z\",\"type\":\"response_item\",\"payload\":{\"type\":\"function_call_output\",\"call_id\":\"call-z\",\"output\":\"ok\"}}\n",
"{\"timestamp\":\"2026-07-27T10:00:03Z\",\"type\":\"compacted\",\"payload\":{\"window_id\":\"w2\",\"message\":\"zstd summary\"}}\n",
"not-json\n"
)
.as_bytes(),
)
.expect("write");
encoder.finish().expect("finish");
let mut engine = UnifiedEngine::new_with_codex_data_dir(
temp.path().to_path_buf(),
temp.path().to_path_buf(),
codex_home,
)
.expect("engine");
engine
.load_claude_tables(&["codex_tool_calls"])
.expect("load");
let rows = engine
.query(
"SELECT tool_name, cmd, session_id, cwd
FROM codex_tool_calls",
)
.expect("query");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["tool_name"], serde_json::json!("exec_command"));
assert_eq!(rows[0]["cmd"], serde_json::json!("cargo test"));
assert_eq!(rows[0]["session_id"], serde_json::json!("thread-z"));
assert_eq!(rows[0]["cwd"], serde_json::json!("/repo"));
let thread = engine
.query(
"SELECT state, compressed, event_count, user_message_count,
tool_call_count, compaction_count
FROM codex_threads WHERE thread_id = 'thread-z'",
)
.expect("thread");
assert_eq!(thread[0]["state"], serde_json::json!("archived"));
assert_eq!(thread[0]["compressed"], serde_json::json!(1));
assert_eq!(thread[0]["event_count"], serde_json::json!(6));
assert_eq!(thread[0]["user_message_count"], serde_json::json!(1));
assert_eq!(thread[0]["tool_call_count"], serde_json::json!(1));
assert_eq!(thread[0]["compaction_count"], serde_json::json!(1));
let message = engine
.query("SELECT role, text, is_canonical FROM codex_messages")
.expect("messages");
assert_eq!(message[0]["role"], serde_json::json!("user"));
assert_eq!(message[0]["text"], serde_json::json!("zstd history"));
assert_eq!(message[0]["is_canonical"], serde_json::json!(1));
let execution = engine
.query(
"SELECT tool_name, cmd, output_text
FROM codex_tool_executions",
)
.expect("execution");
assert_eq!(execution[0]["output_text"], serde_json::json!("ok"));
let compaction = engine
.query("SELECT window_id, summary_text FROM codex_compactions")
.expect("compaction");
assert_eq!(compaction[0]["window_id"], serde_json::json!("w2"));
assert_eq!(
compaction[0]["summary_text"],
serde_json::json!("zstd summary")
);
let errors = engine
.query("SELECT error_kind FROM codex_ingest_errors")
.expect("errors");
assert_eq!(errors.len(), 1);
assert_eq!(errors[0]["error_kind"], serde_json::json!("json"));
}
}