test(ai-session): e2e tests for capture, sync and resume; scan new dirs in the tree watcher (atuinsh/atuin#4368)

Cargo.lock

@@ -280,6 +280,7 @@ dependencies = [
  "atuin-octavo",
  "atuin-pty-proxy",
  "atuin-scripts",
+ "atuin-server",
  "atuin-vt100",
  "axoupdater",
  "clap",
@@ -314,6 +315,7 @@ dependencies = [
  "rpassword",
  "rstest",
  "runtime-format",
+ "rusqlite",
  "rust-embed",
  "rustix",
  "secrecy",

crates/atuin-common/src/fs/tree_watcher/poller.rs

@@ -64,12 +64,7 @@ where
                                     pending.push(debounced.event);
                                 }
                             }
-                            if !pending.is_empty() {
-                                let kinds = Self::resolve_kinds(&self.stat, &pending).await;
-                                for event in &pending {
-                                    self.observe_event(event, &kinds);
-                                }
-                            }
+                            self.apply(&pending).await;
                             if force_scan {
                                 self.walk.poll(&mut self.tracker).await;
                             }
@@ -82,6 +77,45 @@ where
         }
     }
 
+    /// Bring the tracker in line with a batch of filesystem events, and with what is in any
+    /// directory they created.
+    async fn apply(&mut self, events: &[notify::Event]) {
+        if events.is_empty() {
+            return;
+        }
+        let kinds = Self::resolve_kinds(&self.stat, events).await;
+        for event in events {
+            self.observe_event(event, &kinds);
+        }
+        self.walk.discover(Self::appeared_dirs(events, &kinds), &mut self.tracker).await;
+    }
+
+    /// The directories `events` say appeared (created, or moved in) that are still there. Of a
+    /// rename with both ends, only where it went: a case-only rename on a case-insensitive
+    /// filesystem leaves the old name resolving to the same directory, which would be walked
+    /// again under it.
+    fn appeared_dirs(
+        events: &[notify::Event],
+        kinds: &HashMap<Arc<Path>, PathFingerprint>,
+    ) -> Vec<Arc<Path>> {
+        events
+            .iter()
+            .flat_map(|event| match event.kind {
+                EventKind::Create(_)
+                | EventKind::Modify(ModifyKind::Name(
+                    RenameMode::To | RenameMode::Any | RenameMode::Other,
+                )) => event.paths.as_slice(),
+                EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => {
+                    event.paths.get(1..).unwrap_or_default()
+                }
+                _ => &[],
+            })
+            .filter_map(|path| kinds.get_key_value(path.as_path()))
+            .filter(|(_, fingerprint)| matches!(fingerprint, PathFingerprint::Dir))
+            .map(|(path, _)| Arc::clone(path))
+            .collect()
+    }
+
     /// Stat every path referenced by `events`, keeping only the ones that still exist.
     async fn resolve_kinds<'a>(
         stat: &StatPoller,
@@ -259,6 +293,46 @@ mod tests {
         assert_eq!(found.len(), yielded);
     }
 
+    /// A directory created with a file already in it -- written before the backend began watching
+    /// the directory, so no event of its own says so -- yields the file at once, not at the next
+    /// full scan. So does one nested deeper, and one in a directory moved in. A rename with both
+    /// ends walks only where it went, even while the old name still resolves (a case-only rename
+    /// on a case-insensitive filesystem), so no file is yielded under a name it no longer has.
+    #[rstest]
+    #[case::created(EventKind::Create(notify::event::CreateKind::Folder))]
+    #[case::moved_in(EventKind::Modify(ModifyKind::Name(RenameMode::To)))]
+    #[case::renamed(EventKind::Modify(ModifyKind::Name(RenameMode::Both)))]
+    #[tokio::test]
+    async fn a_new_directory_yields_the_files_already_in_it(#[case] kind: EventKind) {
+        let root = tempfile::tempdir().unwrap();
+        let dir = root.path().join("project");
+        std::fs::create_dir_all(dir.join("nested")).unwrap();
+        for file in [dir.join("a.jsonl"), dir.join("nested/b.jsonl")] {
+            std::fs::write(file, "{}\n").unwrap();
+        }
+        // Where a rename came from, still there: a stand-in for the old name of a case-only one.
+        let old = root.path().join("old");
+        std::fs::create_dir_all(&old).unwrap();
+        std::fs::write(old.join("c.jsonl"), "{}\n").unwrap();
+        let pool = BlockingPool::new(NonZeroUsize::MIN);
+        let (found, files) = flume::unbounded();
+        let mut poller = TreeWatcherPoller::new(
+            WalkPoller::new(pool.clone(), root.path().to_owned(), true),
+            StatPoller::new(pool, Duration::ZERO),
+            FileTracker::new(|_: &Path| true, found),
+        );
+
+        let paths = match kind {
+            EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => vec![old, dir.clone()],
+            _ => vec![dir.clone()],
+        };
+        poller.apply(&[event(kind, paths)]).await;
+
+        let yielded: HashSet<PathBuf> = files.drain().map(|f| f.path().to_owned()).collect();
+        let want = HashSet::from([dir.join("a.jsonl"), dir.join("nested/b.jsonl")]);
+        assert_eq!(yielded, want);
+    }
+
     #[rstest]
     fn observe_remove_event_forgets() {
         let (mut poller, found) = poller();

crates/atuin-common/src/fs/tree_watcher/walk.rs

@@ -52,6 +52,30 @@ impl WalkPoller {
         tracker.reconcile(walk);
     }
 
+    /// Walk the trees under `dirs` in the pool and observe everything in them, pruning nothing.
+    ///
+    /// For directories that just appeared: the backend starts watching one only once it has seen
+    /// it created, so whatever was written into it before then (a new file in a new directory,
+    /// the usual case) raises no event of its own, and would wait for the next full scan.
+    /// A directory inside another of `dirs` (`mkdir -p` reports each level) is walked once, with
+    /// the outermost.
+    pub async fn discover<F>(&self, mut dirs: Vec<Arc<Path>>, tracker: &mut FileTracker<F>)
+    where
+        F: Fn(&Path) -> bool,
+    {
+        if dirs.is_empty() || !self.recursive {
+            return;
+        }
+        // Sorted, a directory comes right after any other it is inside.
+        dirs.sort();
+        dirs.dedup_by(|inner, outer| inner.starts_with(&**outer));
+        let walks = self.pool.run(move || dirs.iter().map(|dir| Self::scan(dir, true)).collect());
+        let walks: Vec<WalkResult> = walks.await.unwrap_or_default();
+        for (path, fingerprint) in walks.into_iter().flat_map(|walk| walk.entries) {
+            tracker.observe_fingerprint(path, fingerprint);
+        }
+    }
+
     /// Walk `root` best-effort, returning every readable entry plus the set of directories that
     /// were fully read without error.
     fn scan(root: &Path, recursive: bool) -> WalkResult {

crates/atuin/Cargo.toml

@@ -166,10 +166,14 @@ tree-sitter-powershell = { workspace = true }
 windows-sys = { workspace = true }
 
 [dev-dependencies]
+# An in-process sync server for the AI session e2e tests (tests/e2e_ai_sessions.rs).
+atuin-server = { workspace = true }
 divan = { workspace = true }
 parking_lot = { workspace = true }
 proptest = { workspace = true }
 rstest = { workspace = true }
+# opencode keeps its sessions in SQLite: the e2e tests write one as it does.
+rusqlite = { workspace = true }
 tempfile = { workspace = true }
 toml_edit = { workspace = true, features = ["serde"] }
 

crates/atuin/tests/README.md

@@ -13,6 +13,16 @@ macOS, with Homebrew Bash on macOS.
 - `e2e_fresh_install`: CLI startup, keys, shell init, and doctor.
 - `e2e_pty`: shell hooks, search, selection, quoting, resize, and filters.
 - `e2e_daemon`: startup, concurrent writers, persistence, and restart.
+- `e2e_ai_sessions`: AI agent sessions captured from each agent's transcripts,
+  synced between two machines through an in-process sync server, and resumed:
+  restored, caught up, forked, switched, or continued in another agent.
+- `e2e_ai_resume_picker`: `atuin ai resume`'s picker and chooser on a PTY, and
+  through each shell's widget.
+
+The AI session tests write transcripts the way Claude Code, Codex, opencode and
+Pi do (`common/agents.rs`), and put stand-ins for those agents on `PATH` that
+print how they were run. Set `ATUIN_E2E_LOG` (an `ATUIN_LOG` filter) to have
+their daemons log more for a failure to show.
 
 ## Add a shell setup
 

crates/atuin/tests/common/agents.rs

@@ -0,0 +1,700 @@
+//! AI coding agents as the daemon sees them: their transcripts, written the way each agent writes
+//! its own, and stand-in executables that only say how they were run.
+//!
+//! Every turn written here is a prompt, some reasoning, one shell command with its output, and a
+//! reply, each holding a word no other turn has, so a test can tell exactly which of them a
+//! session, a search or a restored transcript carries.
+
+use std::fs;
+use std::io::Write as _;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicI64, Ordering};
+
+use serde_json::{Value, json};
+use time::OffsetDateTime;
+use time::format_description::well_known::Rfc3339;
+
+/// A coding agent atuin captures sessions from.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Agent {
+    Claude,
+    Codex,
+    Opencode,
+    Pi,
+}
+
+impl Agent {
+    pub const ALL: [Self; 4] = [Self::Claude, Self::Codex, Self::Opencode, Self::Pi];
+
+    /// The `harness` atuin's JSON output names it by.
+    pub fn harness(self) -> &'static str {
+        match self {
+            Self::Claude => "claude-code",
+            Self::Codex => "codex",
+            Self::Opencode => "opencode",
+            Self::Pi => "pi",
+        }
+    }
+
+    /// Its executable, and the name `atuin ai resume --in` takes.
+    pub fn program(self) -> &'static str {
+        match self {
+            Self::Claude => "claude",
+            Self::Codex => "codex",
+            Self::Opencode => "opencode",
+            Self::Pi => "pi",
+        }
+    }
+
+    /// The name it gives the shell tool every turn here calls.
+    pub fn tool(self) -> &'static str {
+        match self {
+            Self::Claude => "Bash",
+            Self::Codex => "exec_command",
+            Self::Opencode | Self::Pi => "bash",
+        }
+    }
+
+    /// The name atuin's status lines give it.
+    pub fn label(self) -> &'static str {
+        match self {
+            Self::Claude => "Claude Code",
+            Self::Codex => "Codex",
+            Self::Opencode => "opencode",
+            Self::Pi => "Pi",
+        }
+    }
+
+    /// A fresh session id of the agent's own shape.
+    pub fn new_id(self) -> String {
+        let uuid = atuin_common::utils::uuid_v7();
+        match self {
+            Self::Opencode => format!("ses_{}", uuid.as_simple()),
+            _ => uuid.as_hyphenated().to_string(),
+        }
+    }
+
+    /// Where, under `home`, the agent keeps its sessions: created up front, as a daemon only looks
+    /// for a missing one again after a backoff.
+    pub fn root(self, home: &Path) -> PathBuf {
+        match self {
+            Self::Claude => home.join(".claude/projects"),
+            Self::Codex => home.join(".codex/sessions"),
+            Self::Opencode => home.join(".local/share/opencode/opencode.db"),
+            Self::Pi => home.join(".pi/agent/sessions"),
+        }
+    }
+
+    /// Lay out the agent's data under `home` as the agent itself would before its first session.
+    pub fn install(self, home: &Path) {
+        match self {
+            Self::Opencode => opencode_db(home),
+            // Codex keeps a lock per open thread here; with the directory present, atuin asks the
+            // locks whether a session is open, rather than every `codex` process on the machine.
+            Self::Codex => {
+                fs::create_dir_all(self.root(home)).unwrap();
+                fs::create_dir_all(home.join(".codex/thread-writer-locks")).unwrap();
+            }
+            _ => fs::create_dir_all(self.root(home)).unwrap(),
+        }
+    }
+}
+
+/// A word found nowhere else, for searches and for telling turns apart.
+pub fn word() -> String {
+    format!("w{}", atuin_common::utils::uuid_v7().as_simple())
+}
+
+/// One exchange: a prompt, reasoning, a shell command and its output, and the reply.
+#[derive(Clone, Debug)]
+pub struct Turn {
+    pub prompt: String,
+    pub thinking: String,
+    pub command: String,
+    pub output: String,
+    pub reply: String,
+}
+
+impl Turn {
+    pub fn new() -> Self {
+        Self {
+            prompt: format!("please look into {}", word()),
+            thinking: format!("thinking about {}", word()),
+            command: format!("echo {}", word()),
+            output: format!("output {}", word()),
+            reply: format!("all done with {}", word()),
+        }
+    }
+
+    /// What this turn says in conversation, as a session's messages carry it: its prompt and its
+    /// reply.
+    pub fn said(&self) -> [&str; 2] {
+        [&self.prompt, &self.reply]
+    }
+
+    /// The turn as a transcript holds it, reasoning aside: its prompt, the command it ran, what
+    /// that printed, and its reply.
+    pub fn whole(&self) -> [&str; 4] {
+        [&self.prompt, &self.command, &self.output, &self.reply]
+    }
+}
+
+/// Everything `turns` hold, reasoning aside, in order (see [`Turn::whole`]).
+pub fn whole(turns: &[Turn]) -> Vec<String> {
+    turns.iter().flat_map(|t| t.whole().map(str::to_owned)).collect()
+}
+
+/// Everything `turns` say in conversation, in order.
+pub fn said(turns: &[Turn]) -> Vec<String> {
+    turns.iter().flat_map(|t| t.said().map(str::to_owned)).collect()
+}
+
+/// A session's transcript on one machine.
+#[derive(Clone, Debug)]
+pub struct Transcript {
+    pub agent: Agent,
+    pub id: String,
+    /// The transcript file, or for opencode its database.
+    pub path: PathBuf,
+}
+
+/// Each line a millisecond after the last, so every agent orders them as written.
+fn tick() -> OffsetDateTime {
+    static LAST: AtomicI64 = AtomicI64::new(0);
+    let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).unwrap();
+    let last =
+        LAST.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |last| Some(now.max(last + 1)));
+    let ms = now.max(last.unwrap() + 1);
+    OffsetDateTime::from_unix_timestamp_nanos(i128::from(ms) * 1_000_000).unwrap()
+}
+
+fn rfc3339(at: OffsetDateTime) -> String {
+    at.format(&Rfc3339).unwrap()
+}
+
+fn millis(at: OffsetDateTime) -> i64 {
+    i64::try_from(at.unix_timestamp_nanos() / 1_000_000).unwrap()
+}
+
+fn short() -> String {
+    atuin_common::utils::uuid_v7().as_simple().to_string()
+}
+
+impl Transcript {
+    /// Write a new session of `agent`, working in `cwd`, under `home`.
+    pub fn create(home: &Path, agent: Agent, cwd: &Path, turns: &[Turn]) -> Self {
+        let id = agent.new_id();
+        let started = tick();
+        let path = match agent {
+            Agent::Claude => agent.root(home).join(claude_project(cwd)).join(format!("{id}.jsonl")),
+            Agent::Codex => {
+                let day = started.format(time::macros::format_description!("[year]/[month]/[day]"));
+                let stamp = started.format(time::macros::format_description!(
+                    "[year]-[month]-[day]T[hour]-[minute]-[second]"
+                ));
+                agent
+                    .root(home)
+                    .join(day.unwrap())
+                    .join(format!("rollout-{}-{id}.jsonl", stamp.unwrap()))
+            }
+            Agent::Pi => {
+                let dir = format!(
+                    "--{}--",
+                    cwd.display().to_string().trim_matches('/').replace('/', "-")
+                );
+                let stamp = rfc3339(started).replace([':', '.'], "-");
+                agent.root(home).join(dir).join(format!("{stamp}_{id}.jsonl"))
+            }
+            Agent::Opencode => agent.root(home),
+        };
+        let transcript = Self { agent, id, path };
+        match agent {
+            Agent::Claude | Agent::Pi | Agent::Codex => {
+                fs::create_dir_all(transcript.path.parent().unwrap()).unwrap();
+                let header = match agent {
+                    Agent::Codex => vec![json!({
+                        "timestamp": rfc3339(started),
+                        "type": "session_meta",
+                        "payload": {
+                            "id": transcript.id,
+                            "timestamp": rfc3339(started),
+                            "cwd": cwd,
+                            "originator": "codex_cli_rs",
+                            "cli_version": "0.50.0",
+                            "instructions": null,
+                            "git": {"branch": "main"},
+                        },
+                    })],
+                    Agent::Pi => vec![json!({
+                        "type": "session",
+                        "version": 3,
+                        "id": transcript.id,
+                        "timestamp": rfc3339(started),
+                        "cwd": cwd,
+                    })],
+                    _ => Vec::new(),
+                };
+                write_lines(&transcript.path, &header);
+            }
+            Agent::Opencode => {
+                let title = format!("opencode session {}", word());
+                opencode_events(home, &[(
+                    "session.created.1",
+                    json!({
+                        "sessionID": transcript.id,
+                        "info": {
+                            "id": transcript.id,
+                            "title": title,
+                            "directory": cwd,
+                            "time": {"created": millis(started), "updated": millis(started)},
+                        },
+                    }),
+                )]);
+            }
+        }
+        transcript.append(cwd, turns);
+        transcript
+    }
+
+    /// A Claude Code subagent `parent` spawned, working in `cwd`: its own transcript beside the
+    /// parent's, whose lines name the parent's session.
+    pub fn claude_subagent(parent: &Self, cwd: &Path, turns: &[Turn]) -> Self {
+        assert_eq!(parent.agent, Agent::Claude);
+        let id = format!("agent-{}", &short()[16..]);
+        let dir = parent.path.with_extension("").join("subagents");
+        fs::create_dir_all(&dir).unwrap();
+        let subagent = Self {
+            agent: Agent::Claude,
+            path: dir.join(format!("{id}.jsonl")),
+            id,
+        };
+        subagent.append(cwd, turns);
+        subagent
+    }
+
+    /// Where this agent keeps the transcript of session `id` under `home`, when it has one.
+    pub fn find(home: &Path, agent: Agent, id: &str) -> Option<Self> {
+        assert_ne!(agent, Agent::Opencode, "opencode keeps every session in one database");
+        let mut dirs = vec![agent.root(home)];
+        while let Some(dir) = dirs.pop() {
+            // A directory gone or unreadable holds nothing to find: the rest may.
+            let Ok(entries) = fs::read_dir(dir) else {
+                continue;
+            };
+            for entry in entries.flatten() {
+                let path = entry.path();
+                if path.is_dir() {
+                    dirs.push(path);
+                } else if path.file_stem().unwrap().to_string_lossy().ends_with(id) {
+                    let id = id.to_owned();
+                    return Some(Self { agent, id, path });
+                }
+            }
+        }
+        None
+    }
+
+    /// The transcript as it stands.
+    pub fn read(&self) -> String {
+        fs::read_to_string(&self.path).unwrap()
+    }
+
+    /// Carry the session on with `turns`, as the agent would: after whatever its transcript ends
+    /// with, whoever wrote it.
+    pub fn append(&self, cwd: &Path, turns: &[Turn]) {
+        match self.agent {
+            Agent::Claude => self.append_claude(cwd, turns),
+            Agent::Codex => self.append_codex(turns),
+            Agent::Pi => self.append_pi(turns),
+            Agent::Opencode => self.append_opencode(cwd, turns),
+        }
+    }
+
+    /// The last line of the transcript that has `key`, as the agent finds where to go on from.
+    fn last(&self, key: &str) -> Option<String> {
+        fs::read_to_string(&self.path)
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| serde_json::from_str::<Value>(line).ok())
+            .filter(|line| line["type"] != "session")
+            .filter_map(|line| line[key].as_str().map(str::to_owned))
+            .next_back()
+    }
+
+    fn append_claude(&self, cwd: &Path, turns: &[Turn]) {
+        let mut parent = self.last("uuid");
+        // A subagent's lines name the session it works for.
+        let session = match self.id.strip_prefix("agent-") {
+            Some(_) => self
+                .path
+                .ancestors()
+                .nth(2)
+                .unwrap()
+                .file_name()
+                .unwrap()
+                .to_string_lossy()
+                .into_owned(),
+            None => self.id.clone(),
+        };
+        let mut lines = Vec::new();
+        let mut line = |kind: &str, message: Value| {
+            let uuid = uuid::Uuid::new_v4().to_string();
+            lines.push(json!({
+                "parentUuid": parent,
+                "isSidechain": session != self.id,
+                "userType": "external",
+                "cwd": cwd,
+                "sessionId": session,
+                "version": "2.1.0",
+                "gitBranch": "main",
+                "type": kind,
+                "message": message,
+                "uuid": uuid,
+                "timestamp": rfc3339(tick()),
+            }));
+            parent = Some(uuid);
+        };
+        for turn in turns {
+            let call = format!("toolu_{}", short());
+            let usage = json!({
+                "input_tokens": 10,
+                "output_tokens": 5,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 0,
+            });
+            line("user", json!({"role": "user", "content": turn.prompt}));
+            line(
+                "assistant",
+                json!({
+                    "id": format!("msg_{}", short()),
+                    "type": "message",
+                    "role": "assistant",
+                    "model": "claude-opus-5-5",
+                    "content": [
+                        {"type": "thinking", "thinking": turn.thinking, "signature": "sig"},
+                        {"type": "tool_use", "id": call, "name": self.agent.tool(), "input": {"command": turn.command}},
+                    ],
+                    "stop_reason": "tool_use",
+                    "usage": usage,
+                }),
+            );
+            line(
+                "user",
+                json!({
+                    "role": "user",
+                    "content": [{"type": "tool_result", "tool_use_id": call, "content": turn.output, "is_error": false}],
+                }),
+            );
+            line(
+                "assistant",
+                json!({
+                    "id": format!("msg_{}", short()),
+                    "type": "message",
+                    "role": "assistant",
+                    "model": "claude-opus-5-5",
+                    "content": [{"type": "text", "text": turn.reply}],
+                    "stop_reason": "end_turn",
+                    "usage": usage,
+                }),
+            );
+        }
+        write_lines(&self.path, &lines);
+    }
+
+    fn append_codex(&self, turns: &[Turn]) {
+        let mut lines = Vec::new();
+        let mut item = |payload: Value| {
+            lines.push(
+                json!({"timestamp": rfc3339(tick()), "type": "response_item", "payload": payload}),
+            );
+        };
+        for turn in turns {
+            let call = format!("call_{}", short());
+            item(
+                json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": turn.prompt}]}),
+            );
+            item(json!({
+                "type": "reasoning",
+                "summary": [{"type": "summary_text", "text": turn.thinking}],
+                "encrypted_content": null,
+            }));
+            item(json!({
+                "type": "function_call",
+                "name": self.agent.tool(),
+                "arguments": json!({"cmd": turn.command}).to_string(),
+                "call_id": call,
+            }));
+            item(json!({
+                "type": "function_call_output",
+                "call_id": call,
+                "output": json!({"output": turn.output, "metadata": {"exit_code": 0}}).to_string(),
+            }));
+            item(
+                json!({"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": turn.reply}]}),
+            );
+        }
+        write_lines(&self.path, &lines);
+    }
+
+    fn append_pi(&self, turns: &[Turn]) {
+        let mut parent = self.last("id");
+        let mut lines = Vec::new();
+        let mut entry = |message: Value| {
+            let id = short()[24..].to_owned();
+            let at = tick();
+            let mut message = message;
+            message["timestamp"] = json!(millis(at));
+            lines.push(json!({
+                "type": "message",
+                "id": id,
+                "parentId": parent,
+                "timestamp": rfc3339(at),
+                "message": message,
+            }));
+            parent = Some(id);
+        };
+        let usage = json!({
+            "input": 10, "output": 5, "cacheRead": 0, "cacheWrite": 0, "totalTokens": 15,
+            "cost": {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0, "total": 0},
+        });
+        for turn in turns {
+            let call = format!("toolu_{}", short());
+            entry(json!({"role": "user", "content": [{"type": "text", "text": turn.prompt}]}));
+            entry(json!({
+                "role": "assistant",
+                "content": [
+                    {"type": "thinking", "thinking": turn.thinking},
+                    {"type": "toolCall", "id": call, "name": self.agent.tool(), "arguments": {"command": turn.command}},
+                ],
+                "api": "anthropic-messages",
+                "provider": "anthropic",
+                "model": "claude-opus-5-5",
+                "usage": usage,
+                "stopReason": "toolUse",
+            }));
+            entry(json!({
+                "role": "toolResult",
+                "toolCallId": call,
+                "toolName": self.agent.tool(),
+                "content": [{"type": "text", "text": turn.output}],
+                "isError": false,
+            }));
+            entry(json!({
+                "role": "assistant",
+                "content": [{"type": "text", "text": turn.reply}],
+                "api": "anthropic-messages",
+                "provider": "anthropic",
+                "model": "claude-opus-5-5",
+                "usage": usage,
+                "stopReason": "stop",
+            }));
+        }
+        write_lines(&self.path, &lines);
+    }
+
+    fn append_opencode(&self, cwd: &Path, turns: &[Turn]) {
+        let home = self.path.ancestors().nth(4).unwrap();
+        let mut events = Vec::new();
+        for turn in turns {
+            let (asked, answered) = (tick(), tick());
+            let (user, assistant) = (format!("msg_{}", short()), format!("msg_{}", short()));
+            // A part's own clock, else the time it was written, orders it.
+            let part = |message: &str, part: Value, at: OffsetDateTime| {
+                let mut part = part;
+                part["id"] = json!(format!("prt_{}", short()));
+                part["messageID"] = json!(message);
+                part["sessionID"] = json!(self.id);
+                (
+                    "message.part.updated.1",
+                    json!({"sessionID": self.id, "part": part, "time": millis(at)}),
+                )
+            };
+            let done = |at: OffsetDateTime| json!({"start": millis(at), "end": millis(tick())});
+            events.push(("message.updated.1", json!({
+                "sessionID": self.id,
+                "info": {"id": user, "sessionID": self.id, "role": "user", "time": {"created": millis(asked)}},
+            })));
+            events.push(part(&user, json!({"type": "text", "text": turn.prompt}), asked));
+            events.push((
+                "message.updated.1",
+                json!({
+                    "sessionID": self.id,
+                    "info": {
+                        "id": assistant,
+                        "sessionID": self.id,
+                        "role": "assistant",
+                        "parentID": user,
+                        "modelID": "claude-opus-5-5",
+                        "providerID": "anthropic",
+                        "path": {"cwd": cwd, "root": cwd},
+                        "time": {"created": millis(answered)},
+                    },
+                }),
+            ));
+            events.push(part(
+                &assistant,
+                json!({"type": "reasoning", "text": turn.thinking, "time": done(answered)}),
+                answered,
+            ));
+            events.push(part(
+                &assistant,
+                json!({
+                    "type": "tool",
+                    "callID": format!("call_{}", short()),
+                    "tool": self.agent.tool(),
+                    "state": {
+                        "status": "completed",
+                        "input": {"command": turn.command},
+                        "output": turn.output,
+                        "time": done(answered),
+                    },
+                }),
+                answered,
+            ));
+            events.push(part(
+                &assistant,
+                json!({"type": "text", "text": turn.reply, "time": done(answered)}),
+                answered,
+            ));
+        }
+        opencode_events(home, &events);
+    }
+}
+
+fn write_lines(path: &Path, lines: &[Value]) {
+    let mut file = fs::OpenOptions::new().create(true).append(true).open(path).unwrap();
+    let mut text = String::new();
+    for line in lines {
+        text.push_str(&line.to_string());
+        text.push('\n');
+    }
+    // One write, so the daemon never reads half a turn.
+    file.write_all(text.as_bytes()).unwrap();
+}
+
+/// Claude Code's directory for the project at `cwd`: every character but a letter or digit a dash.
+fn claude_project(cwd: &Path) -> String {
+    cwd.display()
+        .to_string()
+        .chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() {
+                c
+            } else {
+                '-'
+            }
+        })
+        .collect()
+}
+
+/// opencode's database: the event log it records sessions in, and the tables it projects them
+/// into.
+fn opencode_db(home: &Path) {
+    let path = Agent::Opencode.root(home);
+    fs::create_dir_all(path.parent().unwrap()).unwrap();
+    let db = rusqlite::Connection::open(&path).unwrap();
+    db.execute_batch(
+        "PRAGMA journal_mode = WAL;
+         CREATE TABLE event (id TEXT PRIMARY KEY, aggregate_id TEXT NOT NULL, seq INTEGER NOT \
+         NULL, type TEXT NOT NULL, data TEXT NOT NULL);
+         CREATE UNIQUE INDEX event_aggregate_seq_idx ON event (aggregate_id, seq);
+         CREATE TABLE session (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT NOT NULL, title \
+         TEXT NOT NULL, revert TEXT, time_created INTEGER NOT NULL, time_updated INTEGER NOT \
+         NULL);
+         CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT NOT NULL, time_created INTEGER \
+         NOT NULL, time_updated INTEGER NOT NULL, data TEXT NOT NULL);
+         CREATE TABLE part (id TEXT PRIMARY KEY, message_id TEXT NOT NULL, session_id TEXT NOT \
+         NULL, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL, data TEXT NOT NULL);",
+    )
+    .unwrap();
+}
+
+/// Append `events` to opencode's event log, each continuing its session's sequence, and project
+/// them, in one transaction as opencode commits them.
+fn opencode_events(home: &Path, events: &[(&str, Value)]) {
+    let mut db = rusqlite::Connection::open(Agent::Opencode.root(home)).unwrap();
+    db.busy_timeout(std::time::Duration::from_secs(10)).unwrap();
+    let tx = db.transaction().unwrap();
+    for (kind, data) in events {
+        let session = data["sessionID"].as_str().unwrap();
+        tx.execute(
+            "INSERT INTO event (id, aggregate_id, seq, type, data) VALUES (?1, ?2, (SELECT \
+             coalesce(max(seq), -1) + 1 FROM event WHERE aggregate_id = ?2), ?3, ?4)",
+            rusqlite::params![format!("evt_{}", short()), session, kind, data.to_string()],
+        )
+        .unwrap();
+        // The projections keep a row's info without the ids their columns hold.
+        let without = |value: &Value, keys: &[&str]| {
+            let mut value = value.clone();
+            for key in keys {
+                value.as_object_mut().unwrap().remove(*key);
+            }
+            value.to_string()
+        };
+        match *kind {
+            "session.created.1" => {
+                let info = &data["info"];
+                tx.execute(
+                    "INSERT INTO session (id, directory, title, time_created, time_updated) \
+                     VALUES (?1, ?2, ?3, ?4, ?4)",
+                    rusqlite::params![
+                        session,
+                        info["directory"].as_str(),
+                        info["title"].as_str(),
+                        info["time"]["created"].as_i64()
+                    ],
+                )
+                .unwrap();
+            }
+            "message.updated.1" => {
+                let info = &data["info"];
+                let created = info["time"]["created"].as_i64();
+                tx.execute(
+                    "INSERT INTO message (id, session_id, time_created, time_updated, data) \
+                     VALUES (?1, ?2, ?3, ?3, ?4)",
+                    rusqlite::params![
+                        info["id"].as_str(),
+                        session,
+                        created,
+                        without(info, &["id", "sessionID"])
+                    ],
+                )
+                .unwrap();
+            }
+            _ => {
+                let part = &data["part"];
+                tx.execute(
+                    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, \
+                     data) VALUES (?1, ?2, ?3, ?4, ?4, ?5)",
+                    rusqlite::params![
+                        part["id"].as_str(),
+                        part["messageID"].as_str(),
+                        session,
+                        data["time"].as_i64(),
+                        without(part, &["id", "messageID", "sessionID"])
+                    ],
+                )
+                .unwrap();
+            }
+        }
+    }
+    tx.commit().unwrap();
+}
+
+/// Stand-ins for the agents' executables, on `PATH` ahead of anything installed: each prints how
+/// it was run and where, and exits.
+pub fn install_fakes(bin: &Path, agents: &[Agent]) {
+    use std::os::unix::fs::PermissionsExt as _;
+    fs::create_dir_all(bin).unwrap();
+    for agent in agents {
+        let path = bin.join(agent.program());
+        fs::write(&path, format!("#!/bin/sh\necho \"AGENT-RAN {} $* IN $PWD\"\n", agent.program()))
+            .unwrap();
+        fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).unwrap();
+    }
+}
+
+/// The line a fake agent prints when run with `args` in `cwd`.
+pub fn ran(agent: Agent, args: &str, cwd: &Path) -> String {
+    format!("AGENT-RAN {} {args} IN {}", agent.program(), cwd.display())
+}

crates/atuin/tests/common/ai.rs

@@ -0,0 +1,447 @@
+//! Machines running atuin with AI session capture on, and a sync server between them.
+//!
+//! A [`Machine`] is a temporary home with its own daemon, data and host id, the agents of
+//! [`agents`](crate::agents) laid out in it and stand-ins for their executables on its `PATH`.
+//! Two machines given the same [`SyncServer`] share one account, so what one captures the other's
+//! daemon syncs down, as on two computers.
+
+use std::net::SocketAddr;
+use std::os::unix::process::CommandExt as _;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+use std::time::{Duration, Instant};
+
+use serde_json::Value;
+use tempfile::TempDir;
+
+use crate::agents::{Agent, install_fakes};
+use crate::common::{FreshEnv, Process, TIMEOUT, output};
+
+/// An atuin sync server on a port of its own, run on a runtime of its own so the blocking tests
+/// that use it never starve it.
+pub struct SyncServer {
+    pub address: String,
+    _dir: TempDir,
+    runtime: Option<tokio::runtime::Runtime>,
+}
+
+impl SyncServer {
+    pub fn start() -> Self {
+        let dir = tempfile::tempdir().unwrap();
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()
+            .unwrap();
+        let settings = atuin_server::Settings {
+            host: "127.0.0.1".to_owned(),
+            port: 0,
+            path: String::new(),
+            open_registration: true,
+            max_record_size: atuin_common::units::ByteSize::b(1024 * 1024 * 1024),
+            register_webhook_url: None,
+            register_webhook_username: String::new(),
+            db_settings: atuin_server::db::DbSettings {
+                db_uri: format!("sqlite://{}", dir.path().join("server.db").display())
+                    .parse()
+                    .unwrap(),
+            },
+            metrics: atuin_server::settings::Metrics::default(),
+            fake_version: None,
+        };
+        let listener = runtime.block_on(tokio::net::TcpListener::bind("127.0.0.1:0")).unwrap();
+        let addr: SocketAddr = listener.local_addr().unwrap();
+        runtime.spawn(async move {
+            atuin_server::launch_with_tcp_listener(settings, listener, std::future::pending())
+                .await
+                .expect("sync server failed");
+        });
+        let server = Self {
+            address: format!("http://{addr}"),
+            _dir: dir,
+            runtime: Some(runtime),
+        };
+        wait("the sync server", || std::net::TcpStream::connect(addr).is_ok());
+        server
+    }
+}
+
+impl Drop for SyncServer {
+    fn drop(&mut self) {
+        if let Some(runtime) = self.runtime.take() {
+            runtime.shutdown_background();
+        }
+    }
+}
+
+/// Poll `check` until it gives a value, and return that; after [`TIMEOUT`], fail with what it
+/// last said was missing.
+pub fn wait_for<T>(what: &str, mut check: impl FnMut() -> Result<T, String>) -> T {
+    let deadline = Instant::now() + TIMEOUT;
+    loop {
+        match check() {
+            Ok(value) => return value,
+            Err(why) => assert!(Instant::now() < deadline, "timed out waiting for {what}: {why}"),
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+}
+
+/// Poll `ready` until it holds, failing after [`TIMEOUT`].
+pub fn wait(what: &str, mut ready: impl FnMut() -> bool) {
+    wait_for(what, || {
+        if ready() {
+            Ok(())
+        } else {
+            Err(String::new())
+        }
+    });
+}
+
+/// One computer: a home, a daemon capturing sessions, and the agents' executables.
+pub struct Machine {
+    pub env: FreshEnv,
+    /// The agents' stand-ins. Not in the temporary home: that may be on a filesystem that runs
+    /// nothing.
+    bin: TempDir,
+    daemon: Option<Process>,
+}
+
+/// Cargo's scratch directory for integration tests, made if it isn't there: a test archive run in
+/// a fresh checkout has none.
+fn target_tmp() -> &'static Path {
+    let dir = Path::new(env!("CARGO_TARGET_TMPDIR"));
+    std::fs::create_dir_all(dir).unwrap();
+    dir
+}
+
+/// Where a test keeps what both machines see, like a checkout at the same path on each: a
+/// session's working directory has to exist on the machine resuming it for the agent to start
+/// there.
+pub fn shared_dir() -> TempDir {
+    tempfile::Builder::new().prefix("project-").tempdir_in(target_tmp()).unwrap()
+}
+
+impl Machine {
+    /// A machine on its own, syncing nowhere.
+    pub fn new() -> Self {
+        Self::with_config("", "capture_sessions = true\n")
+    }
+
+    /// A machine syncing nowhere with the settings `top` (top-level keys, then any tables) and
+    /// the `[ai]` settings `ai`.
+    pub fn with_config(top: &str, ai: &str) -> Self {
+        Self::configured(None, top, ai)
+    }
+
+    /// A machine with the settings `top` and `[ai]` settings `ai`, syncing through the server at
+    /// `sync`, if any, as often as its daemon can.
+    fn configured(sync: Option<&str>, top: &str, ai: &str) -> Self {
+        let env = FreshEnv::new();
+        let bin = tempfile::Builder::new().prefix("bin-").tempdir_in(target_tmp()).unwrap();
+        install_fakes(bin.path(), &Agent::ALL);
+        for agent in Agent::ALL {
+            agent.install(env.home());
+        }
+        let (auto_sync, address, frequency) = match sync {
+            Some(address) => {
+                (true, format!("sync_address = \"{address}\"\n"), "sync_frequency = 1\n")
+            }
+            None => (false, String::new(), ""),
+        };
+        env.write_config(&format!(
+            "local_timeout = 15\nauto_sync = {auto_sync}\n{address}{top}[daemon]\nenabled = \
+             true\nautostart = false\n{frequency}[ai]\n{ai}"
+        ));
+        // Finish migrations before the daemon and the commands race for the databases.
+        env.run(&["store", "status"]);
+        Self {
+            env,
+            bin,
+            daemon: None,
+        }
+    }
+
+    /// Two machines of one account on `server`, their daemons running.
+    pub fn pair(server: &SyncServer) -> (Self, Self) {
+        let one = Self::syncing(server, None);
+        let key = one.run(&["key", "--base64"]).trim().to_owned();
+        let two = Self::syncing(server, Some((&one, &key)));
+        (one, two)
+    }
+
+    /// A machine syncing through `server`: registering a new account, or logging into `login`'s
+    /// with its key. Logged in before its daemon starts, which keeps the key it starts with.
+    fn syncing(server: &SyncServer, login: Option<(&Self, &str)>) -> Self {
+        let mut machine = Self::configured(Some(&server.address), "", "capture_sessions = true\n");
+        match login {
+            None => {
+                let user = machine.user();
+                machine.run(&[
+                    "register",
+                    "-u",
+                    &user,
+                    "-e",
+                    &format!("{user}@example.com"),
+                    "-p",
+                    &user,
+                ]);
+            }
+            Some((other, key)) => {
+                let user = other.user();
+                machine.run(&["login", "-u", &user, "-p", &user, "-k", key]);
+            }
+        }
+        machine.start_daemon();
+        machine
+    }
+
+    /// The account name this machine registers, and the password: unique to the home.
+    fn user(&self) -> String {
+        let name = self.env.home().file_name().unwrap().to_string_lossy();
+        name.replace(|c: char| !c.is_ascii_alphanumeric(), "")
+    }
+
+    pub fn home(&self) -> &Path {
+        self.env.home()
+    }
+
+    /// This machine's `PATH`: the agents' stand-ins first.
+    pub fn path_var(&self) -> String {
+        format!("{}:{}", self.bin.path().display(), self.env.path_var())
+    }
+
+    /// `atuin args` on this machine, finding the agents' stand-ins on its `PATH`.
+    pub fn atuin(&self, args: &[&str]) -> Command {
+        let mut command = self.env.atuin(args);
+        command.env("PATH", self.path_var());
+        // The harness turns automatic sync off for every command; the config decides here.
+        command.env_remove("ATUIN_AUTO_SYNC");
+        command
+    }
+
+    pub fn run(&self, args: &[&str]) -> String {
+        output(self.atuin(args))
+    }
+
+    pub fn start_daemon(&mut self) {
+        assert!(self.daemon.is_none(), "the daemon is already running");
+        let mut daemon = self.atuin(&["daemon", "start", "--show-logs"]);
+        // `ATUIN_E2E_LOG` sets what the daemons log, as `ATUIN_LOG` does, for a failure to show.
+        if let Some(filter) = std::env::var_os("ATUIN_E2E_LOG") {
+            daemon.env("ATUIN_LOG", filter);
+        }
+        let daemon = Process::spawn(daemon);
+        let socket = self.env.socket();
+        wait("the daemon", || socket.exists());
+        self.daemon = Some(daemon);
+        // `ai session` commands wait for its first replay of the store, so once one answers, the
+        // daemon is capturing.
+        self.sessions();
+    }
+
+    pub fn stop_daemon(&mut self) {
+        let daemon = self.daemon.take().expect("the daemon isn't running");
+        self.run(&["daemon", "stop"]);
+        let socket = self.env.socket();
+        wait("the daemon to stop", || !socket.exists());
+        drop(daemon);
+    }
+
+    /// The daemon's log so far, for a failure message.
+    pub fn logs(&self) -> String {
+        self.daemon.as_ref().map(Process::logs).unwrap_or_default()
+    }
+
+    /// Every session the daemon has, as `atuin ai session list` gives them.
+    pub fn sessions(&self) -> Vec<Value> {
+        let list = self.run(&["ai", "session", "list", "--style", "json"]);
+        serde_json::from_str::<Value>(&list).unwrap().as_array().unwrap().clone()
+    }
+
+    /// The session `id` (the agent's), as `atuin ai session list` gives it.
+    pub fn session(&self, id: &str) -> Option<Value> {
+        self.sessions().into_iter().find(|s| s["session_id"] == id)
+    }
+
+    /// The session `id` and its messages, as `atuin ai session show` gives them.
+    pub fn show(&self, id: &str) -> Value {
+        let show = self.run(&["ai", "session", "show", id, "--style", "json"]);
+        serde_json::from_str(&show).unwrap()
+    }
+
+    /// The text of each message of session `id` that says something in conversation, in order.
+    pub fn said(&self, id: &str) -> Vec<String> {
+        said_in(&self.show(id))
+    }
+
+    /// Wait for `check` to give a value, as [`wait_for`], saying on failure what the daemon
+    /// logged as well.
+    fn wait_for<T>(&self, what: &str, mut check: impl FnMut() -> Result<T, String>) -> T {
+        wait_for(what, || check().map_err(|why| format!("{why}\ndaemon log:\n{}", self.logs())))
+    }
+
+    /// What session `id` says in conversation, or why there's nothing to say.
+    fn current(&self, id: &str) -> Result<Vec<String>, String> {
+        match self.session(id) {
+            Some(_) => Ok(self.said(id)),
+            None => Err(format!("there's no session {id}")),
+        }
+    }
+
+    /// Wait for session `id` to hold exactly `want` in conversation, failing with what it holds.
+    pub fn wait_until_said(&self, id: &str, want: &[String]) {
+        self.wait_for(&format!("session {id} to say {want:#?}"), || {
+            let said = self.current(id)?;
+            if said == want {
+                Ok(())
+            } else {
+                Err(format!("it says {said:#?}"))
+            }
+        });
+    }
+
+    /// Wait for session `id` to hold `want` in conversation, in order, among whatever else.
+    pub fn wait_until_holds(&self, id: &str, want: &[String]) {
+        self.wait_for(&format!("session {id} to hold {want:#?}"), || {
+            let said = self.current(id)?.join("\n");
+            if holds_in_order(&said, want) {
+                Ok(())
+            } else {
+                Err(format!("it says {said:?}"))
+            }
+        });
+    }
+
+    /// Wait for a session of `agent` whose parent is session `parent` to appear, and return it.
+    /// A new session is listed as soon as its first row is stored, which can be before the row
+    /// that names its parent.
+    pub fn wait_for_child(&self, agent: Agent, parent: &str) -> Value {
+        self.wait_for(&format!("a {agent:?} session under {parent}"), || {
+            self.sessions()
+                .into_iter()
+                .find(|s| s["harness"] == agent.harness() && s["parent"]["session_id"] == parent)
+                .ok_or_else(|| "none yet".to_owned())
+        })
+    }
+
+    /// `atuin ai resume args`, run from `cwd` as a shell there would, without a terminal: never
+    /// the picker, which a test can't answer.
+    pub fn resume(&self, args: &[&str], cwd: &Path) -> Output {
+        let mut command = self.atuin(&[&["ai", "resume"], args].concat());
+        command.current_dir(cwd).env("PWD", cwd);
+        // SAFETY: `setsid` is async-signal-safe and touches no memory of this process.
+        unsafe {
+            command.pre_exec(|| rustix::process::setsid().map(drop).map_err(Into::into));
+        }
+        Process::spawn(command).wait()
+    }
+
+    /// This machine's host id.
+    pub fn host_id(&self) -> String {
+        let status = self.run(&["store", "status"]);
+        status
+            .lines()
+            .find_map(|l| l.strip_prefix("host: ")?.strip_suffix(" <- CURRENT HOST"))
+            .unwrap_or_else(|| panic!("no current host in {status}"))
+            .to_owned()
+    }
+
+    /// The way atuin names this machine to another: `@` and the end of its host id.
+    pub fn short_host(&self) -> String {
+        let id = self.host_id();
+        format!("@{}", &id[id.len() - 8..])
+    }
+
+    /// Atuin's data directory here.
+    pub fn data_dir(&self) -> PathBuf {
+        self.env.data_dir()
+    }
+}
+
+impl Drop for Machine {
+    fn drop(&mut self) {
+        if self.daemon.is_some() {
+            let _ = Process::spawn(self.env.atuin(&["daemon", "stop"])).try_wait();
+        }
+    }
+}
+
+/// The text of each message of `show` (`atuin ai session show --style json`) that says something
+/// in conversation: a prompt, or a reply.
+pub fn said_in(show: &Value) -> Vec<String> {
+    show["messages"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .filter(|m| m["role"] == "user" || m["role"] == "assistant")
+        .flat_map(|m| m["content"].as_array().unwrap().iter())
+        .filter(|c| c["type"] == "text")
+        .map(|c| c["text"].as_str().unwrap().to_owned())
+        .collect()
+}
+
+/// One tool call of a session, as `atuin ai session show --style json` gives it.
+#[derive(Debug, PartialEq, Eq)]
+pub struct Call {
+    pub name: String,
+    /// What it was given, as JSON text; empty when capture kept none.
+    pub input: String,
+    /// What it returned; empty when capture kept none.
+    pub output: String,
+}
+
+/// Every tool call of `show`, in order, each with what it returned.
+pub fn calls_in(show: &Value) -> Vec<Call> {
+    let content: Vec<&Value> = show["messages"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .flat_map(|m| m["content"].as_array().unwrap())
+        .collect();
+    let text = |value: &Value| value.as_str().unwrap_or_default().to_owned();
+    content
+        .iter()
+        .filter(|c| c["type"] == "tool_call")
+        .map(|call| Call {
+            name: text(&call["name"]),
+            input: text(&call["input"]),
+            output: content
+                .iter()
+                .find(|c| c["type"] == "tool_result" && c["tool_use_id"] == call["id"])
+                .map(|result| text(&result["content"]))
+                .unwrap_or_default(),
+        })
+        .collect()
+}
+
+/// Whether `haystack` holds each of `needles`, in order.
+pub fn holds_in_order(haystack: &str, needles: &[String]) -> bool {
+    let mut rest = haystack;
+    needles.iter().all(|needle| match rest.find(needle.as_str()) {
+        Some(at) => {
+            rest = &rest[at + needle.len()..];
+            true
+        }
+        None => false,
+    })
+}
+
+/// `output`'s stdout and stderr, asserting it succeeded.
+pub fn succeeded(output: &Output) -> (String, String) {
+    let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
+    let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
+    assert!(output.status.success(), "failed:\nstdout: {stdout}\nstderr: {stderr}");
+    (stdout, stderr)
+}
+
+/// `output`'s stderr, asserting it failed.
+pub fn failed(output: &Output) -> String {
+    let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
+    assert!(
+        !output.status.success(),
+        "succeeded:\nstdout: {}\nstderr: {stderr}",
+        String::from_utf8_lossy(&output.stdout)
+    );
+    stderr
+}

crates/atuin/tests/common/shell.rs

@@ -36,6 +36,32 @@ pub struct Shell {
 
 impl Shell {
     pub fn start(path: &Path, settings: Option<&str>) -> Option<Self> {
+        let setup = Setup::find(path)?;
+        let env = FreshEnv::new();
+        if let Some(settings) = settings {
+            env.write_config(settings);
+        }
+        // Finish migrations before background shell hooks can open the databases.
+        env.run(&["store", "status"]);
+        let (pty, config) = setup.spawn(&env, BTreeMap::new(), "");
+        Some(Self { pty, env, config })
+    }
+}
+
+/// A shell setup whose shell, and every file it needs, is installed.
+pub struct Setup {
+    path: PathBuf,
+    config: ShellConfig,
+    executable: PathBuf,
+    /// Where each file it needs is.
+    files: BTreeMap<String, String>,
+}
+
+impl Setup {
+    /// The shell setup at `path`; `None` (or a failure, where they are required) when its shell
+    /// or a file it needs isn't installed. Before anything is set up for it, so a setup skipped
+    /// costs nothing.
+    pub fn find(path: &Path) -> Option<Self> {
         let config: ShellConfig = toml_edit::de::from_str(&fs::read_to_string(path).unwrap())
             .unwrap_or_else(|e| panic!("{}: {e}", path.display()));
         let override_var = format!("ATUIN_E2E_{}", config.shell.to_uppercase());
@@ -52,7 +78,7 @@ impl Shell {
             missing(path, &config.shell);
             return None;
         };
-        let mut vars = BTreeMap::new();
+        let mut files = BTreeMap::new();
         for (name, default) in &config.required_files {
             let value = std::env::var(name).unwrap_or_else(|_| {
                 default.replace("$HOME", &std::env::var("HOME").unwrap_or_default())
@@ -61,20 +87,44 @@ impl Shell {
                 missing(path, &format!("{name}={value}"));
                 return None;
             }
-            vars.insert(name.clone(), value);
+            files.insert(name.clone(), value);
         }
-        let env = FreshEnv::new();
-        if let Some(settings) = settings {
-            env.write_config(settings);
+        Some(Self {
+            path: path.to_owned(),
+            config,
+            executable,
+            files,
+        })
+    }
+
+    /// The shell started in `env`, with `vars` set (over the files it needs, where they name the
+    /// same) and `init_args` passed to `atuin init`, at its prompt.
+    pub fn spawn(
+        self,
+        env: &FreshEnv,
+        mut vars: BTreeMap<String, String>,
+        init_args: &str,
+    ) -> (PtyShell, ShellConfig) {
+        let Self {
+            path,
+            mut config,
+            executable,
+            files,
+        } = self;
+        for (name, value) in files {
+            vars.entry(name).or_insert(value);
+        }
+        if !init_args.is_empty() {
+            let init = format!("atuin init {}", config.shell);
+            assert!(config.script.contains(&init), "{}: no `{init}`", path.display());
+            config.script = config.script.replace(&init, &format!("{init} {init_args}"));
         }
-        // Finish migrations before background shell hooks can open the databases.
-        env.run(&["store", "status"]);
         let rc = env.home().join(&config.rc);
         fs::create_dir_all(rc.parent().unwrap()).unwrap();
         fs::write(rc, &config.script).unwrap();
-        let pty = PtyShell::spawn(&executable, &config.args, &env, &vars);
+        let pty = PtyShell::spawn(&executable, &config.args, env, &vars);
         pty.wait_for_prompt();
-        Some(Self { pty, env, config })
+        (pty, config)
     }
 }
 

crates/atuin/tests/e2e_ai_resume_picker.rs

@@ -0,0 +1,131 @@
+//! `atuin ai resume`'s picker on a terminal: choosing a session, and where to resume it, runs the
+//! agent on it -- straight from the picker, and through each shell's widget, with and without
+//! `enter_accept`.
+
+#![cfg(all(unix, feature = "daemon", feature = "ai"))]
+
+mod common;
+
+#[allow(dead_code, reason = "shared with the other AI session tests")]
+#[path = "common/agents.rs"]
+mod agents;
+#[allow(dead_code, reason = "shared with the other AI session tests")]
+#[path = "common/ai.rs"]
+mod ai;
+#[allow(dead_code, reason = "shared with the other e2e test binaries")]
+#[path = "common/pty.rs"]
+mod pty;
+#[allow(dead_code, reason = "shared with the other e2e test binaries")]
+#[path = "common/shell.rs"]
+mod shell;
+
+use std::collections::BTreeMap;
+use std::path::{Path, PathBuf};
+
+use agents::{Agent, Transcript, Turn, ran, said};
+use ai::{Machine, shared_dir};
+use pty::PtyShell;
+use rstest::rstest;
+
+/// A machine with two Claude Code sessions captured, both run in `project`: the first is the one
+/// to pick.
+fn two_sessions(machine: &mut Machine, project: &Path) -> (Transcript, Turn) {
+    machine.start_daemon();
+    let (wanted, other) = (Turn::new(), Turn::new());
+    let other_session =
+        Transcript::create(machine.home(), Agent::Claude, project, std::slice::from_ref(&other));
+    let session =
+        Transcript::create(machine.home(), Agent::Claude, project, std::slice::from_ref(&wanted));
+    for (session, turn) in [(&session, &wanted), (&other_session, &other)] {
+        machine.wait_until_said(&session.id, &said(std::slice::from_ref(turn)));
+    }
+    (session, wanted)
+}
+
+/// Search the open picker for `turn`'s session by a word of its prompt, wait until it is the
+/// one shown, and accept it: the chooser opens on where to resume it.
+fn pick(pty: &PtyShell, turn: &Turn) {
+    let word = turn.prompt.split(' ').next_back().unwrap();
+    pty.wait_for("2 sessions");
+    pty.send_str(word);
+    pty.wait_for_screen("only the session searched for", |s| {
+        s.contains(" 1 session") && s.contains(word)
+    });
+    pty.send_enter();
+    pty.wait_for("Resume in");
+}
+
+/// The digit of the chooser's line for continuing in `agent`, once it is drawn.
+fn line_for(pty: &PtyShell, agent: Agent) -> char {
+    let is_line = |l: &str| l.contains(agent.label()) && l.contains("continue");
+    let screen =
+        pty.wait_for_screen(&format!("the line for {agent:?}"), |s| s.lines().any(is_line));
+    let line = screen.lines().find(|l| is_line(l)).unwrap();
+    line.chars().find(char::is_ascii_digit).unwrap()
+}
+
+/// With `enter_accept`, the picker runs the agent in place of itself: on the session chosen, in
+/// its own agent, where it ran, or -- chosen by its digit in the chooser -- on a new session
+/// continuing it in another.
+#[rstest]
+#[case::its_own_agent(None)]
+#[case::another_agent(Some(Agent::Codex))]
+fn the_picker_resumes_the_session_chosen(#[case] into: Option<Agent>) {
+    let project = shared_dir();
+    let mut machine = Machine::with_config("enter_accept = true\n", "capture_sessions = true\n");
+    let (session, turn) = two_sessions(&mut machine, project.path());
+
+    let vars = BTreeMap::from([("PATH".to_owned(), machine.path_var())]);
+    let pty = PtyShell::spawn(
+        Path::new(env!("CARGO_BIN_EXE_atuin")),
+        &["ai".to_owned(), "resume".to_owned()],
+        &machine.env,
+        &vars,
+    );
+    pick(&pty, &turn);
+
+    let Some(into) = into else {
+        pty.send_enter();
+        pty.wait_for(&ran(Agent::Claude, &format!("--resume {}", session.id), project.path()));
+        return;
+    };
+    pty.send(line_for(&pty, into).to_string().as_bytes());
+    let continued = machine.wait_for_child(into, &session.id);
+    assert_eq!(continued["parent_kind"], "continuation");
+    let id = continued["session_id"].as_str().unwrap();
+    pty.wait_for(&ran(into, &format!("resume {id}"), project.path()));
+}
+
+/// Through the shell widget, the session chosen resumes in the shell: run at once with
+/// `enter_accept`, else left on the command line to run.
+#[rstest]
+fn the_shell_widget_resumes_the_session_chosen(
+    #[files("tests/shells/*.toml")] setup: PathBuf,
+    #[values(false, true)] enter_accept: bool,
+) {
+    let Some(setup) = shell::Setup::find(&setup) else {
+        return;
+    };
+    let project = shared_dir();
+    let top = format!("enter_accept = {enter_accept}\n");
+    let mut machine = Machine::with_config(&top, "capture_sessions = true\n");
+    let (session, turn) = two_sessions(&mut machine, project.path());
+
+    let vars = BTreeMap::from([("PATH".to_owned(), machine.path_var())]);
+    let (pty, _) = setup.spawn(&machine.env, vars, "--bind-ai-resume");
+    // ctrl-]
+    pty.send(b"\x1d");
+    pick(&pty, &turn);
+    pty.send_enter();
+
+    let command = format!("claude --resume {}", session.id);
+    if !enter_accept {
+        pty.wait_for_screen("the command on the command line", |s| {
+            !s.contains("Resume in")
+                && s.lines().any(|l| l.contains(shell::PROMPT) && l.contains(&command))
+        });
+        assert!(!pty.screen().contains("AGENT-RAN"), "it ran before it was accepted");
+        pty.send_enter();
+    }
+    pty.wait_for_line(&ran(Agent::Claude, &format!("--resume {}", session.id), project.path()));
+}

crates/atuin/tests/e2e_ai_sessions.rs

@@ -0,0 +1,536 @@
+//! AI coding-agent sessions, end to end: captured from each agent's own transcripts by the daemon,
+//! synced between machines, and resumed with `atuin ai resume` -- restored, caught up, forked,
+//! switched or continued in another agent -- by the real binary, against transcripts written the
+//! way each agent writes them.
+//!
+//! Each test checks something a user relies on that no single crate can see whole: that what an
+//! agent wrote is what atuin shows and syncs, minus what it must never keep; that a session
+//! written back out reads as the same session to the agent and to atuin; and that what `resume`
+//! runs is the agent, on that session, where it ran.
+
+#![cfg(all(unix, feature = "daemon", feature = "ai", feature = "sync"))]
+
+mod common;
+
+#[allow(dead_code, reason = "shared with the PTY tests")]
+#[path = "common/agents.rs"]
+mod agents;
+#[allow(dead_code, reason = "shared with the PTY tests")]
+#[path = "common/ai.rs"]
+mod ai;
+
+use std::path::Path;
+
+use agents::{Agent, Transcript, Turn, said, whole, word};
+use ai::{Machine, SyncServer, shared_dir, succeeded};
+use atuin_common::harnesstools::resume::quote;
+use atuin_common::secrets::REDACTED;
+use regex::Regex;
+use rstest::{fixture, rstest};
+use tempfile::TempDir;
+
+/// A machine on its own, its daemon capturing.
+#[fixture]
+fn machine() -> Machine {
+    let mut machine = Machine::new();
+    machine.start_daemon();
+    machine
+}
+
+/// Where the sessions under test ran.
+#[fixture]
+fn project() -> TempDir {
+    shared_dir()
+}
+
+/// Whether `query` finds session `id` in `machine`'s search.
+fn finds(machine: &Machine, query: &str, id: &str) -> bool {
+    let found = machine.run(&["ai", "session", "search", query, "--style", "json"]);
+    let found: serde_json::Value = serde_json::from_str(&found).unwrap();
+    found.as_array().unwrap().iter().any(|m| m["session"]["session_id"] == id)
+}
+
+/// The last word of `text`: the one [`agents::word`] made, found nowhere else.
+fn last_word(text: &str) -> &str {
+    text.split(' ').next_back().unwrap()
+}
+
+/// An agent's transcript is captured whole: listed with where it ran, its conversation in order,
+/// and each tool call with what it was given and what it returned -- or, with `capture_tools`
+/// off, the tool's name alone. Reasoning is kept only as a marker, a secret anywhere is redacted,
+/// and search finds whole words of what was said and what a call was given, never what it
+/// returned.
+#[rstest]
+fn capture_keeps_the_conversation_and_tool_calls(
+    project: TempDir,
+    #[values(Agent::Claude, Agent::Codex, Agent::Opencode, Agent::Pi)] agent: Agent,
+    #[values(true, false)] capture_tools: bool,
+) {
+    let ai = format!("capture_sessions = true\ncapture_tools = {capture_tools}\n");
+    let mut machine = Machine::with_config("", &ai);
+    machine.start_daemon();
+    let secret = format!("ghp_{}{}", &word()[1..], "abcd");
+    let mut turns = vec![Turn::new(), Turn::new()];
+    turns[1].prompt = format!("use the token {secret} for {}", word());
+    turns[1].output = format!("token {secret} for {}", word());
+    let transcript = Transcript::create(machine.home(), agent, project.path(), &turns);
+
+    let mut want = said(&turns);
+    want[2] = want[2].replace(&secret, REDACTED);
+    machine.wait_until_said(&transcript.id, &want);
+
+    let session = machine.session(&transcript.id).unwrap();
+    assert_eq!(session["harness"], agent.harness());
+    assert_eq!(session["cwd"], project.path().to_str().unwrap());
+
+    let show = machine.show(&transcript.id);
+    let calls = ai::calls_in(&show);
+    assert_eq!(calls.len(), turns.len(), "one call a turn: {calls:#?}");
+    for (call, turn) in calls.iter().zip(&turns) {
+        assert_eq!(call.name, agent.tool());
+        let output = turn.output.replace(&secret, REDACTED);
+        if capture_tools {
+            assert!(call.input.contains(&turn.command), "{call:#?}");
+            assert!(call.output.contains(&output), "{call:#?}");
+        } else {
+            assert_eq!((call.input.as_str(), call.output.as_str()), ("", ""), "{call:#?}");
+        }
+    }
+    let shown = show.to_string();
+    for hidden in [&turns[0].thinking, &turns[1].thinking, &secret] {
+        assert!(!shown.contains(hidden.as_str()), "{hidden:?} was captured: {shown}");
+    }
+
+    let id = &transcript.id;
+    for said in [&turns[0].prompt, &turns[1].reply] {
+        assert!(finds(&machine, last_word(said), id), "{said:?} didn't find the session");
+    }
+    // Words match whole; only the picker, searching as you type, matches a prefix.
+    let prefix = &last_word(&turns[0].prompt)[..20];
+    assert!(!finds(&machine, prefix, id), "{prefix:?} found the session");
+    let command = last_word(&turns[0].command);
+    assert_eq!(finds(&machine, command, id), capture_tools, "search for the command");
+    for hidden in [&turns[0].thinking, &turns[0].output, &secret] {
+        assert!(!finds(&machine, last_word(hidden), id), "{hidden:?} found the session");
+    }
+}
+
+/// The history and output filters apply to the commands an agent runs as to the ones typed: one
+/// history would leave out is kept by its tool's name alone, and one whose output the output
+/// capture would leave out keeps what it ran, not what it printed.
+#[rstest]
+fn capture_applies_the_history_and_output_filters_to_agent_commands(project: TempDir) {
+    let top =
+        "history_filter = [\"^echo dropped-\"]\n[output]\ncommand_filter = [\"^echo quiet-\"]\n";
+    let mut machine = Machine::with_config(top, "capture_sessions = true\n");
+    machine.start_daemon();
+    let mut turns = vec![Turn::new(), Turn::new(), Turn::new()];
+    turns[0].command = format!("echo dropped-{}", word());
+    turns[1].command = format!("echo quiet-{}", word());
+    let transcript = Transcript::create(machine.home(), Agent::Claude, project.path(), &turns);
+    machine.wait_until_said(&transcript.id, &said(&turns));
+
+    let calls = ai::calls_in(&machine.show(&transcript.id));
+    let kept: Vec<(bool, bool)> = calls
+        .iter()
+        .zip(&turns)
+        .map(|(call, turn)| {
+            (call.input.contains(&turn.command), call.output.contains(&turn.output))
+        })
+        .collect();
+    assert_eq!(kept, [(false, false), (true, false), (true, true)], "{calls:#?}");
+    assert!(calls.iter().all(|call| call.name == Agent::Claude.tool()), "{calls:#?}");
+}
+
+/// A transcript the agent goes on writing is followed, and one it wrote to while the daemon was
+/// down is picked up where it was left: either way, every turn is in the session exactly once.
+#[rstest]
+fn capture_follows_a_transcript_exactly_once_across_restarts(
+    mut machine: Machine,
+    project: TempDir,
+    #[values(Agent::Claude, Agent::Codex, Agent::Opencode, Agent::Pi)] agent: Agent,
+) {
+    let turns: Vec<Turn> = (0..3).map(|_| Turn::new()).collect();
+
+    let transcript = Transcript::create(machine.home(), agent, project.path(), &turns[..1]);
+    machine.wait_until_said(&transcript.id, &said(&turns[..1]));
+
+    transcript.append(project.path(), &turns[1..2]);
+    machine.wait_until_said(&transcript.id, &said(&turns[..2]));
+
+    machine.stop_daemon();
+    transcript.append(project.path(), &turns[2..]);
+    machine.start_daemon();
+    machine.wait_until_said(&transcript.id, &said(&turns));
+}
+
+/// Turning capture on in the config takes effect in the running daemon, which then captures what
+/// the agents wrote while it was off.
+#[rstest]
+fn capture_starts_when_the_config_turns_it_on(project: TempDir) {
+    let mut machine = Machine::with_config("", "capture_sessions = false\n");
+    machine.start_daemon();
+    let turns = [Turn::new()];
+    let transcript = Transcript::create(machine.home(), Agent::Claude, project.path(), &turns);
+    // Longer than the daemon takes to see a new transcript when capturing.
+    std::thread::sleep(std::time::Duration::from_secs(3));
+    assert_eq!(machine.sessions(), Vec::<serde_json::Value>::new(), "captured while off");
+
+    let config = machine.home().join(".config/atuin/config.toml");
+    let on = std::fs::read_to_string(&config)
+        .unwrap()
+        .replace("capture_sessions = false", "capture_sessions = true");
+    std::fs::write(&config, on).unwrap();
+    machine.wait_until_said(&transcript.id, &said(&turns));
+}
+
+/// The command `atuin ai resume --print` prints to resume `transcript` in its own agent, from
+/// `cwd`.
+fn resume_line(transcript: &Transcript, cwd: &Path) -> String {
+    let id = transcript.id.as_str();
+    let args = match transcript.agent {
+        Agent::Claude => vec!["--resume", id],
+        Agent::Codex => vec!["resume", id],
+        Agent::Opencode => vec!["--session", id],
+        Agent::Pi => vec!["--session", transcript.path.to_str().unwrap()],
+    };
+    let mut line =
+        format!("cd -- {} && {}", quote(cwd.to_str().unwrap()), transcript.agent.program());
+    for arg in args {
+        line.push(' ');
+        line.push_str(&quote(arg));
+    }
+    line
+}
+
+/// A subagent isn't a session to resume, but the session it works for, which is listed as its
+/// parent: naming the subagent resumes that, saying so.
+#[rstest]
+fn a_subagent_resumes_as_the_session_it_works_for(machine: Machine, project: TempDir) {
+    let turns = [Turn::new()];
+    let parent = Transcript::create(machine.home(), Agent::Claude, project.path(), &turns);
+    let subagent = Transcript::claude_subagent(&parent, project.path(), &[Turn::new()]);
+    machine.wait_until_said(&parent.id, &said(&turns));
+    let session = machine.wait_for_child(Agent::Claude, &parent.id);
+    assert_eq!(session["session_id"], subagent.id.as_str());
+    assert_eq!(session["parent_kind"], "subagent");
+
+    let (stdout, stderr) = succeeded(&machine.resume(&[&subagent.id, "--print"], machine.home()));
+    assert_eq!(stdout.trim(), resume_line(&parent, project.path()));
+    assert!(stderr.contains(&format!("{} is a subagent", subagent.id)), "{stderr}");
+}
+
+/// A session whose transcript is on this machine resumes in its agent where it ran, named by the
+/// agent's id, by its atuin id, or by enough of that, and is left as it was. How a name is
+/// resolved is the same for every agent; where its transcript is, and how it resumes, isn't.
+#[rstest]
+#[case::claude(Agent::Claude, "agent id")]
+#[case::codex(Agent::Codex, "agent id")]
+#[case::opencode(Agent::Opencode, "agent id")]
+#[case::pi(Agent::Pi, "agent id")]
+#[case::by_atuin_id(Agent::Opencode, "atuin id")]
+#[case::by_atuin_id_prefix(Agent::Opencode, "atuin id prefix")]
+fn a_local_session_resumes_in_its_agent_where_it_ran(
+    machine: Machine,
+    project: TempDir,
+    #[case] agent: Agent,
+    #[case] named_by: &str,
+) {
+    let turns = [Turn::new()];
+    let transcript = Transcript::create(machine.home(), agent, project.path(), &turns);
+    machine.wait_until_said(&transcript.id, &said(&turns));
+    let before = std::fs::read(&transcript.path).unwrap();
+
+    let atuin_id =
+        machine.session(&transcript.id).unwrap()["atuin_id"].as_str().unwrap().to_owned();
+    let name = match named_by {
+        "agent id" => transcript.id.clone(),
+        "atuin id" => atuin_id,
+        // Past the timestamp the id starts with, which sessions started together share.
+        _ => atuin_id[..16].to_owned(),
+    };
+    let (stdout, stderr) = succeeded(&machine.resume(&[&name, "--print"], machine.home()));
+    assert_eq!(stdout.trim(), resume_line(&transcript, project.path()), "{stderr}");
+    assert_eq!(std::fs::read(&transcript.path).unwrap(), before, "the transcript changed");
+}
+
+/// Two machines of one account, and a session captured on the first, on the second written back
+/// out from sync by resuming it there: what the second's agent would open.
+struct Restored {
+    _server: SyncServer,
+    project: TempDir,
+    here: Machine,
+    there: Machine,
+    turns: Vec<Turn>,
+    /// The session on the machine it was captured on.
+    original: Transcript,
+    /// Its transcript written back out on the other.
+    restored: Transcript,
+}
+
+fn restored(agent: Agent) -> Restored {
+    let server = SyncServer::start();
+    let project = project();
+    let (there, here) = Machine::pair(&server);
+    let turns = vec![Turn::new(), Turn::new()];
+    let original = Transcript::create(there.home(), agent, project.path(), &turns);
+    there.wait_until_said(&original.id, &said(&turns));
+    here.wait_until_said(&original.id, &said(&turns));
+    assert!(Transcript::find(here.home(), agent, &original.id).is_none());
+
+    let (stdout, stderr) = succeeded(&here.resume(&[&original.id, "--print"], here.home()));
+    let restored = Transcript::find(here.home(), agent, &original.id)
+        .unwrap_or_else(|| panic!("nothing restored: {stdout} {stderr}"));
+    assert_eq!(stdout.trim(), resume_line(&restored, project.path()), "{stderr}");
+    Restored {
+        _server: server,
+        project,
+        here,
+        there,
+        turns,
+        original,
+        restored,
+    }
+}
+
+/// A session captured on one machine is listed on another, under the same atuin id, and resuming
+/// it there writes it back out as its agent's own transcript: the conversation and its tool calls
+/// as they were, without the reasoning sync never held. The agent going on with it
+/// there carries on the same session -- nothing in it read twice -- which syncs back.
+#[rstest]
+fn a_session_from_another_machine_restores_as_the_same_session(
+    #[values(Agent::Claude, Agent::Codex, Agent::Pi)] agent: Agent,
+) {
+    let Restored {
+        _server,
+        project,
+        here,
+        there,
+        turns,
+        original,
+        restored,
+    } = restored(agent);
+    assert_eq!(
+        here.session(&original.id).unwrap()["atuin_id"],
+        there.session(&original.id).unwrap()["atuin_id"]
+    );
+
+    let text = restored.read();
+    assert!(ai::holds_in_order(&text, &whole(&turns)), "the session isn't all there:\n{text}");
+    assert!(!text.contains("[ran a shell command]"), "a tool call became a note:\n{text}");
+    for turn in &turns {
+        assert!(!text.contains(&turn.thinking), "the reasoning was restored:\n{text}");
+    }
+    let said = said(&turns);
+
+    let more = vec![Turn::new()];
+    restored.append(project.path(), &more);
+    let all = [said, agents::said(&more)].concat();
+    here.wait_until_said(&original.id, &all);
+    there.wait_until_said(&original.id, &all);
+}
+
+/// A copy behind the session's other machine catches up with it as it resumes, saying so, and
+/// once caught up resumes as it is.
+#[rstest]
+fn a_copy_behind_another_machine_catches_up_as_it_resumes(
+    #[values(Agent::Claude, Agent::Codex, Agent::Pi)] agent: Agent,
+) {
+    let Restored {
+        _server,
+        project,
+        here,
+        there,
+        mut turns,
+        original,
+        restored,
+    } = restored(agent);
+    turns.push(Turn::new());
+    original.append(project.path(), &turns[2..]);
+    here.wait_until_said(&original.id, &said(&turns));
+
+    let (stdout, stderr) = succeeded(&here.resume(&[&original.id, "--print"], here.home()));
+    assert_eq!(stdout.trim(), resume_line(&restored, project.path()));
+    let caught_up =
+        Regex::new(&format!(r"^atuin: caught up: \d+ messages? from {}\n$", there.short_host()));
+    assert!(caught_up.unwrap().is_match(&stderr), "{stderr}");
+    let text = restored.read();
+    assert!(ai::holds_in_order(&text, &said(&turns)), "not caught up:\n{text}");
+
+    let (_, stderr) = succeeded(&here.resume(&[&original.id, "--print"], here.home()));
+    assert_eq!(stderr, "", "caught up twice");
+    assert_eq!(restored.read(), text);
+}
+
+/// Where a copy and the session's other machine went different ways, resuming it is a choice:
+/// it fails saying so, and naming the flags that choose. `--as-is` resumes the copy untouched,
+/// `--fork` forks the newest branch into a new session, and `--switch` puts the copy on the
+/// other machine's branch, keeping it whole as a backup.
+#[rstest]
+fn a_diverged_copy_resumes_only_as_chosen(
+    #[values(Agent::Claude, Agent::Codex, Agent::Pi)] agent: Agent,
+) {
+    let Restored {
+        _server,
+        project,
+        here,
+        there,
+        turns,
+        original,
+        restored,
+    } = restored(agent);
+    // Theirs is the newest, which a copy elsewhere would catch up to.
+    let (ours, theirs) = (Turn::new(), Turn::new());
+    restored.append(project.path(), std::slice::from_ref(&ours));
+    original.append(project.path(), std::slice::from_ref(&theirs));
+    ai::wait("both branches", || {
+        let show = here.show(&original.id).to_string();
+        show.contains(&theirs.reply) && show.contains(&ours.reply)
+    });
+    let copy = restored.read();
+    let resume =
+        |flags: &[&str]| here.resume(&[&[original.id.as_str()][..], flags].concat(), here.home());
+
+    let stderr = ai::failed(&resume(&["--print"]));
+    assert!(
+        stderr.contains(&format!("this copy went another way than {}'s", there.short_host())),
+        "{stderr}"
+    );
+    for flag in ["--as-is", "--fork", "--switch"] {
+        assert!(stderr.contains(flag), "{flag} not offered: {stderr}");
+    }
+    assert_eq!(restored.read(), copy);
+
+    let (stdout, _) = succeeded(&resume(&["--as-is", "--print"]));
+    assert_eq!(stdout.trim(), resume_line(&restored, project.path()));
+    assert_eq!(restored.read(), copy);
+
+    let (_, stderr) = succeeded(&resume(&["--fork", "--print"]));
+    assert!(stderr.contains(&format!("forked into a new {} session", agent.label())), "{stderr}");
+    let fork = here.wait_for_child(agent, &original.id);
+    assert_eq!(fork["parent_kind"], "fork");
+    let fork_id = fork["session_id"].as_str().unwrap();
+    let newest = [said(&turns), said(std::slice::from_ref(&theirs))].concat();
+    here.wait_until_holds(fork_id, &newest);
+    assert_eq!(restored.read(), copy);
+
+    let (_, stderr) = succeeded(&resume(&["--switch", "--print"]));
+    let backup = stderr
+        .split("(your copy is at ")
+        .nth(1)
+        .and_then(|rest| rest.split(')').next())
+        .unwrap_or_else(|| panic!("no backup named: {stderr}"));
+    assert!(
+        stderr.starts_with(&format!("atuin: switched to {}'s branch: ", there.short_host())),
+        "{stderr}"
+    );
+    assert!(
+        Path::new(backup).starts_with(here.data_dir().join("ai/switched").join(agent.harness())),
+        "{backup}"
+    );
+    assert_eq!(std::fs::read_to_string(backup).unwrap(), copy, "the backup isn't the copy");
+    let text = restored.read();
+    let theirs_now = [said(&turns), said(std::slice::from_ref(&theirs))].concat();
+    assert!(ai::holds_in_order(&text, &theirs_now), "not switched:\n{text}");
+    assert!(!text.contains(&ours.reply), "our branch is still in it:\n{text}");
+}
+
+/// Forking writes the session out as a new one of its agent, with the whole conversation, linked
+/// to it as a fork, and resumes that; the original is left as it was.
+#[rstest]
+fn forking_a_session_leaves_it_and_links_the_fork(
+    machine: Machine,
+    project: TempDir,
+    #[values(Agent::Claude, Agent::Codex, Agent::Pi)] agent: Agent,
+) {
+    let turns = vec![Turn::new(), Turn::new()];
+    let original = Transcript::create(machine.home(), agent, project.path(), &turns);
+    machine.wait_until_said(&original.id, &said(&turns));
+    let before = original.read();
+
+    let (stdout, stderr) =
+        succeeded(&machine.resume(&[&original.id, "--fork", "--print"], machine.home()));
+    assert_eq!(stderr.trim(), format!("atuin: forked into a new {} session", agent.label()));
+    let fork = machine.wait_for_child(agent, &original.id);
+    assert_eq!(fork["parent_kind"], "fork");
+    let fork =
+        Transcript::find(machine.home(), agent, fork["session_id"].as_str().unwrap()).unwrap();
+    assert_eq!(stdout.trim(), resume_line(&fork, project.path()));
+    machine.wait_until_holds(&fork.id, &said(&turns));
+    assert_eq!(original.read(), before, "the original changed");
+}
+
+/// Continuing a session in another agent writes it out as a new session of that agent, which
+/// carries the conversation on, each shell command now that agent's own shell tool with what it
+/// printed, and only the reasoning left out, as the status line says. It is linked to the
+/// original as its continuation, and resumed.
+#[rstest]
+#[case::claude_to_codex(Agent::Claude, Agent::Codex)]
+#[case::claude_to_pi(Agent::Claude, Agent::Pi)]
+#[case::codex_to_claude(Agent::Codex, Agent::Claude)]
+#[case::codex_to_pi(Agent::Codex, Agent::Pi)]
+#[case::opencode_to_claude(Agent::Opencode, Agent::Claude)]
+#[case::opencode_to_codex(Agent::Opencode, Agent::Codex)]
+#[case::pi_to_claude(Agent::Pi, Agent::Claude)]
+#[case::pi_to_codex(Agent::Pi, Agent::Codex)]
+fn continuing_in_another_agent_carries_the_conversation(
+    machine: Machine,
+    project: TempDir,
+    #[case] from: Agent,
+    #[case] to: Agent,
+) {
+    let turns = vec![Turn::new(), Turn::new()];
+    let original = Transcript::create(machine.home(), from, project.path(), &turns);
+    machine.wait_until_said(&original.id, &said(&turns));
+
+    let (stdout, stderr) = succeeded(
+        &machine.resume(&[&original.id, "--in", to.program(), "--print"], machine.home()),
+    );
+    assert_eq!(stderr.trim(), format!("atuin: continuing in {}: reasoning dropped", to.label()));
+    let continued = machine.wait_for_child(to, &original.id);
+    assert_eq!(continued["parent_kind"], "continuation");
+    assert_eq!(continued["parent"]["harness"], from.harness());
+    let continued =
+        Transcript::find(machine.home(), to, continued["session_id"].as_str().unwrap()).unwrap();
+    assert_eq!(stdout.trim(), resume_line(&continued, project.path()));
+
+    machine.wait_until_holds(&continued.id, &said(&turns));
+    let calls = ai::calls_in(&machine.show(&continued.id));
+    assert_eq!(calls.len(), turns.len(), "one call a turn: {calls:#?}");
+    for (call, turn) in calls.iter().zip(&turns) {
+        assert_eq!(call.name, to.tool(), "{call:#?}");
+        assert!(call.input.contains(&turn.command), "{call:#?}");
+        assert!(call.output.contains(&turn.output), "{call:#?}");
+    }
+    let text = continued.read();
+    for turn in &turns {
+        assert!(!text.contains(&turn.thinking), "the reasoning was carried over:\n{text}");
+    }
+}
+
+/// The index is the record store's to rebuild: what every session says, and how sessions link,
+/// is the same after `atuin store rebuild ai-session` as before.
+#[rstest]
+fn rebuilding_the_index_reproduces_it(machine: Machine, project: TempDir) {
+    let mut ids = Vec::new();
+    let turns = [Turn::new()];
+    for agent in Agent::ALL {
+        let transcript = Transcript::create(machine.home(), agent, project.path(), &turns);
+        machine.wait_until_said(&transcript.id, &said(&turns));
+        ids.push(transcript.id);
+    }
+    // A continuation, for the link between sessions.
+    succeeded(&machine.resume(&[&ids[0], "--in", "codex", "--print"], machine.home()));
+    let continued = machine.wait_for_child(Agent::Codex, &ids[0]);
+    ids.push(continued["session_id"].as_str().unwrap().to_owned());
+    machine.wait_until_holds(&ids[4], &said(&turns));
+
+    let index = |machine: &Machine| -> Vec<serde_json::Value> {
+        ids.iter().map(|id| machine.show(id)).collect()
+    };
+    let before = index(&machine);
+    let out = machine.run(&["store", "rebuild", "ai-session"]);
+    assert!(out.contains("rebuilding"), "{out}");
+    // `ai session` commands wait for the rebuild.
+    assert_eq!(index(&machine), before);
+}

docs/docs/ai/sessions.md

@@ -75,10 +75,10 @@ after the daemon restarts. Sessions already captured keep what they had.
 atuin ai session search "flaky test"
 ```
 
-- The last word of the query matches as a prefix once it's at least two
-  characters long, so `atuin ai session search "migr"` finds sessions
-  mentioning "migration." A single character still matches only a whole word.
-  Earlier words always match whole.
+- Each word of the query matches a whole word: `atuin ai session search
+  "migration"` finds sessions mentioning migration, but `"migr"` doesn't.
+  The `atuin ai resume` picker searches as you type, so there the last word
+  also matches as a prefix once it's at least two characters long.
 - An empty query (`atuin ai session search ""`) lists the newest sessions
   first, rather than returning nothing.
 - `--harness` limits the search to one agent.