diff --git a/src/adapters/AGENTS.md b/src/adapters/AGENTS.md
index 334f16c..f25ca55 100644
--- a/src/adapters/AGENTS.md
+++ b/src/adapters/AGENTS.md
@@ -14,9 +14,14 @@ apply when working here.
   wires the adapter into sync, search, the TUI source filter, and the CLI
   `--source` flag. No schema change is needed — `sessions.source` is a value,
   not a column per tool.
-- `events.rs`, `file_scan.rs`, and `sync_state.rs` are shared helpers, not
-  adapters. Prefer `file_scan::run_file_scan_with_options` over hand-rolled
-  mtime tracking for file-based sources.
+- `events.rs`, `file_scan.rs`, `sync_state.rs`, `json_util.rs`, and `paths.rs`
+  are shared helpers, not adapters. Prefer them over hand-rolling:
+  `file_scan::run_file_scan_with_options` for mtime tracking,
+  `json_util::jsonl_indexed` for per-line JSONL read loops,
+  `json_util::rfc3339_ms` and `json_util::json_i64` for timestamp/number
+  coercion, `paths::resolve_home_dir` for `~/…` directory resolution, and
+  `first_timestamp`/`last_timestamp` (`mod.rs`) for started_at/updated_at
+  fallback chains.
 - Usage and session events are extensions on the same `RawSession`
   (`with_usage`, `with_events`), each with its own parser version. Bump the
   parser version when parsing changes — that is what triggers backfill for
diff --git a/src/adapters/antigravity.rs b/src/adapters/antigravity.rs
index e76c207..afc901a 100644
--- a/src/adapters/antigravity.rs
+++ b/src/adapters/antigravity.rs
@@ -8,6 +8,7 @@ use tracing::debug;
 use walkdir::WalkDir;
 
 use crate::adapters::file_scan::{self, FileScanEntry};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
 };
@@ -65,13 +66,10 @@ impl SourceAdapter for AntigravityAdapter {
 }
 
 fn resolve_antigravity_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".gemini/antigravity-cli");
-    if !dir.exists() {
-        debug!("~/.gemini/antigravity-cli not found, skipping Antigravity CLI");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(
+        ".gemini/antigravity-cli",
+        "~/.gemini/antigravity-cli not found, skipping Antigravity CLI",
+    )
 }
 
 fn scan_for_sync_impl(
diff --git a/src/adapters/claude_code.rs b/src/adapters/claude_code.rs
index fbcedb1..79a8a69 100644
--- a/src/adapters/claude_code.rs
+++ b/src/adapters/claude_code.rs
@@ -9,8 +9,11 @@ use walkdir::WalkDir;
 
 use crate::adapters::events;
 use crate::adapters::file_scan::{self, FileScanEntry};
+use crate::adapters::json_util::{jsonl_indexed, rfc3339_ms};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
+    first_timestamp,
 };
 use crate::db::store::Store;
 use crate::types::{RawSessionEvent, RawUsageEvent, Role, TokenSource};
@@ -92,13 +95,7 @@ fn load_session_indexes(claude_dir: &Path) -> SessionIndexes {
 }
 
 fn resolve_claude_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".claude");
-    if !dir.exists() {
-        debug!("~/.claude not found, skipping Claude Code");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(".claude", "~/.claude not found, skipping Claude Code")
 }
 
 fn scan_for_sync_impl(
@@ -302,11 +299,13 @@ fn parse_claude_session_file(
     }
 
     let meta = indexes.live.get(&entry.session_id);
-    let started_at = meta
-        .and_then(|m| m.started_at)
-        .or_else(|| parsed.messages.first().and_then(|m| m.timestamp))
-        .or_else(|| parsed.usage_events.first().map(|event| event.timestamp))
-        .unwrap_or(0);
+    let started_at = first_timestamp(
+        meta.and_then(|m| m.started_at),
+        &parsed.messages,
+        &parsed.usage_events,
+        &[],
+    )
+    .unwrap_or(0);
     let directory =
         meta.and_then(|m| m.cwd.clone()).or_else(|| parsed.cwd.clone()).or(entry.directory);
     let entrypoint = meta.and_then(|m| m.entrypoint.clone());
@@ -364,16 +363,8 @@ fn parse_conversation_jsonl(
     let mut last_ts: Option<i64> = None;
     let source_path = path.to_string_lossy().to_string();
 
-    for (line_index, line) in reader.lines().enumerate() {
-        let line = line?;
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-        let v: Value = match serde_json::from_str(line) {
-            Ok(v) => v,
-            Err(_) => continue,
-        };
+    for item in jsonl_indexed(reader.lines()) {
+        let (line_index, v) = item?;
 
         if cwd.is_none()
             && let Some(c) = v.get("cwd").and_then(|s| s.as_str())
@@ -421,11 +412,7 @@ fn parse_conversation_jsonl(
         };
 
         let text = extract_content(message.get("content"));
-        let timestamp = v
-            .get("timestamp")
-            .and_then(|t| t.as_str())
-            .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
-            .map(|dt| dt.timestamp_millis());
+        let timestamp = rfc3339_ms(v.get("timestamp"));
 
         let message_seq =
             if !is_machinery && !text.is_empty() { Some(messages.len() as u32) } else { None };
diff --git a/src/adapters/cline.rs b/src/adapters/cline.rs
index 6735867..8e03ff4 100644
--- a/src/adapters/cline.rs
+++ b/src/adapters/cline.rs
@@ -5,6 +5,7 @@ use serde_json::Value;
 use tracing::debug;
 
 use crate::adapters::file_scan::{self, FileScanEntry};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
 };
@@ -67,14 +68,10 @@ fn scan_for_sync_impl(
 }
 
 fn resolve_tasks_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home
-        .join("Library/Application Support/Code/User/globalStorage/saoudrizwan.claude-dev/tasks");
-    if !dir.exists() {
-        debug!("Cline tasks directory not found, skipping Cline");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(
+        "Library/Application Support/Code/User/globalStorage/saoudrizwan.claude-dev/tasks",
+        "Cline tasks directory not found, skipping Cline",
+    )
 }
 
 fn collect_cline_entries(tasks_dir: &Path) -> Vec<FileScanEntry> {
diff --git a/src/adapters/codex.rs b/src/adapters/codex.rs
index d4c06f4..190e1cd 100644
--- a/src/adapters/codex.rs
+++ b/src/adapters/codex.rs
@@ -8,8 +8,11 @@ use walkdir::WalkDir;
 
 use crate::adapters::events;
 use crate::adapters::file_scan::{self, FileScanEntry};
+use crate::adapters::json_util::{jsonl_indexed, rfc3339_ms};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
+    first_timestamp, last_timestamp,
 };
 use crate::db::store::Store;
 use crate::types::{RawSessionEvent, RawUsageEvent, Role, TokenSource};
@@ -98,13 +101,7 @@ fn open_url_command(url: String) -> ResumeCommand {
 }
 
 fn resolve_codex_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".codex");
-    if !dir.exists() {
-        debug!("~/.codex not found, skipping Codex");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(".codex", "~/.codex not found, skipping Codex")
 }
 
 fn scan_for_sync_impl(
@@ -226,16 +223,8 @@ fn parse_codex_session_with_options(
     let mut forked_child_inherited_reported_total: Option<i64> = None;
     let source_path = path.to_string_lossy().to_string();
 
-    for (line_index, line) in reader.lines().enumerate() {
-        let line = line?;
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-        let v: Value = match serde_json::from_str(line) {
-            Ok(v) => v,
-            Err(_) => continue,
-        };
+    for item in jsonl_indexed(reader.lines()) {
+        let (line_index, v) = item?;
 
         let msg_type = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
 
@@ -290,11 +279,7 @@ fn parse_codex_session_with_options(
                             &model,
                         );
                     }
-                    meta_timestamp = payload
-                        .get("timestamp")
-                        .and_then(|t| t.as_str())
-                        .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
-                        .map(|dt| dt.timestamp_millis());
+                    meta_timestamp = rfc3339_ms(payload.get("timestamp"));
                     if payload.get("forked_from_id").and_then(|s| s.as_str()).is_some() {
                         forked_child_waiting_for_turn_context = true;
                         forked_child_inherited_baseline = None;
@@ -425,21 +410,14 @@ fn parse_codex_session_with_options(
         path.file_stem().and_then(|s| s.to_str()).unwrap_or("unknown").to_string()
     });
 
-    let started_at = meta_timestamp
-        .or_else(|| messages.first().and_then(|m| m.timestamp))
-        .or_else(|| usage_events.first().map(|event| event.timestamp))
-        .or_else(|| events.first().and_then(|event| event.timestamp))
-        .unwrap_or(0);
+    let started_at =
+        first_timestamp(meta_timestamp, &messages, &usage_events, &events).unwrap_or(0);
 
     Ok(Some(RawSession {
         source_id,
         directory: meta_cwd,
         started_at,
-        updated_at: messages
-            .last()
-            .and_then(|m| m.timestamp)
-            .or_else(|| usage_events.last().map(|event| event.timestamp))
-            .or_else(|| events.last().and_then(|event| event.timestamp)),
+        updated_at: last_timestamp(None, &messages, &usage_events, &events),
         entrypoint: None,
         messages,
         usage_events,
@@ -850,10 +828,7 @@ fn apply_pending_codex_model(
 }
 
 fn parse_timestamp(v: &Value) -> Option<i64> {
-    v.get("timestamp")
-        .and_then(|t| t.as_str())
-        .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
-        .map(|dt| dt.timestamp_millis())
+    rfc3339_ms(v.get("timestamp"))
 }
 
 fn push_codex_message(
diff --git a/src/adapters/copilot.rs b/src/adapters/copilot.rs
index def06d8..af98720 100644
--- a/src/adapters/copilot.rs
+++ b/src/adapters/copilot.rs
@@ -8,8 +8,11 @@ use tracing::debug;
 
 use crate::adapters::events;
 use crate::adapters::file_scan::{self, FileScanEntry, FileScanOptions};
+use crate::adapters::json_util::{jsonl_indexed, rfc3339_ms};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
+    first_timestamp, last_timestamp,
 };
 use crate::db::store::Store;
 use crate::types::{RawSessionEvent, Role};
@@ -65,13 +68,10 @@ impl SourceAdapter for CopilotAdapter {
 }
 
 fn resolve_copilot_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".copilot/session-state");
-    if !dir.exists() {
-        debug!("~/.copilot/session-state not found, skipping Copilot CLI");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(
+        ".copilot/session-state",
+        "~/.copilot/session-state not found, skipping Copilot CLI",
+    )
 }
 
 fn scan_for_sync_impl(
@@ -212,16 +212,8 @@ where
     let mut messages = Vec::new();
     let mut session_events = Vec::new();
 
-    for line in lines {
-        let line = line?;
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-        let v: Value = match serde_json::from_str(line) {
-            Ok(v) => v,
-            Err(_) => continue,
-        };
+    for item in jsonl_indexed(lines) {
+        let (_, v) = item?;
 
         let event_type = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
         let timestamp = parse_timestamp(&v);
@@ -231,11 +223,7 @@ where
             "session.start" => {
                 if let Some(data) = v.get("data") {
                     session_id = data.get("sessionId").and_then(|s| s.as_str()).map(String::from);
-                    meta_started_at = data
-                        .get("startTime")
-                        .and_then(|t| t.as_str())
-                        .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
-                        .map(|dt| dt.timestamp_millis());
+                    meta_started_at = rfc3339_ms(data.get("startTime"));
                     directory = data
                         .get("context")
                         .and_then(|c| c.get("cwd"))
@@ -349,9 +337,8 @@ where
     }
 
     let source_id = session_id.unwrap_or_else(|| fallback_id.to_string());
-    let started_at =
-        meta_started_at.or_else(|| messages.first().and_then(|m| m.timestamp)).unwrap_or(0);
-    let updated_at = messages.last().and_then(|m| m.timestamp);
+    let started_at = first_timestamp(meta_started_at, &messages, &[], &[]).unwrap_or(0);
+    let updated_at = last_timestamp(None, &messages, &[], &[]);
 
     let mut session =
         RawSession::search_only(source_id, directory, started_at, updated_at, None, messages);
@@ -411,10 +398,7 @@ fn extract_tool_requests(tool_requests: Option<&Value>) -> String {
 }
 
 fn parse_timestamp(v: &Value) -> Option<i64> {
-    v.get("timestamp")
-        .and_then(|t| t.as_str())
-        .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
-        .map(|dt| dt.timestamp_millis())
+    rfc3339_ms(v.get("timestamp"))
 }
 
 #[cfg(test)]
diff --git a/src/adapters/cursor.rs b/src/adapters/cursor.rs
index 926bad7..a01c67f 100644
--- a/src/adapters/cursor.rs
+++ b/src/adapters/cursor.rs
@@ -10,8 +10,11 @@ use tracing::debug;
 use walkdir::WalkDir;
 
 use crate::adapters::events;
+use crate::adapters::json_util::json_i64;
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
+    first_timestamp, last_timestamp,
 };
 use crate::db::store::{EventSessionStateMeta, Store, UsageSessionStateMeta};
 use crate::types::{RawSessionEvent, RawUsageEvent, Role, TokenSource};
@@ -333,14 +336,17 @@ fn parse_composer_session(
         return Ok(None);
     }
 
-    let started_at = json_i64(data.get("createdAt"))
-        .or(meta.created_at)
-        .or_else(|| messages.first().and_then(|msg| msg.timestamp))
-        .unwrap_or(0);
-    let updated_at = json_i64(data.get("lastUpdatedAt"))
-        .or(json_i64(data.get("conversationCheckpointLastUpdatedAt")))
-        .or(meta.last_updated_at)
-        .or_else(|| messages.last().and_then(|msg| msg.timestamp));
+    let started_at =
+        first_timestamp(json_i64(data.get("createdAt")).or(meta.created_at), &messages, &[], &[])
+            .unwrap_or(0);
+    let updated_at = last_timestamp(
+        json_i64(data.get("lastUpdatedAt"))
+            .or(json_i64(data.get("conversationCheckpointLastUpdatedAt")))
+            .or(meta.last_updated_at),
+        &messages,
+        &[],
+        &[],
+    );
 
     Ok(Some(ParsedComposerSession {
         messages,
@@ -1005,13 +1011,7 @@ fn collect_cursor_project_key_matches(
 }
 
 fn resolve_projects_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".cursor").join("projects");
-    if !dir.exists() {
-        debug!("~/.cursor/projects not found, skipping Cursor");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(".cursor/projects", "~/.cursor/projects not found, skipping Cursor")
 }
 
 fn resolve_global_state_db_path() -> Option<PathBuf> {
@@ -1155,14 +1155,6 @@ fn render_json_fragment(value: &Value) -> Option<String> {
     }
 }
 
-fn json_i64(value: Option<&Value>) -> Option<i64> {
-    value.and_then(|value| {
-        value
-            .as_i64()
-            .or_else(|| value.as_u64().map(|value| value as i64))
-            .or_else(|| value.as_f64().map(|value| value as i64))
-    })
-}
 #[cfg(test)]
 mod tests {
     use std::io::Write;
diff --git a/src/adapters/gemini.rs b/src/adapters/gemini.rs
index 603e8ed..ab223e6 100644
--- a/src/adapters/gemini.rs
+++ b/src/adapters/gemini.rs
@@ -6,6 +6,7 @@ use serde_json::Value;
 use tracing::debug;
 use walkdir::WalkDir;
 
+use crate::adapters::json_util::json_i64;
 use crate::adapters::{RawMessage, RawSession, ResumeCommand, SourceAdapter};
 use crate::types::{RawUsageEvent, Role, TokenSource};
 
@@ -217,15 +218,6 @@ fn extract_gemini_usage_event(
     })
 }
 
-fn json_i64(value: Option<&Value>) -> Option<i64> {
-    value.and_then(|value| {
-        value
-            .as_i64()
-            .or_else(|| value.as_u64().map(|v| v as i64))
-            .or_else(|| value.as_f64().map(|v| v as i64))
-    })
-}
-
 fn extract_tool_calls(tool_calls: Option<&Value>) -> String {
     let Some(arr) = tool_calls.and_then(|v| v.as_array()) else {
         return String::new();
diff --git a/src/adapters/grok.rs b/src/adapters/grok.rs
index 2a346d3..b508d7b 100644
--- a/src/adapters/grok.rs
+++ b/src/adapters/grok.rs
@@ -7,6 +7,8 @@ use serde_json::Value;
 use tracing::debug;
 
 use crate::adapters::file_scan::{self, FileScanEntry, FileScanOptions};
+use crate::adapters::json_util::{json_i64, jsonl_indexed, rfc3339_ms};
+use crate::adapters::paths::resolve_home_dir;
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
 };
@@ -102,13 +104,7 @@ struct PendingAgentMessage {
 }
 
 fn resolve_grok_sessions_dir() -> anyhow::Result<Option<PathBuf>> {
-    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
-    let dir = home.join(".grok").join("sessions");
-    if !dir.exists() {
-        debug!("~/.grok/sessions not found, skipping Grok");
-        return Ok(None);
-    }
-    Ok(Some(dir))
+    resolve_home_dir(".grok/sessions", "~/.grok/sessions not found, skipping Grok")
 }
 
 fn scan_for_sync_impl(
@@ -332,8 +328,8 @@ fn load_grok_summary(session_dir: &Path, fallback_id: &str) -> anyhow::Result<Gr
         .filter(|model| !model.is_empty())
         .map(str::to_string);
 
-    let started_at = parse_rfc3339_ms(doc.get("created_at")).unwrap_or(0);
-    let updated_at = parse_rfc3339_ms(doc.get("updated_at"));
+    let started_at = rfc3339_ms(doc.get("created_at")).unwrap_or(0);
+    let updated_at = rfc3339_ms(doc.get("updated_at"));
     Ok(GrokSummary { session_id, directory, started_at, updated_at, current_model_id })
 }
 
@@ -352,16 +348,8 @@ fn parse_grok_updates_reader<R: BufRead>(
     let mut prompts: Vec<PromptTokenState> = Vec::new();
     let mut prompt_index: HashMap<String, usize> = HashMap::new();
     let mut last_model: Option<String> = None;
-    for line in reader.lines() {
-        let line = line?;
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-
-        let Ok(doc) = serde_json::from_str::<Value>(line) else {
-            continue;
-        };
+    for item in jsonl_indexed(reader.lines()) {
+        let (_, doc) = item?;
 
         let params = match doc.get("params") {
             Some(params) => params,
@@ -373,7 +361,7 @@ fn parse_grok_updates_reader<R: BufRead>(
         };
         let session_update =
             update.get("sessionUpdate").and_then(|value| value.as_str()).unwrap_or("");
-        let timestamp_ms = doc.get("timestamp").and_then(json_i64).map(|ts| ts * 1000);
+        let timestamp_ms = json_i64(doc.get("timestamp")).map(|ts| ts * 1000);
         track_prompt_tokens(params, timestamp_ms, &mut prompts, &mut prompt_index, &mut last_model);
 
         match session_update {
@@ -448,7 +436,7 @@ fn track_prompt_tokens(
     else {
         return;
     };
-    let Some(total_tokens) = meta.get("totalTokens").and_then(json_i64) else {
+    let Some(total_tokens) = json_i64(meta.get("totalTokens")) else {
         return;
     };
     let index = match prompt_index.get(prompt_id) {
@@ -512,7 +500,7 @@ fn agent_chunk_key(params: &Value) -> AgentChunkKey {
             .and_then(|prompt| prompt.as_str())
             .filter(|prompt| !prompt.is_empty())
             .map(str::to_string),
-        turn_start_ms: meta.and_then(|meta| meta.get("turnStartMs")).and_then(json_i64),
+        turn_start_ms: json_i64(meta.and_then(|meta| meta.get("turnStartMs"))),
     }
 }
 
@@ -625,18 +613,6 @@ fn format_tool_call_result(update: &Value) -> Option<String> {
     Some(out)
 }
 
-fn parse_rfc3339_ms(value: Option<&Value>) -> Option<i64> {
-    let text = value.and_then(|value| value.as_str())?;
-    chrono::DateTime::parse_from_rfc3339(text).ok().map(|dt| dt.timestamp_millis())
-}
-
-fn json_i64(value: &Value) -> Option<i64> {
-    value
-        .as_i64()
-        .or_else(|| value.as_u64().map(|value| value as i64))
-        .or_else(|| value.as_f64().map(|value| value as i64))
-}
-
 #[cfg(test)]
 mod tests {
     use std::io::Cursor;
diff --git a/src/adapters/json_util.rs b/src/adapters/json_util.rs
new file mode 100644
index 0000000..b74dcd5
--- /dev/null
+++ b/src/adapters/json_util.rs
@@ -0,0 +1,32 @@
+use std::io;
+
+use serde_json::Value;
+
+pub(crate) fn rfc3339_ms(value: Option<&Value>) -> Option<i64> {
+    let text = value?.as_str()?;
+    chrono::DateTime::parse_from_rfc3339(text).ok().map(|dt| dt.timestamp_millis())
+}
+
+pub(crate) fn json_i64(value: Option<&Value>) -> Option<i64> {
+    let value = value?;
+    value
+        .as_i64()
+        .or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
+        .or_else(|| value.as_f64().map(|value| value as i64))
+}
+
+pub(crate) fn jsonl_indexed(
+    lines: impl IntoIterator<Item = io::Result<String>>,
+) -> impl Iterator<Item = io::Result<(usize, Value)>> {
+    lines.into_iter().enumerate().filter_map(|(index, line)| match line {
+        Ok(line) => {
+            let trimmed = line.trim();
+            if trimmed.is_empty() {
+                None
+            } else {
+                serde_json::from_str::<Value>(trimmed).ok().map(|value| Ok((index, value)))
+            }
+        }
+        Err(error) => Some(Err(error)),
+    })
+}
diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs
index d3c86fd..dd89569 100644
--- a/src/adapters/mod.rs
+++ b/src/adapters/mod.rs
@@ -8,8 +8,10 @@ pub(crate) mod events;
 pub(crate) mod file_scan;
 pub(crate) mod gemini;
 pub(crate) mod grok;
+pub(crate) mod json_util;
 pub(crate) mod kiro;
 pub(crate) mod opencode;
+pub(crate) mod paths;
 pub(crate) mod pi;
 pub(crate) mod sync_state;
 
@@ -110,6 +112,28 @@ pub(crate) struct RawMessage {
     pub(crate) timestamp: Option<i64>,
 }
 
+pub(crate) fn first_timestamp(
+    meta: Option<i64>,
+    messages: &[RawMessage],
+    usage_events: &[RawUsageEvent],
+    events: &[RawSessionEvent],
+) -> Option<i64> {
+    meta.or_else(|| messages.first().and_then(|message| message.timestamp))
+        .or_else(|| usage_events.first().map(|event| event.timestamp))
+        .or_else(|| events.first().and_then(|event| event.timestamp))
+}
+
+pub(crate) fn last_timestamp(
+    meta: Option<i64>,
+    messages: &[RawMessage],
+    usage_events: &[RawUsageEvent],
+    events: &[RawSessionEvent],
+) -> Option<i64> {
+    meta.or_else(|| messages.last().and_then(|message| message.timestamp))
+        .or_else(|| usage_events.last().map(|event| event.timestamp))
+        .or_else(|| events.last().and_then(|event| event.timestamp))
+}
+
 #[derive(Default)]
 pub(crate) struct SyncScanStats {
     pub(crate) skipped_sessions: u32,
diff --git a/src/adapters/paths.rs b/src/adapters/paths.rs
new file mode 100644
index 0000000..260f895
--- /dev/null
+++ b/src/adapters/paths.rs
@@ -0,0 +1,14 @@
+use std::path::PathBuf;
+
+pub(crate) fn resolve_home_dir(
+    relative: &str,
+    missing_message: &str,
+) -> anyhow::Result<Option<PathBuf>> {
+    let home = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("no home dir"))?;
+    let dir = home.join(relative);
+    if !dir.exists() {
+        tracing::debug!("{missing_message}");
+        return Ok(None);
+    }
+    Ok(Some(dir))
+}
diff --git a/src/adapters/pi.rs b/src/adapters/pi.rs
index 23135b0..8da9ad7 100644
--- a/src/adapters/pi.rs
+++ b/src/adapters/pi.rs
@@ -8,8 +8,10 @@ use tracing::debug;
 use walkdir::WalkDir;
 
 use crate::adapters::file_scan::{self, FileScanEntry};
+use crate::adapters::json_util::{json_i64, jsonl_indexed, rfc3339_ms};
 use crate::adapters::{
     RawMessage, RawSession, ResumeCommand, SourceAdapter, SyncScanResult, SyncScanStats,
+    first_timestamp,
 };
 use crate::db::store::Store;
 use crate::types::{RawUsageEvent, Role, TokenSource};
@@ -259,11 +261,9 @@ fn parse_pi_session_file(
         return Ok(None);
     }
 
-    let started_at = parsed
-        .started_at
-        .or_else(|| parsed.messages.first().and_then(|message| message.timestamp))
-        .or_else(|| parsed.usage_events.first().map(|event| event.timestamp))
-        .unwrap_or(0);
+    let started_at =
+        first_timestamp(parsed.started_at, &parsed.messages, &parsed.usage_events, &[])
+            .unwrap_or(0);
 
     Ok(Some(RawSession {
         source_id: parsed.session_id.unwrap_or(entry.session_id),
@@ -297,17 +297,8 @@ fn parse_pi_session(path: &Path, fallback_timestamp: i64) -> anyhow::Result<Pars
     let mut messages = Vec::new();
     let mut usage_events = Vec::new();
 
-    for (line_index, line) in reader.lines().enumerate() {
-        let line = line?;
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-
-        let entry: Value = match serde_json::from_str(line) {
-            Ok(entry) => entry,
-            Err(_) => continue,
-        };
+    for item in jsonl_indexed(reader.lines()) {
+        let (line_index, entry) = item?;
 
         match entry.get("type").and_then(|value| value.as_str()).unwrap_or("") {
             "session" => {
@@ -407,9 +398,7 @@ fn parse_pi_message(
     messages: &mut Vec<RawMessage>,
     usage_events: &mut Vec<RawUsageEvent>,
 ) {
-    let timestamp = message
-        .get("timestamp")
-        .and_then(number_to_i64)
+    let timestamp = json_i64(message.get("timestamp"))
         .or_else(|| parse_entry_timestamp(entry))
         .unwrap_or(fallback_timestamp);
 
@@ -551,7 +540,7 @@ fn non_empty_str(value: Option<&Value>) -> Option<&str> {
 }
 
 fn usage_count(usage: &Value, keys: &[&str]) -> i64 {
-    keys.iter().find_map(|key| usage.get(*key).and_then(number_to_i64)).unwrap_or(0).max(0)
+    keys.iter().find_map(|key| json_i64(usage.get(*key))).unwrap_or(0).max(0)
 }
 
 fn extract_content(content: Option<&Value>) -> String {
@@ -630,18 +619,7 @@ fn extract_bash_execution_content(message: &Value) -> String {
 }
 
 fn parse_entry_timestamp(entry: &Value) -> Option<i64> {
-    entry
-        .get("timestamp")
-        .and_then(|value| value.as_str())
-        .and_then(|timestamp| chrono::DateTime::parse_from_rfc3339(timestamp).ok())
-        .map(|timestamp| timestamp.timestamp_millis())
-}
-
-fn number_to_i64(value: &Value) -> Option<i64> {
-    value
-        .as_i64()
-        .or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
-        .or_else(|| value.as_f64().map(|value| value as i64))
+    rfc3339_ms(entry.get("timestamp"))
 }
 
 #[cfg(test)]