diff --git a/bt-daemon/src/agents/antigravity.rs b/bt-daemon/src/agents/antigravity.rs index 1216b92..3358a58 100644 --- a/bt-daemon/src/agents/antigravity.rs +++ b/bt-daemon/src/agents/antigravity.rs @@ -19,6 +19,7 @@ impl Agent for Antigravity { impl Register for Antigravity { fn register(self: Arc, registrar: &mut Registrar) { registrar.setup.insert(self.clone()); - registrar.translate.insert(self); + registrar.translate.insert(self.clone()); + registrar.import.insert(self); } } diff --git a/bt-daemon/src/agents/claude.rs b/bt-daemon/src/agents/claude.rs index 396f68a..215ce59 100644 --- a/bt-daemon/src/agents/claude.rs +++ b/bt-daemon/src/agents/claude.rs @@ -20,6 +20,7 @@ impl Register for Claude { fn register(self: Arc, registrar: &mut Registrar) { registrar.setup.insert(self.clone()); registrar.run.insert(self.clone()); - registrar.translate.insert(self); + registrar.translate.insert(self.clone()); + registrar.import.insert(self); } } diff --git a/bt-daemon/src/agents/codex.rs b/bt-daemon/src/agents/codex.rs index 75f9a74..2f68ed0 100644 --- a/bt-daemon/src/agents/codex.rs +++ b/bt-daemon/src/agents/codex.rs @@ -20,6 +20,7 @@ impl Register for Codex { fn register(self: Arc, registrar: &mut Registrar) { registrar.setup.insert(self.clone()); registrar.run.insert(self.clone()); - registrar.translate.insert(self); + registrar.translate.insert(self.clone()); + registrar.import.insert(self); } } diff --git a/bt-daemon/src/agents/cursor.rs b/bt-daemon/src/agents/cursor.rs index 56b5038..d975c4a 100644 --- a/bt-daemon/src/agents/cursor.rs +++ b/bt-daemon/src/agents/cursor.rs @@ -20,6 +20,7 @@ impl Register for Cursor { fn register(self: Arc, registrar: &mut Registrar) { registrar.setup.insert(self.clone()); registrar.run.insert(self.clone()); - registrar.translate.insert(self); + registrar.translate.insert(self.clone()); + registrar.import.insert(self); } } diff --git a/bt-daemon/src/agents/mod.rs b/bt-daemon/src/agents/mod.rs index 6303881..cc86aed 100644 --- a/bt-daemon/src/agents/mod.rs +++ b/bt-daemon/src/agents/mod.rs @@ -23,6 +23,7 @@ pub(crate) use pi::Pi; use crate::managed_run::ManagedRun; use crate::setup::Setup; +use crate::transcript_import::TranscriptImport; use crate::translate::TranslatorFactory; use std::sync::{Arc, OnceLock}; @@ -85,6 +86,7 @@ pub(crate) struct Registrar { pub setup: Agents, pub run: Agents, pub translate: Agents, + pub import: Agents, } pub(crate) trait Register { diff --git a/bt-daemon/src/transcript_import/antigravity.rs b/bt-daemon/src/transcript_import/antigravity.rs index bd52456..85cb5e4 100644 --- a/bt-daemon/src/transcript_import/antigravity.rs +++ b/bt-daemon/src/transcript_import/antigravity.rs @@ -1,4 +1,7 @@ -use super::{envelope, validate_session_id}; +use super::{ + envelope, validate_session_id, IncrementalRecords, TailSession, TranscriptImport, + TranscriptLayout, +}; use crate::wire::Envelope; use anyhow::bail; use serde_json::{json, Value}; @@ -7,14 +10,36 @@ use std::path::{Path, PathBuf}; const TRANSCRIPT_NAME: &str = "transcript_full.jsonl"; +impl TranscriptImport for crate::agents::Antigravity { + fn discover(&self, home: &Path) -> anyhow::Result> { + layout(roots(home)).discover() + } + + fn find(&self, home: &Path, session_id: &str) -> anyhow::Result { + layout(roots(home)).find(session_id) + } + + fn tail(&self) -> Box { + Box::new(Tail::default()) + } +} + #[derive(Default)] pub(super) struct Tail { emitted: usize, stopped: bool, } -impl Tail { - pub(super) fn poll( +impl TailSession for Tail { + fn envelopes( + &mut self, + path: &Path, + records: &IncrementalRecords, + ) -> anyhow::Result> { + envelopes(path, &records.values, &records.end_offsets) + } + + fn poll( &mut self, events: Vec, _len: u64, @@ -36,11 +61,21 @@ impl Tail { } } -pub(super) fn roots(home: &Path) -> Vec { +/// The transcript layout under these roots. +pub(super) fn layout(roots: Vec) -> TranscriptLayout { + TranscriptLayout { + display_name: crate::agents::Agent::identity(&crate::agents::Antigravity).display_name, + roots, + session_id: transcript_session_id, + filename_matches, + } +} + +fn roots(home: &Path) -> Vec { vec![home.join(".gemini/antigravity-cli/brain")] } -pub(super) fn transcript_session_id(path: &Path) -> Option { +fn transcript_session_id(path: &Path) -> Option { if path.file_name().and_then(|name| name.to_str()) != Some(TRANSCRIPT_NAME) { return None; } @@ -54,7 +89,7 @@ pub(super) fn transcript_session_id(path: &Path) -> Option { is_transcript_path(path, session_id).then(|| session_id.to_owned()) } -pub(super) fn filename_matches(path: &Path, session_id: &str) -> bool { +fn filename_matches(path: &Path, session_id: &str) -> bool { validate_session_id(session_id).is_ok() && is_transcript_path(path, session_id) } diff --git a/bt-daemon/src/transcript_import/claude.rs b/bt-daemon/src/transcript_import/claude.rs index 74e7f96..d7eccaa 100644 --- a/bt-daemon/src/transcript_import/claude.rs +++ b/bt-daemon/src/transcript_import/claude.rs @@ -1,6 +1,7 @@ use super::{ envelope, file_session_id, find_jsonl_files, read_complete_jsonl_records, read_jsonl_records, - string_at, timestamp_bounds, timestamp_ms, validate_session_id, + string_at, timestamp_bounds, timestamp_ms, validate_session_id, IncrementalRecords, + TailSession, TranscriptImport, TranscriptLayout, }; use crate::wire::Envelope; use anyhow::bail; @@ -9,6 +10,20 @@ use std::collections::{HashMap, HashSet}; use std::io::BufRead; use std::path::{Path, PathBuf}; +impl TranscriptImport for crate::agents::Claude { + fn discover(&self, home: &Path) -> anyhow::Result> { + layout(roots(home)).discover() + } + + fn find(&self, home: &Path, session_id: &str) -> anyhow::Result { + layout(roots(home)).find(session_id) + } + + fn tail(&self) -> Box { + Box::new(Tail::default()) + } +} + #[derive(Default)] pub(super) struct Tail { started: bool, @@ -17,8 +32,21 @@ pub(super) struct Tail { last_len: u64, } -impl Tail { - pub(super) fn poll( +impl TailSession for Tail { + fn envelopes( + &mut self, + path: &Path, + records: &IncrementalRecords, + ) -> anyhow::Result> { + envelopes( + path, + &records.values, + &records.end_offsets, + records.read_offset, + ) + } + + fn poll( &mut self, events: Vec, len: u64, @@ -76,7 +104,7 @@ impl Tail { } } -pub(super) fn transcript_session_id(path: &Path) -> Option { +fn transcript_session_id(path: &Path) -> Option { let file = std::fs::File::open(path).ok()?; for line in std::io::BufReader::new(file).lines().map_while(Result::ok) { let Ok(record) = serde_json::from_str::(&line) else { @@ -93,14 +121,24 @@ pub(super) fn transcript_session_id(path: &Path) -> Option { None } -pub(super) fn roots(home: &Path) -> Vec { +/// The transcript layout under these roots. +pub(super) fn layout(roots: Vec) -> TranscriptLayout { + TranscriptLayout { + display_name: crate::agents::Agent::identity(&crate::agents::Claude).display_name, + roots, + session_id: transcript_session_id, + filename_matches, + } +} + +fn roots(home: &Path) -> Vec { let claude_home = std::env::var_os("CLAUDE_CONFIG_DIR") .map(PathBuf::from) .unwrap_or_else(|| home.join(".claude")); vec![claude_home.join("projects")] } -pub(super) fn filename_matches(path: &Path, session_id: &str) -> bool { +fn filename_matches(path: &Path, session_id: &str) -> bool { path.file_name().and_then(|name| name.to_str()) == Some(&format!("{session_id}.jsonl")) } diff --git a/bt-daemon/src/transcript_import/codex.rs b/bt-daemon/src/transcript_import/codex.rs index 09f5895..bde0947 100644 --- a/bt-daemon/src/transcript_import/codex.rs +++ b/bt-daemon/src/transcript_import/codex.rs @@ -1,6 +1,7 @@ use super::{ envelope, file_session_id, find_jsonl_files, read_jsonl_records, string_at, timestamp_bounds, - timestamp_ms, validate_session_id, + timestamp_ms, validate_session_id, IncrementalRecords, TailSession, TranscriptImport, + TranscriptLayout, }; use crate::wire::Envelope; use anyhow::bail; @@ -9,6 +10,20 @@ use std::collections::{HashMap, HashSet}; use std::io::BufRead; use std::path::{Path, PathBuf}; +impl TranscriptImport for crate::agents::Codex { + fn discover(&self, home: &Path) -> anyhow::Result> { + layout(roots(home)).discover() + } + + fn find(&self, home: &Path, session_id: &str) -> anyhow::Result { + layout(roots(home)).find(session_id) + } + + fn tail(&self) -> Box { + Box::new(Tail::default()) + } +} + #[derive(Default)] pub(super) struct Tail { started: bool, @@ -16,8 +31,16 @@ pub(super) struct Tail { last_len: u64, } -impl Tail { - pub(super) fn poll( +impl TailSession for Tail { + fn envelopes( + &mut self, + path: &Path, + records: &IncrementalRecords, + ) -> anyhow::Result> { + envelopes(path, &records.values) + } + + fn poll( &mut self, events: Vec, len: u64, @@ -64,7 +87,7 @@ impl Tail { } } -pub(super) fn transcript_session_id(path: &Path) -> Option { +fn transcript_session_id(path: &Path) -> Option { let file = std::fs::File::open(path).ok()?; for line in std::io::BufReader::new(file).lines().map_while(Result::ok) { let Ok(record) = serde_json::from_str::(&line) else { @@ -85,7 +108,17 @@ pub(super) fn transcript_session_id(path: &Path) -> Option { None } -pub(super) fn roots(home: &Path) -> Vec { +/// The transcript layout under these roots. +pub(super) fn layout(roots: Vec) -> TranscriptLayout { + TranscriptLayout { + display_name: crate::agents::Agent::identity(&crate::agents::Codex).display_name, + roots, + session_id: transcript_session_id, + filename_matches, + } +} + +fn roots(home: &Path) -> Vec { let codex_home = std::env::var_os("CODEX_HOME") .map(PathBuf::from) .unwrap_or_else(|| home.join(".codex")); @@ -95,7 +128,7 @@ pub(super) fn roots(home: &Path) -> Vec { ] } -pub(super) fn filename_matches(path: &Path, session_id: &str) -> bool { +fn filename_matches(path: &Path, session_id: &str) -> bool { path.file_name() .and_then(|name| name.to_str()) .is_some_and(|name| name.ends_with(&format!("{session_id}.jsonl"))) diff --git a/bt-daemon/src/transcript_import/cursor.rs b/bt-daemon/src/transcript_import/cursor.rs index 7a8cdfb..af5000a 100644 --- a/bt-daemon/src/transcript_import/cursor.rs +++ b/bt-daemon/src/transcript_import/cursor.rs @@ -1,18 +1,32 @@ -use super::{envelope, validate_session_id}; +use super::{ + envelope, validate_session_id, IncrementalRecords, TailSession, TranscriptImport, + TranscriptLayout, +}; use crate::translate::cursor::{IMPORT_CHECKPOINT, IMPORT_START, IMPORT_STOP}; use crate::wire::Envelope; use anyhow::{bail, Context}; use serde_json::{json, Value}; +use std::io::Write; use std::path::{Path, PathBuf}; /// Cursor stores transcript conversations under /// ~/.cursor/projects//agent-transcripts//.jsonl. /// The directory and filename are the only session identity in this format. -pub(super) fn roots(home: &Path) -> Vec { +/// The transcript layout under these roots. +pub(super) fn layout(roots: Vec) -> TranscriptLayout { + TranscriptLayout { + display_name: crate::agents::Agent::identity(&crate::agents::Cursor).display_name, + roots, + session_id: transcript_session_id, + filename_matches, + } +} + +fn roots(home: &Path) -> Vec { vec![home.join(".cursor/projects")] } -pub(super) fn transcript_session_id(path: &Path) -> Option { +fn transcript_session_id(path: &Path) -> Option { let session_id = path.file_stem()?.to_str()?; // Cursor writes child agent transcripts with an `agent-` identifier. // They do not carry a recoverable parent tool-call link, so they are not @@ -24,7 +38,7 @@ pub(super) fn transcript_session_id(path: &Path) -> Option { filename_matches(path, session_id).then(|| session_id.to_owned()) } -pub(super) fn filename_matches(path: &Path, session_id: &str) -> bool { +fn filename_matches(path: &Path, session_id: &str) -> bool { !session_id.starts_with("agent-") && validate_session_id(session_id).is_ok() && path.extension().and_then(|value| value.to_str()) == Some("jsonl") @@ -52,15 +66,88 @@ fn is_conversation_record(record: &Value) -> bool { .is_some_and(|message| message.get("content").is_some()) } +impl TranscriptImport for crate::agents::Cursor { + fn discover(&self, home: &Path) -> anyhow::Result> { + layout(roots(home)).discover() + } + + fn find(&self, home: &Path, session_id: &str) -> anyhow::Result { + layout(roots(home)).find(session_id) + } + + fn tail(&self) -> Box { + Box::new(Tail::default()) + } +} + +/// Cursor rewrites its transcript between turns. The import keeps one +/// translator across rewrites, so it reads a private snapshot that only ever +/// grows: records that survive a rewrite are not appended again. #[derive(Default)] pub(super) struct Tail { started: bool, last_len: u64, stopped: bool, + snapshot: Option, + snapshot_records: usize, } -impl Tail { - pub(super) fn poll( +impl TailSession for Tail { + fn envelopes( + &mut self, + path: &Path, + records: &IncrementalRecords, + ) -> anyhow::Result> { + if self.snapshot.is_none() { + self.snapshot = Some( + tempfile::Builder::new() + .prefix("bt-cursor-import-") + .tempdir()?, + ); + } + let snapshot = self + .snapshot + .as_ref() + .unwrap() + .path() + .join("transcript.jsonl"); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&snapshot)?; + for record in &records.values[self.snapshot_records..] { + serde_json::to_writer(&mut file, record)?; + file.write_all(b"\n")?; + } + file.flush()?; + let through = file.metadata()?.len(); + self.snapshot_records = records.values.len(); + envelopes_with_snapshot(path, &snapshot, through, &records.values) + } + + // Keep Cursor's translator alive so its session root and turn ordinal + // continue across native transcript replacement. A fresh snapshot path + // makes it rewind only transcript input. + fn keeps_translator_on_rewrite(&self) -> bool { + true + } + + // Cursor can rewrite a transcript while retaining earlier turns. Keep + // those records out of the rotated snapshot so the live translator does + // not claim their prompts twice. + fn rewritten(&mut self, previous: &[Value], current: &[Value]) { + self.snapshot_records = current + .iter() + .zip(previous) + .take_while(|(current, previous)| current == previous) + .count(); + } + + fn tolerates_incomplete_final_record(&self) -> bool { + true + } + + fn poll( &mut self, events: Vec, len: u64, @@ -86,15 +173,7 @@ impl Tail { } } -pub(super) fn envelopes( - path: &Path, - through: u64, - records: &[Value], -) -> anyhow::Result> { - envelopes_with_snapshot(path, path, through, records) -} - -pub(super) fn envelopes_with_snapshot( +fn envelopes_with_snapshot( path: &Path, snapshot: &Path, through: u64, diff --git a/bt-daemon/src/transcript_import/mod.rs b/bt-daemon/src/transcript_import/mod.rs index 034e01d..6d71345 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -1,9 +1,10 @@ +use crate::agents::{registrar, Agent}; use crate::wire::Envelope; use crate::ImportSource; use anyhow::{bail, Context}; use serde_json::Value; use std::collections::BTreeMap; -use std::io::{BufRead, Read, Seek, Write}; +use std::io::{BufRead, Read, Seek}; use std::path::{Path, PathBuf}; mod antigravity; @@ -14,96 +15,166 @@ mod pipeline; pub use pipeline::{import_transcript, import_transcripts, run_import}; +/// Importing one agent's native transcripts. Implemented in each agent's +/// module here and registered in [`crate::agents`]. +pub(crate) trait TranscriptImport: Agent { + /// Every session transcript the agent has saved under `home`. + fn discover(&self, home: &Path) -> anyhow::Result>; + /// The saved transcript for one session under `home`. + fn find(&self, home: &Path, session_id: &str) -> anyhow::Result; + /// State for following one growing transcript. + fn tail(&self) -> Box; +} + +/// Where an agent saves its transcripts and how they are named. Each agent +/// builds one privately; this holds the search shared by all of them. +pub(super) struct TranscriptLayout { + pub display_name: &'static str, + /// Directories searched recursively for `.jsonl` transcripts. + pub roots: Vec, + /// The session a transcript belongs to, if the path is one. + pub session_id: fn(&Path) -> Option, + /// Whether a path is named for a session id. + pub filename_matches: fn(&Path, &str) -> bool, +} + +/// Follows one growing transcript, turning the records read so far into +/// synthetic hook events and emitting only those not emitted before. +pub(crate) trait TailSession: Send { + /// Synthetic hook events for every record read so far. + fn envelopes( + &mut self, + path: &Path, + records: &IncrementalRecords, + ) -> anyhow::Result>; + + /// The events from `events` to emit now. `finalize` ends the session. + fn poll( + &mut self, + events: Vec, + len: u64, + finalize: bool, + ) -> anyhow::Result>; + + /// Whether a rewritten transcript continues with the same translator + /// instead of starting a new one. If so, the fresh tail that replaces this + /// one receives [`Self::rewritten`]. + fn keeps_translator_on_rewrite(&self) -> bool { + false + } + + /// Called on a fresh tail after a rewrite, with the records from before + /// and after it, when [`Self::keeps_translator_on_rewrite`] is true. + fn rewritten(&mut self, _previous: &[Value], _current: &[Value]) {} + + /// Whether an attached import may stop at a final record that is still + /// being written when it shuts down. + fn tolerates_incomplete_final_record(&self) -> bool { + false + } +} + +fn importer(source: ImportSource) -> &'static dyn TranscriptImport { + let name = source.identity().id; + registrar() + .import + .get(name) + .unwrap_or_else(|| panic!("no transcript import is registered for {name}")) +} + pub(crate) fn resolve_transcripts( session_ids: &[String], all: bool, source: ImportSource, ) -> anyhow::Result> { + let home = std::env::var_os("HOME") + .or_else(|| std::env::var_os("USERPROFILE")) + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from(".")); + let importer = importer(source); match (all, session_ids.is_empty()) { - (true, true) => discover_transcripts(source), + (true, true) => importer.discover(&home), (false, false) => session_ids .iter() - .map(|session_id| resolve_transcript(session_id, source)) + .map(|session_id| importer.find(&home, session_id)) .collect(), (true, false) => bail!("--all cannot be combined with explicit session ids"), (false, true) => bail!("provide at least one session id or use --all"), } } -pub(crate) fn resolve_transcript( - session_id: &str, - source: ImportSource, -) -> anyhow::Result { - validate_session_id(session_id)?; - resolve_transcript_in(session_id, source, &transcript_roots(source)) -} - -fn transcript_roots(source: ImportSource) -> Vec { - let home = std::env::var_os("HOME") - .or_else(|| std::env::var_os("USERPROFILE")) - .map(PathBuf::from) - .unwrap_or_else(|| PathBuf::from(".")); - match source { - ImportSource::Codex => codex::roots(&home), - ImportSource::Cursor => cursor::roots(&home), - ImportSource::Claude => claude::roots(&home), - ImportSource::Antigravity => antigravity::roots(&home), +impl TranscriptLayout { + /// One transcript per session found under the roots, in path order. + pub fn discover(&self) -> anyhow::Result> { + let mut sessions = BTreeMap::>::new(); + for path in self.candidates() { + if let Some(session_id) = (self.session_id)(&path) { + sessions.entry(session_id).or_default().push(path); + } + } + if sessions.is_empty() { + bail!( + "no {} transcripts found; searched {}", + self.display_name, + self.locations() + ); + } + sessions + .into_iter() + .map(|(session_id, paths)| self.single(&session_id, paths)) + .collect() } -} -fn discover_transcripts(source: ImportSource) -> anyhow::Result> { - let roots = transcript_roots(source); - discover_transcripts_in(source, &roots) -} - -fn discover_transcripts_in( - source: ImportSource, - roots: &[PathBuf], -) -> anyhow::Result> { - let mut candidates = Vec::new(); - for root in roots { - find_jsonl_files(root, &mut candidates); + /// The one transcript named for `session_id`. + pub fn find(&self, session_id: &str) -> anyhow::Result { + validate_session_id(session_id)?; + let matches: Vec<_> = self + .candidates() + .into_iter() + .filter(|path| (self.filename_matches)(path, session_id)) + .collect(); + if matches.is_empty() { + bail!( + "no {} transcript found for session {session_id}; searched {}", + self.display_name, + self.locations() + ); + } + self.single(session_id, matches) } - candidates.sort(); - candidates.dedup(); - let mut sessions = BTreeMap::>::new(); - for path in candidates { - let Some(session_id) = transcript_session_id(&path, source) else { - continue; - }; - sessions.entry(session_id).or_default().push(path); - } - if sessions.is_empty() { - let locations = roots - .iter() - .map(|root| root.display().to_string()) - .collect::>() - .join(", "); - bail!( - "no {} transcripts found; searched {locations}", - source_name(source) - ); + fn candidates(&self) -> Vec { + let mut candidates = Vec::new(); + for root in &self.roots { + find_jsonl_files(root, &mut candidates); + } + candidates.sort(); + candidates.dedup(); + candidates } - let mut resolved = Vec::with_capacity(sessions.len()); - for (session_id, paths) in sessions { - match paths.as_slice() { - [path] => resolved.push(path.clone()), - paths => { - let locations = paths + fn single(&self, session_id: &str, paths: Vec) -> anyhow::Result { + match <[PathBuf; 1]>::try_from(paths) { + Ok([path]) => Ok(path), + Err(paths) => bail!( + "multiple {} transcripts found for session {session_id}: {}", + self.display_name, + paths .iter() .map(|path| path.display().to_string()) .collect::>() - .join(", "); - bail!( - "multiple {} transcripts found for session {session_id}: {locations}", - source_name(source) - ); - } + .join(", ") + ), } } - Ok(resolved) + + fn locations(&self) -> String { + self.roots + .iter() + .map(|root| root.display().to_string()) + .collect::>() + .join(", ") + } } fn find_jsonl_files(directory: &Path, matches: &mut Vec) { @@ -126,61 +197,6 @@ fn find_jsonl_files(directory: &Path, matches: &mut Vec) { } } -fn transcript_session_id(path: &Path, source: ImportSource) -> Option { - match source { - ImportSource::Codex => codex::transcript_session_id(path), - ImportSource::Cursor => cursor::transcript_session_id(path), - ImportSource::Claude => claude::transcript_session_id(path), - ImportSource::Antigravity => antigravity::transcript_session_id(path), - } -} - -fn resolve_transcript_in( - session_id: &str, - source: ImportSource, - roots: &[PathBuf], -) -> anyhow::Result { - validate_session_id(session_id)?; - let mut matches = Vec::new(); - for root in roots { - let mut candidates = Vec::new(); - find_jsonl_files(root, &mut candidates); - matches.extend(candidates.into_iter().filter(|path| match source { - ImportSource::Codex => codex::filename_matches(path, session_id), - ImportSource::Cursor => cursor::filename_matches(path, session_id), - ImportSource::Claude => claude::filename_matches(path, session_id), - ImportSource::Antigravity => antigravity::filename_matches(path, session_id), - })); - } - matches.sort(); - matches.dedup(); - match matches.as_slice() { - [path] => Ok(path.clone()), - [] => { - let locations = roots - .iter() - .map(|root| root.display().to_string()) - .collect::>() - .join(", "); - bail!( - "no {} transcript found for session {session_id}; searched {locations}", - source_name(source) - ) - } - paths => { - let locations = paths - .iter() - .map(|path| path.display().to_string()) - .collect::>() - .join(", "); - bail!( - "multiple {} transcripts found for session {session_id}: {locations}", - source_name(source) - ) - } - } -} - fn validate_session_id(session_id: &str) -> anyhow::Result<()> { if session_id.is_empty() || !session_id @@ -192,10 +208,6 @@ fn validate_session_id(session_id: &str) -> anyhow::Result<()> { Ok(()) } -fn source_name(source: ImportSource) -> &'static str { - source.identity().display_name -} - #[cfg(test)] pub(crate) fn transcript_envelopes( path: &Path, @@ -203,31 +215,13 @@ pub(crate) fn transcript_envelopes( ) -> anyhow::Result> { let mut records = IncrementalRecords::default(); records.refresh_with_final_partial(path, true, false)?; - envelopes_from_records(path, source, &records) -} - -fn envelopes_from_records( - path: &Path, - source: ImportSource, - records: &IncrementalRecords, -) -> anyhow::Result> { - match source { - ImportSource::Codex => codex::envelopes(path, &records.values), - ImportSource::Cursor => cursor::envelopes(path, records.read_offset, &records.values), - ImportSource::Claude => claude::envelopes( - path, - &records.values, - &records.end_offsets, - records.read_offset, - ), - ImportSource::Antigravity => { - antigravity::envelopes(path, &records.values, &records.end_offsets) - } - } + importer(source).tail().envelopes(path, &records) } +/// The records of a transcript read so far, re-read incrementally as it +/// grows. #[derive(Default)] -struct IncrementalRecords { +pub(crate) struct IncrementalRecords { values: Vec, end_offsets: Vec, read_offset: u64, @@ -359,34 +353,24 @@ fn read_anchor(path: &Path, through: u64, max_len: usize) -> anyhow::Result, records: IncrementalRecords, retry_envelopes: bool, translator_reset: bool, - cursor_snapshot: Option, - cursor_snapshot_records: usize, allow_incomplete_final_record: bool, } -enum TailState { - Codex(codex::Tail), - Cursor(cursor::Tail), - Claude(claude::Tail), - Antigravity(antigravity::Tail), -} - impl TranscriptTail { pub(crate) fn new(path: PathBuf, source: ImportSource) -> Self { + let importer = importer(source); Self { path, - source, - state: Self::new_state(source), + importer, + session: importer.tail(), records: IncrementalRecords::default(), retry_envelopes: false, translator_reset: false, - cursor_snapshot: None, - cursor_snapshot_records: 0, allow_incomplete_final_record: false, } } @@ -395,17 +379,9 @@ impl TranscriptTail { self.allow_incomplete_final_record = true; } - fn new_state(source: ImportSource) -> TailState { - match source { - ImportSource::Codex => TailState::Codex(codex::Tail::default()), - ImportSource::Cursor => TailState::Cursor(cursor::Tail::default()), - ImportSource::Claude => TailState::Claude(claude::Tail::default()), - ImportSource::Antigravity => TailState::Antigravity(antigravity::Tail::default()), - } - } - pub(crate) fn poll(&mut self, finalize: bool) -> anyhow::Result> { - let previous_cursor_records = if self.source == ImportSource::Cursor { + let keeps_translator = self.session.keeps_translator_on_rewrite(); + let previous_records = if keeps_translator { self.records.values.clone() } else { Vec::new() @@ -413,7 +389,7 @@ impl TranscriptTail { let refresh = match self.records.refresh_with_final_partial( &self.path, finalize, - self.source == ImportSource::Cursor && self.allow_incomplete_final_record, + self.session.tolerates_incomplete_final_record() && self.allow_incomplete_final_record, ) { Ok(refresh) => refresh, Err(_) if !finalize => return Ok(Vec::new()), @@ -423,30 +399,15 @@ impl TranscriptTail { return Ok(Vec::new()); } if refresh == Refresh::Reset { - self.state = Self::new_state(self.source); - if self.source == ImportSource::Cursor { - // Keep Cursor's translator alive so its session root and turn - // ordinal continue across native transcript replacement. - // A fresh snapshot path makes it rewind only transcript input. - self.cursor_snapshot = None; - // Cursor can rewrite a transcript while retaining earlier - // turns. Keep those records out of the rotated snapshot so - // the live translator does not claim their prompts twice. - self.cursor_snapshot_records = self - .records - .values - .iter() - .zip(&previous_cursor_records) - .take_while(|(current, previous)| current == previous) - .count(); + self.session = self.importer.tail(); + if keeps_translator { + self.session + .rewritten(&previous_records, &self.records.values); } else { self.translator_reset = true; } - if self.source != ImportSource::Cursor { - self.cursor_snapshot_records = 0; - } } - let events = match self.envelopes_from_current_records(refresh == Refresh::Reset) { + let events = match self.session.envelopes(&self.path, &self.records) { Ok(events) => events, Err(_) if !finalize => { self.retry_envelopes = true; @@ -458,53 +419,12 @@ impl TranscriptTail { let len = std::fs::metadata(&self.path) .with_context(|| format!("read transcript metadata {}", self.path.display()))? .len(); - match &mut self.state { - TailState::Codex(state) => state.poll(events, len, finalize), - TailState::Cursor(state) => state.poll(events, len, finalize), - TailState::Claude(state) => state.poll(events, len, finalize), - TailState::Antigravity(state) => state.poll(events, len, finalize), - } + self.session.poll(events, len, finalize) } pub(crate) fn take_translator_reset(&mut self) -> bool { std::mem::take(&mut self.translator_reset) } - - fn envelopes_from_current_records( - &mut self, - reset_snapshot: bool, - ) -> anyhow::Result> { - if self.source != ImportSource::Cursor { - return envelopes_from_records(&self.path, self.source, &self.records); - } - if self.cursor_snapshot.is_none() { - self.cursor_snapshot = Some( - tempfile::Builder::new() - .prefix("bt-cursor-import-") - .tempdir()?, - ); - } - let snapshot = self - .cursor_snapshot - .as_ref() - .unwrap() - .path() - .join("transcript.jsonl"); - let mut file = std::fs::OpenOptions::new() - .create(true) - .write(true) - .append(!reset_snapshot) - .truncate(reset_snapshot) - .open(&snapshot)?; - for record in &self.records.values[self.cursor_snapshot_records..] { - serde_json::to_writer(&mut file, record)?; - file.write_all(b"\n")?; - } - file.flush()?; - let through = file.metadata()?.len(); - self.cursor_snapshot_records = self.records.values.len(); - cursor::envelopes_with_snapshot(&self.path, &snapshot, through, &self.records.values) - } } fn read_jsonl_records(path: &Path) -> anyhow::Result> { @@ -610,7 +530,7 @@ mod tests { std::fs::write(&transcript, "{}\n").unwrap(); assert_eq!( - resolve_transcript_in("session-123", ImportSource::Codex, &[root]).unwrap(), + codex::layout(vec![root]).find("session-123").unwrap(), transcript ); } @@ -626,7 +546,7 @@ mod tests { std::fs::write(&transcript, "{}\n").unwrap(); assert_eq!( - resolve_transcript_in("session-123", ImportSource::Claude, &[root]).unwrap(), + claude::layout(vec![root]).find("session-123").unwrap(), transcript ); } @@ -647,7 +567,9 @@ mod tests { std::fs::write(unrelated, "{}\n").unwrap(); assert_eq!( - resolve_transcript_in("conversation-123", ImportSource::Antigravity, &[root]).unwrap(), + antigravity::layout(vec![root]) + .find("conversation-123") + .unwrap(), transcript ); } @@ -668,16 +590,13 @@ mod tests { std::fs::write(&subagent, "{}\n").unwrap(); assert_eq!( - resolve_transcript_in( - "session-123", - ImportSource::Cursor, - std::slice::from_ref(&root) - ) - .unwrap(), + cursor::layout(vec![root.clone()]) + .find("session-123") + .unwrap(), transcript ); assert_eq!( - discover_transcripts_in(ImportSource::Cursor, &[root]).unwrap(), + cursor::layout(vec![root]).discover().unwrap(), vec![transcript] ); } @@ -891,12 +810,9 @@ mod tests { #[test] fn rejects_unsafe_session_ids() { - let error = resolve_transcript_in( - "../session", - ImportSource::Codex, - &[PathBuf::from("unused")], - ) - .unwrap_err(); + let error = codex::layout(vec![PathBuf::from("unused")]) + .find("../session") + .unwrap_err(); assert!(error.to_string().contains("invalid session id")); } @@ -908,18 +824,16 @@ mod tests { std::fs::create_dir_all(&first).unwrap(); std::fs::create_dir_all(&second).unwrap(); - let missing = resolve_transcript_in( - "missing", - ImportSource::Claude, - &[first.clone(), second.clone()], - ) - .unwrap_err(); + let missing = claude::layout(vec![first.clone(), second.clone()]) + .find("missing") + .unwrap_err(); assert!(missing.to_string().contains("no Claude Code transcript")); std::fs::write(first.join("duplicate.jsonl"), "{}\n").unwrap(); std::fs::write(second.join("duplicate.jsonl"), "{}\n").unwrap(); - let ambiguous = - resolve_transcript_in("duplicate", ImportSource::Claude, &[first, second]).unwrap_err(); + let ambiguous = claude::layout(vec![first, second]) + .find("duplicate") + .unwrap_err(); assert!(ambiguous .to_string() .contains("multiple Claude Code transcripts")); @@ -952,7 +866,7 @@ mod tests { std::fs::write(root.join("not-a-transcript.jsonl"), "{}\n").unwrap(); assert_eq!( - discover_transcripts_in(ImportSource::Codex, &[root]).unwrap(), + codex::layout(vec![root]).discover().unwrap(), vec![first, second] ); } @@ -978,7 +892,7 @@ mod tests { std::fs::write(project.join("notes.jsonl"), "{}\n").unwrap(); assert_eq!( - discover_transcripts_in(ImportSource::Claude, &[root]).unwrap(), + claude::layout(vec![root]).discover().unwrap(), vec![transcript] ); }