feat(ai): catch sessions up with sync on resume, and fork from another machine's branch (atuinsh/atuin#4345)

This commit is contained in:
Ellie Huxtable
2026-10-05 22:45:45 +00:00
committed by Ellie Huxtable
parent 2b9591d9ab
commit d5c6bee0f7
16 changed files with 2537 additions and 153 deletions
+366 -37
View File
@@ -16,9 +16,19 @@
//! picker, accepting a session asks where to resume it: its own harness first, then the others
//! installed here (see [`crate::resume_tui::chooser`]; `[ai.sessions] resume_chooser = false`
//! skips that).
//! - A session whose transcript is here is caught up with sync first (see
//! [`crate::resume_tui::catchup`]): when it went on on another machine, the messages this copy
//! lacks are appended. When that can't be done (an agent here has it open, it went another way
//! here, or this copy has messages sync hasn't got), nothing is written, and it fails saying
//! so: `--as-is` resumes this copy as it is, `--fork` forks it instead.
//! - `--fork` forks it instead: it is written out as a new session of its own harness, with the
//! same history, linked to it as a fork (see [`atuin_common::harnesstools::fork`]), and that is
//! resumed. The original is left as it is.
//! resumed. The original is left as it is. A session that went on separately on several
//! machines forks from its newest branch.
//! - `--branch` picks the branch to catch up to, or to fork from: `this`, `@<host id>`, or the
//! start of its head's id.
//! - `--as-is` and `--branch` need an id naming a single session that can resume here: anything
//! else is an error (listing the sessions an id names), never the picker, which would drop them.
//! - From the shell widget (`--shell-widget`), the result goes to stderr using the history
//! search's protocol: `__atuin_accept__:<cmd>` to run it, plain `<cmd>` to edit it, nothing to
//! leave the command line alone.
@@ -30,7 +40,7 @@
use std::io::{self, IsTerminal, Write};
use std::sync::Arc;
use atuin_client::ai_session::{AtuinSessionId, HarnessKind};
use atuin_client::ai_session::{Analysis, AtuinSessionId, HarnessKind, SourceId};
use atuin_client::settings::{AiSessionFilterMode, Settings};
use atuin_client::theme::ThemeManager;
use atuin_common::string::EscapeNonPrintablePosixExt as _;
@@ -38,7 +48,8 @@ use clap::Args;
use eyre::{Result, bail};
use super::session::one_line;
use crate::resume_tui::resumer::{ForkFrom, HarnessResumer, Resume, shell_line};
use crate::resume_tui::catchup::{self, CatchUp, Held};
use crate::resume_tui::resumer::{HarnessResumer, NotResumable, Resume, quote, shell_line};
use crate::resume_tui::sidecar::SidecarSource;
use crate::resume_tui::source::{Relation, harness_label};
use crate::resume_tui::{
@@ -81,6 +92,16 @@ pub struct Cmd {
#[arg(long, conflicts_with = "continue_in")]
fork: bool,
/// Resume this machine's copy of the session QUERY names as it is, without catching it up
/// with sync.
#[arg(long, conflicts_with_all = ["fork", "continue_in"])]
as_is: bool,
/// The branch of the session QUERY names to catch up to, or to fork from with `--fork`:
/// `this`, `@<host id>`, or the start of its head's id. The newest by default.
#[arg(long, value_name = "BRANCH", conflicts_with_all = ["continue_in", "as_is"])]
branch: Option<String>,
/// The filter the picker opens in (default: workspace, widening to global).
#[arg(long, value_enum)]
filter_mode: Option<AiSessionFilterMode>,
@@ -190,12 +211,19 @@ fn picker_has_a_terminal() -> bool {
/// What to say when `query` names several sessions and there's no terminal to pick one on: each
/// on a line of its own, with its harness, when it last changed (in `offset`) and its title.
fn ambiguous_note(query: &str, found: &[SessionRow], offset: time::UtcOffset) -> String {
let format = time::macros::format_description!("[year]-[month]-[day] [hour]:[minute]");
let width = found.iter().map(|r| harness_label(r.handle.harness).len()).max().unwrap_or(0);
let mut note = format!(
let note = format!(
"{query} names {} sessions; run `atuin ai resume {query}` on a terminal to pick one:",
found.len()
);
note + &listing(found, offset)
}
/// `found`, each on a line of its own, with its harness, when it last changed (in `offset`) and
/// its title.
fn listing(found: &[SessionRow], offset: time::UtcOffset) -> String {
let format = time::macros::format_description!("[year]-[month]-[day] [hour]:[minute]");
let width = found.iter().map(|r| harness_label(r.handle.harness).len()).max().unwrap_or(0);
let mut note = String::new();
for row in found {
let when = row.updated_at.to_offset(offset).format(&format).unwrap_or_default();
note.push_str(&format!(
@@ -208,6 +236,59 @@ fn ambiguous_note(query: &str, found: &[SessionRow], offset: time::UtcOffset) ->
note
}
/// What to say when `flag` (`--branch`, `--as-is`) is given with a `query` that names no single
/// session: the picker would drop the flag, so it is an error, listing what `query` names.
fn be_more_specific(
flag: &str,
query: &str,
found: &[SessionRow],
offset: time::UtcOffset,
) -> String {
let query = query.escape_non_printable();
match found.len() {
0 => format!("`{flag}` resumes the session an id names: no session has the id \"{query}\""),
n => format!(
"`{flag}` resumes the session an id names: {query} names {n} sessions; be more \
specific:{}",
listing(found, offset)
),
}
}
/// The session named, `row`, can't be resumed here, for `why`: the picker opens on it instead, so
/// the reason shows (and another can be picked). With `flag` (`--branch`, `--as-is`) given, which
/// the picker would drop, an error saying why.
fn picker_on(
row: SessionRow,
why: NotResumable,
flag: Option<&str>,
) -> Result<(SessionRow, NotResumable)> {
if let Some(flag) = flag {
let message = format!("can't resume {} with `{flag}`: {why}", row.handle.session);
bail!("{}", message.escape_non_printable());
}
Ok((row, why))
}
/// With `flag` (`--branch`, `--as-is`) given and `query` naming no single session, the error
/// saying to be more specific ([`be_more_specific`]).
async fn no_single_session_for(
source: &dyn SessionSource,
query: &str,
flag: Option<&str>,
) -> Result<()> {
let Some(flag) = flag else {
return Ok(());
};
let found = if looks_like_id(query) {
source.find_by_id(query).await?
} else {
Vec::new()
};
let offset = time::UtcOffset::current_local_offset().unwrap_or(time::UtcOffset::UTC);
bail!("{}", be_more_specific(flag, query, &found, offset))
}
/// The sessions `query` names when it is an id naming several (see [`direct_match`]), for the
/// picker to open on; empty when it names one or none. Without a terminal to pick one on, an
/// error listing them.
@@ -292,7 +373,10 @@ pub async fn run(cmd: Cmd, settings: &Settings) -> Result<()> {
let (source, resumer, query) = (source.as_ref(), resumer.as_ref(), query.trim());
let (plan, status) = match cmd.continue_in {
Some(into) => continue_plan(source, resumer, query, into).await?,
None => fork_plan(source, resumer, query).await?,
None => {
let branch = cmd.branch.as_deref();
fork_plan(source, resumer, query, branch, &context.host_id).await?
}
};
// The widget reads stderr for the command: nothing else may go there.
if output != Output::Widget {
@@ -303,6 +387,11 @@ pub async fn run(cmd: Cmd, settings: &Settings) -> Result<()> {
let mut preselect = None;
let target = direct_target(source.as_ref(), query.trim()).await?;
// These only resume the session an id names: the picker would drop them.
let flag = cmd.branch.as_ref().map(|_| "--branch").or(cmd.as_is.then_some("--as-is"));
if target.is_none() {
no_single_session_for(source.as_ref(), query.trim(), flag).await?;
}
// An id naming several sessions opens the picker on them, to pick one.
let matches = if target.is_none() {
ambiguous_matches(source.as_ref(), query.trim(), picker_has_a_terminal()).await?
@@ -316,34 +405,20 @@ pub async fn run(cmd: Cmd, settings: &Settings) -> Result<()> {
{
eprintln!("atuin: {note}");
}
let plan = match resumer.plan(&row).await {
Ok(Resume {
plan,
restore: None,
}) => Ok(plan),
Ok(Resume {
restore: Some(restore),
..
}) => {
let plan = resumer.restore(source.as_ref(), &row, &restore).await;
// The widget reads stderr for the command: nothing else may go there.
if plan.is_ok()
&& output != Output::Widget
&& let Some(note) = &restore.note
{
eprintln!(
"atuin: restored the session from sync; {}",
note.escape_non_printable()
);
}
plan
}
Err(why) => Err(why),
};
let branch = cmd.branch.as_deref();
let (source, resumer) = (source.as_ref(), resumer.as_ref());
let plan = direct_plan(source, resumer, &row, cmd.as_is, branch, &context.host_id).await?;
match plan {
Ok(plan) => return finish(direct_outcome(plan, output, settings.enter_accept), output),
// Open the picker on it instead, so the reason shows (and another can be picked).
Err(why) => preselect = Some((row, why)),
Ok((plan, status)) => {
// The widget reads stderr for the command: nothing else may go there.
if let Some(status) = status
&& output != Output::Widget
{
eprintln!("atuin: {}", status.escape_non_printable());
}
return finish(direct_outcome(plan, output, settings.enter_accept), output);
}
Err(why) => preselect = Some(picker_on(row, why, flag)?),
}
}
@@ -370,6 +445,91 @@ pub async fn run(cmd: Cmd, settings: &Settings) -> Result<()> {
finish(outcome, output)
}
/// `atuin ai resume <id>`: the plan resuming `row` in its own agent, caught up with sync first
/// (with `as_is`, as it is) to the head `branch` names, and what to say of it; inside, why it
/// can't resume here. A choice to make (see [`Held`]) is an error saying how to make it.
async fn direct_plan(
source: &dyn SessionSource,
resumer: &dyn Resumer,
row: &SessionRow,
as_is: bool,
branch: Option<&str>,
here: &str,
) -> Result<Result<(ResumePlan, Option<String>), NotResumable>> {
if as_is {
match resumer.plan(row).await {
Ok(Resume {
plan,
restore: None,
..
}) => return Ok(Ok((plan, None))),
Err(why) => return Ok(Err(why)),
// No copy here to resume as it is: it is restored.
Ok(_) => {}
}
}
let head = match branch {
Some(selector) => Some(pick_head(source, row, selector, here).await?),
None => None,
};
Ok(match resumer.catch_up(source, row, head.as_ref()).await {
Ok(CatchUp::Ready { plan, status }) => Ok((plan, status)),
Ok(CatchUp::Choice(held)) => bail!("{}", choice_note(row, &held, here)),
Err(why) => Err(why),
})
}
/// The head of `row` that `selector` names (`--branch`).
async fn pick_head(
source: &dyn SessionSource,
row: &SessionRow,
selector: &str,
here: &str,
) -> Result<SourceId> {
let analysis = source.analyse(&row.handle).await?;
let heads = analysis.as_ref().map(Analysis::heads).unwrap_or_default();
let now = time::OffsetDateTime::now_utc();
match catchup::pick_branch(heads, selector, now, here) {
Ok(head) => Ok(head.source_id.clone()),
Err(why) => {
// Line by line: the listing of branches keeps its lines.
let lines: Vec<String> =
why.lines().map(|l| l.escape_non_printable().to_string()).collect();
bail!("--branch: {}", lines.join("\n"))
}
}
}
/// What to say when catching `row` up needs a choice: why, and the flags that make it. Ids are
/// agent-supplied: shown without their control characters.
fn choice_note(row: &SessionRow, held: &Held, here: &str) -> String {
let id = quote(row.handle.session.as_ref());
let id = id.escape_non_printable();
let status = held.status();
let mut note =
format!("{}: `--as-is` resumes this copy as it is", status.escape_non_printable());
if let Some(branch) = held.branches.get(held.chosen) {
let pick = if held.chosen == 0 {
String::new()
} else {
format!(" --branch {}", quote(&branch.selector).escape_non_printable())
};
note.push_str(&format!(
", `atuin ai resume {id} --fork{pick}` forks it from {}'s branch",
branch.host
));
}
if held.branches.len() > 1 {
note.push_str("; `--branch` picks another:");
let now = time::OffsetDateTime::now_utc();
for branch in &held.branches {
let line = catchup::describe(&branch.head, now, here);
note.push_str(&format!("\n {} {line}", branch.selector.escape_non_printable()));
}
}
note
}
/// `atuin ai resume <id> --in <harness>`: write the session `query` names out as a new session
/// of `into`, and the plan that resumes it, with the status line saying what was flattened.
async fn continue_plan(
@@ -401,11 +561,19 @@ async fn fork_plan(
source: &dyn SessionSource,
resumer: &dyn Resumer,
query: &str,
branch: Option<&str>,
here: &str,
) -> Result<(ResumePlan, String)> {
let Some((row, redirected)) = direct_target(source, query).await? else {
bail!("`--fork` forks the session an id names: no single session has the id {query:?}");
};
let forked = resumer.fork(source, &row, ForkFrom::default()).await.map_err(|why| {
let head = match branch {
Some(selector) => Some(pick_head(source, &row, selector, here).await?),
None => None,
};
let from = catchup::fork_from(source, &row.handle, head.as_ref()).await;
let forked = async { resumer.fork(source, &row, from?).await };
let forked = forked.await.map_err(|why| {
let message = format!("can't fork {}: {why}", row.handle.session);
eyre::eyre!("{}", message.escape_non_printable())
})?;
@@ -586,6 +754,68 @@ mod tests {
);
}
/// `--branch` and `--as-is` only resume the session an id names: with an id naming several,
/// or none, it is an error listing them, never the picker, which would drop the flag.
#[rstest]
#[tokio::test]
async fn a_flag_on_no_single_session_says_to_be_more_specific() {
use atuin_client::ai_session::HarnessKind;
let mut claude = fake::row(HarnessKind::ClaudeCode, "7aaabc31-1631", "fix the flaky test");
claude.updated_at = time::macros::datetime!(2026-09-27 14:03 UTC);
let mut pi = fake::row(HarnessKind::Pi, "7aaabc31-1631", "plan the dotfiles sync");
pi.updated_at = time::macros::datetime!(2026-09-26 09:12 UTC);
assert_eq!(
be_more_specific("--branch", "7aaabc31", &[claude, pi], time::UtcOffset::UTC),
[
"`--branch` resumes the session an id names: 7aaabc31 names 2 sessions; be more \
specific:",
" Claude Code 2026-09-27 14:03 fix the flaky test (7aaabc31-1631)",
" Pi 2026-09-26 09:12 plan the dotfiles sync (7aaabc31-1631)",
]
.join("\n")
);
assert_eq!(
be_more_specific("--as-is", "flaky", &[], time::UtcOffset::UTC),
"`--as-is` resumes the session an id names: no session has the id \"flaky\""
);
let source = FakeSource::new();
let ambiguous = "ses_4b8e2f1a9c3d7e";
assert!(direct_target(&source, ambiguous).await.unwrap().is_none());
for flag in ["--branch", "--as-is"] {
let err = no_single_session_for(&source, ambiguous, Some(flag)).await.unwrap_err();
let err = err.to_string();
assert!(err.contains("be more specific"), "{err}");
assert!(err.contains("ses_4b8e2f1a9c3d7e6f"), "{err}");
let err = no_single_session_for(&source, "flaky", Some(flag)).await.unwrap_err();
assert!(err.to_string().contains("no session has the id"), "{err}");
}
// Without them, the picker opens on the matches.
assert!(no_single_session_for(&source, ambiguous, None).await.is_ok());
}
/// A session named that can't be resumed opens the picker on it, but not with `--branch` or
/// `--as-is`, which the picker would drop: an error saying why.
#[rstest]
#[case::plain(None, None)]
#[case::branch(Some("--branch"), Some("can't resume s with `--branch`: catching it up"))]
#[case::as_is(Some("--as-is"), Some("can't resume s with `--as-is`: catching it up"))]
fn a_session_that_cant_resume_keeps_its_flags(
#[case] flag: Option<&str>,
#[case] err: Option<&str>,
) {
use atuin_client::ai_session::HarnessKind;
let row = fake::row(HarnessKind::ClaudeCode, "s", "t");
let why = NotResumable::CatchUp("no longer a head".to_owned());
match (picker_on(row, why, flag), err) {
(Ok((row, _)), None) => assert_eq!(row.handle.session.as_ref(), "s"),
(Err(e), Some(want)) => assert!(e.to_string().starts_with(want), "{e}"),
(got, _) => panic!("{got:?}"),
}
}
/// An id naming several sessions opens the picker on all of them, given a terminal to pick
/// one on; without one, it fails, listing them. A unique id, or a query that isn't an id,
/// pins nothing.
@@ -742,12 +972,14 @@ mod tests {
async fn forking_by_id_plans_the_fork() {
let source = FakeSource::new();
let resumer = FakeResumer::default();
let (plan, status) = fork_plan(&source, &resumer, "7f3c9a12").await.unwrap();
let (plan, status) =
fork_plan(&source, &resumer, "7f3c9a12", None, fake::THIS_HOST_ID).await.unwrap();
let fork = "forked-7f3c9a12-5be0-4d7e-9c41-0a8e2b6f4d10";
assert!(plan.args.iter().any(|a| a == fork), "{plan:?}");
assert_eq!(status, "forked into a new Claude Code session");
let err = fork_plan(&source, &resumer, "fix flaky").await.unwrap_err();
let err =
fork_plan(&source, &resumer, "fix flaky", None, fake::THIS_HOST_ID).await.unwrap_err();
assert!(err.to_string().contains("no single session has the id"), "{err}");
let cli = Cli::try_parse_from(["resume", "7f3c9a12", "--fork", "--print"]).unwrap();
@@ -756,6 +988,103 @@ mod tests {
assert!(Cli::try_parse_from(["resume", "7f3c9a12", "--fork", "--in", "pi"]).is_err());
}
/// The fake session `7f3c9a12…`, with sync holding two branches of it: this host's `a`-`b`-
/// `x`-`y`, the newest, and another's `a`-`b`-`c`-`d`.
async fn branched() -> (FakeSource, SessionRow) {
let row = direct_match(&FakeSource::new(), "7f3c9a12").await.unwrap().unwrap();
let rows = fake::synced_rows_of(&row.handle, true);
(FakeSource::new().with_synced(&row.handle, rows), row)
}
/// `atuin ai resume <id>` catches the session's copy here up with sync first, saying what it
/// wrote. A choice to make fails, saying why and naming the flags that make it; `--as-is`
/// resumes the copy as it is, and `--branch` names no branch the session hasn't got.
#[rstest]
#[tokio::test]
async fn resuming_by_id_catches_up_or_says_what_to_choose() {
use crate::resume_tui::catchup::{Why, branches};
let (source, row) = branched().await;
let plan = FakeResumer::default().plan(&row).await.unwrap().plan;
let status = "caught up: 2 messages from @00000002".to_owned();
let ready = FakeResumer {
catch_up: Some(CatchUp::Ready {
plan: plan.clone(),
status: Some(status.clone()),
}),
..FakeResumer::default()
};
let here = fake::THIS_HOST_ID;
let got = direct_plan(&source, &ready, &row, false, None, here).await.unwrap();
assert_eq!(got, Ok((plan.clone(), Some(status))));
let analysis = source.analyse(&row.handle).await.unwrap().unwrap();
let copy = fake::local_tip(&["a", "b", "c", "d"], Some("d"));
let held = FakeResumer {
catch_up: Some(CatchUp::Choice(Box::new(Held {
why: Why::Live,
harness: HarnessKind::ClaudeCode,
plan: plan.clone(),
branches: branches(&analysis, Some(&copy), here),
chosen: 0,
}))),
..FakeResumer::default()
};
let err = direct_plan(&source, &held, &row, false, None, here).await.unwrap_err();
let err = err.to_string();
let id = row.handle.session.as_ref();
let want = format!(
"Claude Code is running this session here: `--as-is` resumes this copy as it is, \
`atuin ai resume {id} --fork` forks it from this machine's branch; `--branch` picks \
another:\n y this machine · "
);
assert!(err.starts_with(&want), "{err}");
assert!(err.contains("\n d @00000002 · "), "{err}");
let got = direct_plan(&source, &held, &row, true, None, here).await.unwrap();
assert_eq!(got, Ok((plan, None)), "as it is");
let err = direct_plan(&source, &held, &row, false, Some("nope"), here).await.unwrap_err();
assert!(err.to_string().starts_with("--branch: no branch is \"nope\""), "{err}");
}
/// `--fork` forks a session that went on separately on several machines from its newest
/// branch, or the one `--branch` names, from the rows on it.
#[rstest]
#[case::the_newest(None, &["a", "b", "x", "y"])]
#[case::by_host(Some("@00000002"), &["a", "b", "c", "d"])]
#[case::by_head(Some("d"), &["a", "b", "c", "d"])]
#[tokio::test]
async fn forking_by_id_forks_from_a_branch(
#[case] branch: Option<&str>,
#[case] want: &[&str],
) {
let (source, _) = branched().await;
let resumer = FakeResumer::default();
let here = fake::THIS_HOST_ID;
fork_plan(&source, &resumer, "7f3c9a12", branch, here).await.unwrap();
let rows = resumer.forks.lock()[0].rows.clone().unwrap();
let ids: Vec<&str> = rows.iter().map(|r| r.source_id.as_str()).collect();
assert_eq!(ids, want);
}
/// `--as-is` and `--branch` go with resuming (and `--branch` with `--fork`), not with each
/// other or `--in`; `--print` still prints.
#[rstest]
#[case::as_is(&["--as-is", "--print"], true)]
#[case::branch(&["--branch", "@00000002"], true)]
#[case::fork_from_a_branch(&["--fork", "--branch", "d", "--print"], true)]
#[case::as_is_and_fork(&["--as-is", "--fork"], false)]
#[case::as_is_and_branch(&["--as-is", "--branch", "d"], false)]
#[case::branch_and_in(&["--branch", "d", "--in", "pi"], false)]
fn catch_up_flags(#[case] flags: &[&str], #[case] ok: bool) {
let args = ["resume", "7f3c9a12"].iter().chain(flags);
let cli = Cli::try_parse_from(args);
assert_eq!(cli.is_ok(), ok, "{flags:?}");
if let Ok(cli) = cli {
assert_eq!(cli.cmd.print, flags.contains(&"--print"));
}
}
#[rstest]
fn parses_flags() {
let cli = Cli::try_parse_from([
+348
View File
@@ -0,0 +1,348 @@
//! Catching this machine's copy of a session up with sync before resuming it in its own agent.
//!
//! A session keeps one id on every machine: each machine's transcript is a line of it, and the
//! synced rows hold every line ([`Analysis`]). Resuming a session here brings its copy to a head
//! (the newest, unless `--branch` picked another), and never merges:
//!
//! - no copy here: it is restored from sync ([`Resumer::restore`]); along the head's branch
//! ([`branch_rows`]) when one was named or the session diverged, else from every row, as before;
//! - the copy is at the head, or went on past it ([`Step::AsIs`], [`Step::Ahead`]): resumed as it
//! is;
//! - the copy is behind on the head's line ([`Step::FastForward`]), and no agent here has it open:
//! the rows it lacks are appended, and it resumes in place;
//! - anything else ([`Step::Choice`]): nothing is written, and the user chooses between resuming
//! the copy as it is and forking from a head ([`Held`]).
//!
//! [`Resumer::restore`]: super::resumer::Resumer::restore
use atuin_client::ai_session::{
Analysis, FastForward, HarnessKind, HarnessSession, Head, Message, SourceId,
};
use atuin_common::harnesstools::rehydrate::RehydrateMessage;
use atuin_common::harnesstools::session::{Content, is_substantive};
use atuin_common::harnesstools::sync::LocalTip;
use time::OffsetDateTime;
use super::clock::When;
use super::resumer::{ForkFrom, NotResumable, ResumePlan};
use super::source::{SessionSource, harness_label};
/// How bringing this machine's copy of a session to a head goes ([`classify`]).
#[derive(Debug, Clone)]
pub enum Step {
/// The copy is at the head.
AsIs,
/// The copy holds the head and went on past it here.
Ahead,
/// The copy is behind on the head's line: append `rows` to it, read as `base`.
FastForward {
rows: Vec<RehydrateMessage>,
base: LocalTip,
},
/// Nothing can be written: resuming the copy as it is, or forking, is the user's choice.
Choice(Why),
}
/// Why catching up needs the user to choose.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Why {
/// The copy is on another branch than the head's.
Diverged,
/// The copy goes on from a row sync hasn't got.
Unsynced,
/// An agent here has the session open, or may have.
Live,
/// The agent's copy couldn't take the rows (see [`SyncError`]).
///
/// [`SyncError`]: atuin_common::harnesstools::sync::SyncError
Refused(String),
}
/// How to bring the copy read as `tip` to `head`, one of `analysis`'s heads, for a session of
/// `harness`.
///
/// The copy is at the head, or ahead of it, only when the head is on the line its tip goes on
/// from: a copy holding the head that was rewound onto a line of its own is not. A tip sync
/// hasn't got may be either, which can't be told: that is a choice ([`Why::Unsynced`]).
///
/// The twig rule: a copy that stopped on a twig off the head's line (a tool result, an attachment:
/// nothing [substantive](is_substantive) past where it leaves the line, and no summary standing in
/// for the conversation, which the model is given as context) is no branch of its own, and is
/// fast-forwarded from where it leaves it, as the agent would go on from there. Only for a harness
/// whose rows name the row they follow (Claude Code, pi), where the rows appended hang from that
/// row; any other reads its transcript in order, twig and all.
pub fn classify(
analysis: &Analysis,
harness: HarnessKind,
head: &SourceId,
tip: &LocalTip,
) -> Step {
match analysis.fast_forward(head, tip) {
FastForward::NotBehind => return Step::AsIs,
FastForward::Behind(rows) => {
return Step::FastForward {
rows,
base: tip.clone(),
};
}
FastForward::Elsewhere => {}
}
let Some(at) = tip.tip_source_id.as_deref() else {
return Step::Choice(Why::Diverged);
};
let mine = analysis.path_to(&SourceId::from(at.to_owned()));
if mine.is_empty() {
// Sync hasn't got the row the copy goes on from: past the head, or a line of its own
// (rewound here, even if it holds the head), which can't be told apart.
return Step::Choice(Why::Unsynced);
}
if mine.iter().any(|m| m.source_id == *head) {
return Step::Ahead;
}
let path = analysis.path_to(head);
let shared = path.iter().zip(&mine).take_while(|(a, b)| a.source_id == b.source_id).count();
let linked = matches!(harness, HarnessKind::ClaudeCode | HarnessKind::Pi);
let twig = linked && !mine[shared..].iter().any(|m| carries_conversation(m));
let rest = &path[shared..];
let holds_none = !rest.iter().any(|m| tip.known_source_ids.contains(m.source_id.as_ref()));
if twig && holds_none && shared > 0 {
let mut base = tip.clone();
base.tip_source_id = Some(path[shared - 1].source_id.to_string());
let rows = rest.iter().map(|&m| m.clone().into()).collect();
return Step::FastForward { rows, base };
}
Step::Choice(Why::Diverged)
}
/// Whether going on from before `row` would lose conversation: it is
/// [substantive](is_substantive), or a summary standing in for earlier conversation (a
/// compaction), which the model is given as context. Only for the twig rule: a summary makes no
/// branch of its own as far as [`Analysis`] goes (see [`is_substantive`]).
fn carries_conversation(row: &Message) -> bool {
is_substantive(&row.role, &row.content)
|| row.content.iter().any(|c| matches!(c, Content::Summary(_)))
}
/// How many messages `rows` hold: prompts and assistant text, as a [`Head`] counts them.
pub fn message_count(rows: &[RehydrateMessage]) -> usize {
rows.iter().filter(|m| is_substantive(&m.role, &m.content)).count()
}
/// A head to fork from, as the chooser offers it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Branch {
pub head: Head,
/// Its host, as [`host_label`] names it.
pub host: String,
/// What `--branch` names it by ([`branch_selector`]).
pub selector: String,
/// Its messages this machine's copy hasn't got: since they split.
pub ahead: usize,
}
impl Branch {
/// The chooser's line: `fork @3f9a12bc's · +11 since they split · 20m ago`.
pub fn line(&self, now: OffsetDateTime) -> String {
let when = When::of(now, self.head.last_at, time::UtcOffset::UTC).phrase();
match self.ahead {
0 => format!("fork {}'s · {when}", self.host),
n => format!("fork {}'s · +{n} since they split · {when}", self.host),
}
}
}
/// `analysis`'s heads, newest first, as [`Branch`]es against the copy read as `tip`; `here` is
/// this host's id.
pub fn branches(analysis: &Analysis, tip: Option<&LocalTip>, here: &str) -> Vec<Branch> {
let heads = analysis.heads();
heads
.iter()
.map(|head| {
let path = analysis.path_to(&head.source_id);
let ahead = path
.iter()
.filter(|m| is_substantive(&m.role, &m.content))
.filter(|m| tip.is_none_or(|t| !t.known_source_ids.contains(m.source_id.as_ref())))
.count();
Branch {
head: head.clone(),
host: host_label(head, here),
selector: branch_selector(heads, head),
ahead,
}
})
.collect()
}
/// Catching up needs a choice: nothing was written.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Held {
pub why: Why,
pub harness: HarnessKind,
/// Resumes this machine's copy as it is.
pub plan: ResumePlan,
/// The session's heads to fork from, newest first.
pub branches: Vec<Branch>,
/// The one catching up went to, in `branches`.
pub chosen: usize,
}
impl Held {
/// The status line.
pub fn status(&self) -> String {
match &self.why {
Why::Live => format!("{} is running this session here", harness_label(self.harness)),
Why::Unsynced => "this copy has messages sync hasn't got".to_owned(),
Why::Diverged => match self.branches.get(self.chosen) {
Some(branch) => format!("this copy went another way than {}'s", branch.host),
None => "this copy went another way".to_owned(),
},
Why::Refused(why) => format!("couldn't catch up: {why}"),
}
}
}
/// What catching up did ([`super::resumer::Resumer::catch_up`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CatchUp {
/// Resume with `plan`; `status` says what was written, if anything.
Ready {
plan: ResumePlan,
status: Option<String>,
},
Choice(Box<Held>),
}
/// `caught up: 2 messages from @3f9a12bc`.
pub fn caught_up(messages: usize, host: &str) -> String {
let s = if messages == 1 {
""
} else {
"s"
};
format!("caught up: {messages} message{s} from {host}")
}
/// How a head's host is named when it is this one.
pub const THIS_MACHINE: &str = "this machine";
/// A head's host as the picker names it: `this machine`, or `@3f9a12bc` for another, by a short
/// form of its id (see [`super::source::host_label`]).
pub fn host_label(head: &Head, here: &str) -> String {
// Rows from before hosts were recorded are this host's.
head.host.map_or_else(
|| THIS_MACHINE.to_owned(),
|host| super::source::host_label(&host.0.as_simple().to_string(), here),
)
}
/// The fewest characters a [`branch_selector`] has, so it stays unique as a session gains heads.
const SELECTOR_MIN: usize = 8;
/// How `--branch` names `head` among `heads` for good: the shortest start of its source id,
/// [`SELECTOR_MIN`] characters or more, that no other head's id starts with; else the whole id.
pub fn branch_selector(heads: &[Head], head: &Head) -> String {
let id = head.source_id.as_ref();
let others: Vec<&str> =
heads.iter().map(|h| h.source_id.as_ref()).filter(|other| *other != id).collect();
let ends = id.char_indices().map(|(at, _)| at).skip(SELECTOR_MIN).chain([id.len()]);
for end in ends {
let prefix = &id[..end];
if others.iter().all(|other| !other.starts_with(prefix)) {
return prefix.to_owned();
}
}
id.to_owned()
}
/// A head in a line: `@3f9a12bc · 2h · 136 msgs`.
pub fn describe(head: &Head, now: OffsetDateTime, here: &str) -> String {
let when = When::of(now, head.last_at, time::UtcOffset::UTC);
format!("{} · {} · {} msgs", host_label(head, here), when.short(), head.messages)
}
/// The head `selector` names among `heads` (`--branch`): `this` for this machine's, `@3f9a12bc`
/// for another host's (by the end of its id, or all of it), else the start of its source id (see
/// [`branch_selector`]). Anything naming no single head is an error listing them.
pub fn pick_branch<'a>(
heads: &'a [Head],
selector: &str,
now: OffsetDateTime,
here: &str,
) -> Result<&'a Head, String> {
if heads.is_empty() {
return Err("its branches aren't known yet: the daemon hasn't synced its messages".into());
}
let selector = selector.trim();
let host = |h: &Head| host_label(h, here);
let found: Vec<&Head> = if selector == "this" {
heads.iter().filter(|h| host(h) == THIS_MACHINE).collect()
} else if let Some(id) = selector.strip_prefix('@') {
let short = format!("@{}", super::source::short_host_id(&super::simple_host_id(id)));
heads.iter().filter(|h| host(h).eq_ignore_ascii_case(&short)).collect()
} else if let Some(exact) = heads.iter().find(|h| h.source_id.as_ref() == selector) {
vec![exact]
} else if selector.is_empty() {
Vec::new()
} else {
heads.iter().filter(|h| h.source_id.as_ref().starts_with(selector)).collect()
};
let listing = || {
let lines = heads.iter().enumerate().map(|(n, h)| {
let latest = if n == 0 {
" (latest)"
} else {
""
};
format!("\n {} {}{latest}", branch_selector(heads, h), describe(h, now, here))
});
lines.collect::<String>()
};
match found.as_slice() {
[head] => Ok(head),
[] => Err(format!("no branch is {selector:?}; its branches:{}", listing())),
several => Err(format!(
"{selector:?} names {} branches; name one by id:{}",
several.len(),
listing()
)),
}
}
/// What forking `session` from `head` starts from: the rows of that head's branch
/// ([`branch_rows`]). Without one, a session that went on separately on several machines forks
/// from its newest head, and any other from every row synced of it ([`ForkFrom::default`]).
pub async fn fork_from(
source: &dyn SessionSource,
session: &HarnessSession,
head: Option<&SourceId>,
) -> Result<ForkFrom, NotResumable> {
let rows = branch_rows(source, session, head)
.await
.map_err(|e| NotResumable::Fork(format!("{e:#}")))?;
Ok(ForkFrom { rows, tip: None })
}
/// The rows a transcript of `session` written out along `head` holds (see
/// [`Analysis::rows_for`]): a head named (`--branch`, or a fork line picked) is followed; without
/// one, a session that went on separately on several machines follows its newest head, so no
/// branch's messages land in another's. `None` for any other session (or one whose branches
/// aren't known): it is written out from every row synced of it, as it always was, rows off every
/// branch included.
pub async fn branch_rows(
source: &dyn SessionSource,
session: &HarnessSession,
head: Option<&SourceId>,
) -> eyre::Result<Option<Vec<RehydrateMessage>>> {
let Some(analysis) = source.analyse(session).await? else {
return Ok(None);
};
let newest = analysis.diverged().then(|| analysis.heads().first()).flatten();
let Some(head) = head.or(newest.map(|h| &h.source_id)) else {
return Ok(None);
};
Ok(Some(analysis.rows_for(head).into_iter().map(|m| m.clone().into()).collect()))
}
#[cfg(test)]
mod tests;
@@ -0,0 +1,217 @@
use atuin_client::ai_session::Message;
use atuin_common::harnesstools::session::{Content, Role};
use atuin_domain::record::HostId;
use rstest::rstest;
use time::Duration;
use super::*;
use crate::resume_tui::fake::{
self, FakeSource, local_tip as tip, synced_handle as handle, synced_row, synced_rows as rows,
};
async fn analysis(rows: Vec<Message>) -> Analysis {
let source = FakeSource::from_rows(Vec::new()).with_synced(&handle(), rows);
source.analyse(&handle()).await.unwrap().unwrap()
}
/// A step, as the rows it appends (`ff:c,d@b`, from where) or its kind.
fn summary(step: &Step) -> String {
match step {
Step::AsIs => "as is".to_owned(),
Step::Ahead => "ahead".to_owned(),
Step::FastForward { rows, base } => {
let ids: Vec<&str> = rows.iter().map(|r| r.source_id.as_str()).collect();
format!("ff:{}@{}", ids.join(","), base.tip_source_id.as_deref().unwrap_or("-"))
}
Step::Choice(why) => format!("{why:?}"),
}
}
#[rstest]
#[case::at_the_head(false, &["a", "b", "c", "d"], "d", "as is")]
#[case::behind(false, &["a", "b"], "b", "ff:c,d@b")]
#[case::past_the_head_synced(false, &["a", "b", "c", "d", "e"], "e", "ahead")]
#[case::past_the_head_unsynced(false, &["a", "b", "c", "d", "z"], "z", "Unsynced")]
#[case::rewound_here_past_the_head(false, &["a", "b", "c", "d", "r"], "r", "Diverged")]
#[case::a_line_of_its_own_unsynced(false, &["a", "b", "z"], "z", "Unsynced")]
#[case::on_another_branch(true, &["a", "b", "x", "y"], "y", "Diverged")]
#[case::behind_on_the_other_branch(true, &["a", "b", "x"], "x", "Diverged")]
#[tokio::test]
async fn a_copy_is_classified_against_the_head(
#[case] diverged: bool,
#[case] known: &[&str],
#[case] at: &str,
#[case] want: &str,
) {
let mut rows = rows(diverged);
// Synced rows of this host's: one past the head, and a prompt rewound to before it.
rows.push(synced_row(&handle(), ("e", Some("d"), 1, 6, false)));
rows.push(synced_row(&handle(), ("r", Some("b"), 1, 7, false)));
let analysis = analysis(rows).await;
let head = SourceId::from("d".to_owned());
let step = classify(&analysis, HarnessKind::ClaudeCode, &head, &tip(known, Some(at)));
assert_eq!(summary(&step), want);
}
/// A copy that stopped on a tool result off the head's line goes on from where it leaves it, when
/// the rows appended hang from the row they follow; one that stopped on a compaction's summary
/// would lose it, and is a branch of its own. A harness that reads its transcript in order would
/// keep the twig: a choice too.
#[rstest]
#[case::a_tool_result(HarnessKind::ClaudeCode, Role::Tool, None, "ff:c,d@b")]
#[case::a_tool_result_in_pi(HarnessKind::Pi, Role::Tool, None, "ff:c,d@b")]
#[case::a_summary(HarnessKind::Pi, Role::System, Some("what went before"), "Diverged")]
#[case::in_order(HarnessKind::Codex, Role::Tool, None, "Diverged")]
#[tokio::test]
async fn a_twig_off_the_heads_line_fast_forwards(
#[case] harness: HarnessKind,
#[case] role: Role,
#[case] summary_text: Option<&str>,
#[case] want: &str,
) {
let mut rows = rows(false);
let mut twig = synced_row(&handle(), ("t", Some("b"), 1, 1, false));
twig.role = role;
twig.content = summary_text.map(|t| Content::Summary(t.to_owned())).into_iter().collect();
rows.push(twig);
let analysis = analysis(rows).await;
let head = SourceId::from("d".to_owned());
let step = classify(&analysis, harness, &head, &tip(&["a", "b", "t"], Some("t")));
assert_eq!(summary(&step), want);
}
/// Each head is offered with what this copy lacks of it, and when it was last written.
#[rstest]
#[tokio::test]
async fn branches_say_what_they_add_and_when() {
let analysis = analysis(rows(true)).await;
let copy = tip(&["a", "b", "x", "y"], Some("y"));
let branches = branches(&analysis, Some(&copy), fake::THIS_HOST_ID);
let lines: Vec<String> = branches.iter().map(|b| b.line(fake::now())).collect();
assert_eq!(lines, [
"fork this machine's · 55m ago",
"fork @00000002's · +2 since they split · 57m ago",
]);
assert_eq!((branches[0].selector.as_str(), branches[1].selector.as_str()), ("y", "d"));
}
#[rstest]
#[case::live(Why::Live, "Claude Code is running this session here")]
#[case::unsynced(Why::Unsynced, "this copy has messages sync hasn't got")]
#[case::diverged(Why::Diverged, "this copy went another way than @00000002's")]
#[case::refused(Why::Refused("it changed".into()), "couldn't catch up: it changed")]
#[tokio::test]
async fn a_choice_says_why(#[case] why: Why, #[case] want: &str) {
let analysis = analysis(rows(false)).await;
let held = Held {
why,
harness: HarnessKind::ClaudeCode,
plan: ResumePlan {
program: "claude".to_owned(),
args: Vec::new(),
cwd: None,
cwd_requirement: atuin_common::harnesstools::resume::CwdRequirement::Preferred,
native_path: None,
},
branches: branches(&analysis, None, fake::THIS_HOST_ID),
chosen: 0,
};
assert_eq!(held.status(), want);
assert_eq!(caught_up(1, "@00000002"), "caught up: 1 message from @00000002");
assert_eq!(caught_up(3, "@00000002"), "caught up: 3 messages from @00000002");
}
/// Forking from a head starts from the rows root to it; without one, a diverged session forks
/// from its newest head, and one that went one way from every row synced.
#[rstest]
#[case::a_head(true, Some("d"), Some(&["a", "b", "c", "d"][..]))]
#[case::the_newest(true, None, Some(&["a", "b", "x", "y"][..]))]
#[case::one_way(false, None, None)]
#[tokio::test]
async fn forks_start_from_a_heads_rows(
#[case] diverged: bool,
#[case] head: Option<&str>,
#[case] want: Option<&[&str]>,
) {
let source = FakeSource::from_rows(Vec::new()).with_synced(&handle(), rows(diverged));
let head = head.map(|h| SourceId::from(h.to_owned()));
let from = fork_from(&source, &handle(), head.as_ref()).await.unwrap();
let ids: Option<Vec<&str>> =
from.rows.as_ref().map(|rows| rows.iter().map(|r| r.source_id.as_str()).collect());
assert_eq!(ids.as_deref(), want);
}
/// A fork from a head brings the rows off the tree that go with its branch: a prompt and reply
/// from before the session had ids (a Pi session from before them), and another host's reply on
/// its own branch only.
#[rstest]
#[case::a_head(Some("d"), &["syn-p", "syn-r", "a", "b", "c", "d", "syn-d"])]
#[case::the_newest(None, &["syn-p", "syn-r", "a", "b", "x", "y"])]
#[tokio::test]
async fn forks_bring_the_rows_off_the_tree_on_their_branch(
#[case] head: Option<&str>,
#[case] want: &[&str],
) {
let mut synced = rows(true);
synced.extend([
synced_row(&handle(), ("syn-p", None, 1, -2, false)),
synced_row(&handle(), ("syn-r", None, 1, -1, true)),
synced_row(&handle(), ("syn-d", None, 2, 4, true)),
]);
let source = FakeSource::from_rows(Vec::new()).with_synced(&handle(), synced);
let head = head.map(|h| SourceId::from(h.to_owned()));
let from = fork_from(&source, &handle(), head.as_ref()).await.unwrap();
let ids: Vec<&str> = from.rows.as_ref().unwrap().iter().map(|r| r.source_id.as_str()).collect();
assert_eq!(ids, want);
}
fn head(id: &str, host: u128, minutes_ago: i64, messages: u64) -> Head {
Head {
source_id: id.to_owned().into(),
host: Some(HostId(uuid::Uuid::from_u128(0x0190_0000_0000_7000_8000_0000_0000_0000 | host))),
last_at: fake::now() - Duration::minutes(minutes_ago),
messages,
}
}
/// A branch's selector is the shortest start of its id that no other branch shares, eight
/// characters at least, or the whole id; and it picks that branch back.
#[rstest]
#[case::uuid(&["7aaabc31-1631-4756", "0c1d2e3f-8a9b-4c5d"], "7aaabc31")]
#[case::sharing_a_long_start(&["prt_fcf1e6fc6001aa", "prt_fcf1e6fc9002bb"], "prt_fcf1e6fc6")]
#[case::a_start_of_another(&["b4", "b40"], "b4")]
fn a_selector_is_the_shortest_start_no_other_branch_shares(
#[case] ids: &[&str],
#[case] want: &str,
) {
let heads: Vec<Head> = ids.iter().map(|id| head(id, 1, 0, 1)).collect();
let selector = branch_selector(&heads, &heads[0]);
assert_eq!(selector, want);
let picked = pick_branch(&heads, &selector, fake::now(), fake::THIS_HOST_ID).unwrap();
assert_eq!(picked, &heads[0]);
}
/// `--branch` takes `this`, `@host` (its short or full id) or the start of a head's id; anything
/// naming no single head is an error listing them.
#[rstest]
#[case::this("this", Ok("h1"))]
#[case::host("@00000002", Ok("h2"))]
#[case::full_host_id("@01900000-0000-7000-8000-000000000002", Ok("h2"))]
#[case::id("h3", Ok("h3"))]
#[case::unknown_host("@buildbox", Err("no branch is \"@buildbox\"; its branches:"))]
#[case::ambiguous_id("h", Err("\"h\" names 3 branches; name one by id:"))]
fn branches_are_picked_by_host_or_id(#[case] selector: &str, #[case] want: Result<&str, &str>) {
let heads = [head("h2", 2, 60, 40), head("h1", 1, 120, 24), head("h3", 3, 180, 5)];
let picked = pick_branch(&heads, selector, fake::now(), fake::THIS_HOST_ID);
match want {
Ok(id) => assert_eq!(picked.unwrap().source_id.as_ref(), id),
Err(start) => {
let why = picked.unwrap_err();
assert!(why.starts_with(start), "{why}");
let listing = "\n h2 @00000002 · 1h · 40 msgs (latest)\n h1 this machine · 2h · \
24 msgs\n h3 @00000003 · 3h · 5 msgs";
assert!(why.ends_with(listing), "{why}");
}
}
assert!(pick_branch(&[], "this", fake::now(), fake::THIS_HOST_ID).is_err());
}
+101 -19
View File
@@ -6,7 +6,7 @@
//! hints at. Right under it, when its harness is installed and atuin can write its sessions,
//! forking it: a new session of the same harness, with the same history (see
//! [`atuin_common::harnesstools::fork`]); `f` selects it, and it is preselected when the original
//! can't resume here or looks to be running here still. It is dimmed, and can't be picked, once
//! can't resume here or an agent here has it open. It is dimmed, and can't be picked, once
//! the session turns out to hold no conversation to fork. Then every other harness installed
//! here, to continue the session in: it is written out as a new session there, its tool calls
//! flattened into notes, and that is resumed. Harnesses
@@ -15,6 +15,11 @@
//! session, a directory that's gone, a harness that isn't installed), and the first line that
//! works is preselected instead.
//!
//! When catching the session's copy here up with sync needs a choice ([`Held`]), the chooser opens
//! again on it, saying why: its first line resumes the copy here as it is, and a fork line for each
//! of the session's heads, newest first, forks from that head; the newest is preselected when an
//! agent here has the session open.
//!
//! The chooser keeps the key's meaning: opened with enter it resumes (or edits, without
//! `enter_accept`), opened with tab it edits, and tab in it always edits.
@@ -34,8 +39,9 @@ use ratatui::widgets::{Block, BorderType, Borders, Clear, Paragraph};
use time::OffsetDateTime;
use unicode_width::UnicodeWidthStr;
use super::catchup::{Branch, Held, Why};
use super::render::{harness_style, style};
use super::resumer::NotResumable;
use super::resumer::{NotResumable, ResumePlan};
use super::source::{SessionRow, harness_badge, harness_label};
use super::state::{InputAction, LIVE_SECS, Pending, Picked, State};
@@ -56,31 +62,49 @@ pub struct Chooser {
/// Whether the selection was moved by hand: if not, it moves off the first line once that
/// turns out not to work.
pub moved: bool,
/// Set when it opened on catching up needing a choice: the first line resumes the copy here
/// as it is, and the fork lines fork from the session's heads.
pub held: Option<Box<Held>>,
}
/// What a line of the chooser does with the session.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Destination {
/// Resume it in its own harness.
Original,
/// Fork it: a new session of its own harness, with the same history.
Fork,
/// Resume the copy here as it is, with this plan, rather than catch it up.
AsIs(Box<ResumePlan>),
/// Fork it: a new session of its own harness, with the same history (that of a head, when
/// given).
Fork(Option<Box<Branch>>),
/// Continue it in another harness.
Continue(HarnessKind),
}
impl Chooser {
/// How many fork lines it has.
fn forks(&self) -> usize {
match &self.held {
Some(held) if self.fork => held.branches.len(),
_ => usize::from(self.fork),
}
}
pub fn len(&self) -> usize {
1 + usize::from(self.fork) + self.targets.len()
1 + self.forks() + self.targets.len()
}
/// What line `n` does.
pub fn line(&self, n: usize) -> Destination {
let first = 1 + usize::from(self.fork);
match n {
0 => Destination::Original,
1 if self.fork => Destination::Fork,
n => self
let first = 1 + self.forks();
match (n, &self.held) {
(0, None) => Destination::Original,
(0, Some(held)) => Destination::AsIs(Box::new(held.plan.clone())),
(n, None) if n < first => Destination::Fork(None),
(n, Some(held)) if n < first => {
Destination::Fork(held.branches.get(n - 1).cloned().map(Box::new))
}
(n, _) => self
.targets
.get(n - first)
.map_or(Destination::Original, |t| Destination::Continue(*t)),
@@ -147,6 +171,31 @@ impl State {
selected: 0,
action,
moved: false,
held: None,
});
self.settle_chooser();
}
/// Open the chooser on `row` for the choice catching it up needs (`held`): resume the copy
/// here as it is, fork from a head (with `fork`), or continue it in one of `targets`. The
/// status line says why.
pub fn open_choice(
&mut self,
row: &SessionRow,
held: Box<Held>,
targets: Vec<HarnessKind>,
fork: bool,
action: Pending,
) {
self.status = Some((held.status(), Meaning::AlertWarn));
self.chooser = Some(Chooser {
row: row.clone(),
targets,
fork: fork && !held.branches.is_empty(),
selected: 0,
action,
moved: false,
held: Some(held),
});
self.settle_chooser();
}
@@ -163,9 +212,12 @@ impl State {
}
/// Move an untouched selection off the first line once its plan says it can't be picked,
/// or onto the fork when the session looks to be running here still (it was written to
/// lately, itself rather than a child: see [`running`]), which resuming it again would race.
/// Never onto a fork with nothing to fork: off it again, if it was moved there already.
/// or onto the fork when an agent here has the session open ([`Resume::live`]), which
/// resuming it again would race. Never onto a fork with nothing to fork: off it again, if it
/// was moved there already. For a choice catching up needs, the copy here as it is, or the
/// fork from the newest head while an agent here has the session open.
///
/// [`Resume::live`]: super::resumer::Resume::live
pub fn settle_chooser(&mut self) {
let Some(chooser) = &self.chooser else {
return;
@@ -175,10 +227,15 @@ impl State {
}
let row = &chooser.row;
let fork = chooser.fork && !self.nothing_to_fork(&row.handle);
let running = fork && row.host_id == self.context.host_id && {
let row = self.target().filter(|r| r.handle == row.handle).unwrap_or(row);
running((self.now)(), row)
};
if let Some(held) = &chooser.held {
let selected = usize::from(fork && held.why == Why::Live);
if let Some(chooser) = self.chooser.as_mut() {
chooser.selected = selected;
}
return;
}
let plan = self.plans.get(&row.handle).and_then(|p| p.as_ref().ok());
let running = fork && plan.is_some_and(|p| p.live);
let selected = if !running && self.original_unavailable(&row.handle).is_none() {
0
} else if fork || !chooser.fork {
@@ -210,6 +267,9 @@ impl State {
// As the list last read it, while it still shows it.
let row = self.target().filter(|r| r.handle == *session).unwrap_or(&chooser.row);
let live = running((self.now)(), row);
if let Some(held) = &chooser.held {
return self.held_choices(chooser, held);
}
let original = match self.plans.get(session) {
Some(Err(why)) => Choice {
harness: session.harness,
@@ -266,6 +326,28 @@ impl State {
std::iter::once(original).chain(fork).chain(continued).collect()
}
/// The chooser's lines for a choice catching up needs.
fn held_choices(&self, chooser: &Chooser, held: &Held) -> Vec<Choice> {
let harness = chooser.row.handle.harness;
let line = |detail: String| Choice {
harness,
detail,
unavailable: None,
};
let detail = match held.why {
Why::Live => "this copy as is · running here",
_ => "this copy as is",
};
let now = (self.now)();
let forks = held.branches.iter().take(chooser.forks()).map(|b| line(b.line(now)));
let continued = chooser.targets.iter().map(|target| Choice {
harness: *target,
detail: "continue".to_owned(),
unavailable: None,
});
std::iter::once(line(detail.to_owned())).chain(forks).chain(continued).collect()
}
/// A key while the chooser is open: move (up/down, ctrl-p/ctrl-n, k/j, f to the fork), pick
/// (enter does what opened the chooser, tab edits, ctrl-y copies the command, a digit picks
/// its line), or go back to the list (esc, q, ctrl-c, ctrl-g).
@@ -324,7 +406,7 @@ impl State {
));
return InputAction::Continue;
}
if line == Destination::Fork && self.nothing_to_fork(&row.handle) {
if matches!(line, Destination::Fork(_)) && self.nothing_to_fork(&row.handle) {
// Its line already says why; the `--fork` command copied would fail as well.
self.status =
Some(("nothing to fork: pick another line".to_owned(), Meaning::AlertError));
+146 -11
View File
@@ -4,19 +4,26 @@
//! It covers all four harnesses, forks and subagents grouped under their roots, sessions from other
//! hosts, a deleted worktree (missing cwd), and live sessions (updated in the last two minutes).
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use async_trait::async_trait;
use atuin_client::ai_session::{HarnessKind, HarnessSession, NativeSessionId};
use atuin_client::ai_session::{
AiSessionDatabase, Analysis, HarnessKind, HarnessSession, Message, NativeSessionId, SourceId,
};
use atuin_common::harnesstools::Harness as _;
use atuin_common::harnesstools::continuation::Flattened;
use atuin_common::harnesstools::rehydrate::RehydrateSession;
use atuin_common::harnesstools::rehydrate::{RehydrateMessage, RehydrateSession};
use atuin_common::harnesstools::resume::CwdRequirement;
use atuin_common::harnesstools::session::Usage;
use atuin_common::harnesstools::session::{Content, Role as Said, Usage};
use atuin_common::harnesstools::sync::{LocalTip, Stamp};
use atuin_domain::record::{HostId, RecordId};
use parking_lot::Mutex;
use time::{Duration, OffsetDateTime};
use super::ResumeContext;
use super::catchup::CatchUp;
use super::resumer::{
Continued, ForkFrom, Forked, NotResumable, Restore, Resume, ResumeError, ResumePlan,
ResumeTarget, Resumer,
@@ -90,6 +97,76 @@ struct FakeSession {
pub struct FakeSource {
sessions: Vec<FakeSession>,
/// The rows sync holds of a session, for its branches ([`SessionSource::analyse`]).
synced: HashMap<HarnessSession, Vec<Message>>,
}
/// The session [`synced_rows`] are of.
pub fn synced_handle() -> HarnessSession {
HarnessSession {
harness: HarnessKind::ClaudeCode,
session: NativeSessionId::from("s".to_owned()),
}
}
/// The rows sync holds of a Claude Code session: `a`-`b` on this host, then `c`-`d` on another;
/// `diverged`, this host went on from `b` with `x`-`y` too.
pub fn synced_rows(diverged: bool) -> Vec<Message> {
synced_rows_of(&synced_handle(), diverged)
}
/// [`synced_rows`], of session `handle`.
pub fn synced_rows_of(handle: &HarnessSession, diverged: bool) -> Vec<Message> {
let mut specs = vec![
("a", None, 1, 0, false),
("b", Some("a"), 1, 1, true),
("c", Some("b"), 2, 2, false),
("d", Some("c"), 2, 3, true),
];
if diverged {
specs.extend([("x", Some("b"), 1, 4, false), ("y", Some("x"), 1, 5, true)]);
}
specs.into_iter().map(|spec| synced_row(handle, spec)).collect()
}
/// A copy holding `known`, going on from `at`.
pub fn local_tip(known: &[&str], at: Option<&str>) -> LocalTip {
LocalTip {
native_path: PathBuf::from("/t/s.jsonl"),
known_source_ids: known.iter().map(|s| (*s).to_owned()).collect::<HashSet<_>>(),
merged: HashMap::new(),
tip_source_id: at.map(str::to_owned),
stamp: Stamp::of(b""),
modified: None,
cwd: None,
}
}
/// The hosts' ids, but for their last digit.
const HOST_BASE: u128 = 0x0190_0000_0000_7000_8000_0000_0000_0000;
/// A row synced of session `handle`: `id`, hanging from `parent`, captured on host `host` (its
/// id's last digit: 1 is this host) `minutes` after the hour, a prompt or (`reply`) a reply.
pub fn synced_row(
handle: &HarnessSession,
(id, parent, host, minutes, reply): (&str, Option<&str>, u128, i64, bool),
) -> Message {
let role = if reply {
Said::Assistant
} else {
Said::User
};
let mut message = Message::builder()
.id(RecordId(atuin_common::utils::uuid_v7()))
.session(handle.clone())
.source_id(SourceId::from(id.to_owned()))
.parent_source_id(parent.map(|p| SourceId::from(p.to_owned())))
.timestamp(now() - Duration::hours(1) + Duration::minutes(minutes))
.role(role)
.content(vec![Content::Text(format!("message {id}"))])
.build();
message.host = Some(HostId(uuid::Uuid::from_u128(HOST_BASE | host)));
message
}
struct Spec {
@@ -668,7 +745,16 @@ impl FakeSource {
s.root = root;
}
Self { sessions }
Self {
sessions,
synced: HashMap::new(),
}
}
/// This source, with `rows` synced of session `handle`.
pub fn with_synced(mut self, handle: &HarnessSession, rows: Vec<Message>) -> Self {
self.synced.insert(handle.clone(), rows);
self
}
/// A source holding only `rows`.
@@ -684,9 +770,17 @@ impl FakeSource {
messages: Vec::new(),
})
.collect(),
synced: HashMap::new(),
}
}
/// Every row synced of `session`, in time order, as the sidecar restores them.
fn synced_messages(&self, session: &HarnessSession) -> Vec<RehydrateMessage> {
let mut rows = self.synced.get(session).cloned().unwrap_or_default();
rows.sort_by_key(|m| m.timestamp);
rows.into_iter().map(RehydrateMessage::from).collect()
}
fn children_of<'a>(
&'a self,
root: &'a HarnessSession,
@@ -846,7 +940,8 @@ impl SessionSource for FakeSource {
Ok(rows)
}
/// The session without its messages: the fake sessions have none.
/// The session, with every row synced of it as its messages (the fake sessions' own messages
/// are preview text only).
async fn rehydrate(
&self,
session: &HarnessSession,
@@ -866,10 +961,21 @@ impl SessionSource for FakeSource {
git_branch: row.branch.clone(),
model: row.model.clone(),
started_at: row.started_at,
messages: Vec::new(),
messages: self.synced_messages(session),
fork_of: None,
})
}
async fn analyse(&self, session: &HarnessSession) -> eyre::Result<Option<Analysis>> {
let Some(rows) = self.synced.get(session) else {
return Ok(None);
};
let db = AiSessionDatabase::in_memory().await?;
for row in rows {
db.append(row).await?;
}
Ok(Some(db.analyse(session).await?))
}
}
/// A [`Resumer`] for the fake sessions: the harnesses' real plans, with the deleted worktree as the
@@ -877,13 +983,19 @@ impl SessionSource for FakeSource {
/// Other hosts' sessions are restored from sync, into the directory they ran in; restoring
/// writes nothing.
pub struct FakeResumer {
missing: HashSet<PathBuf>,
pub missing: HashSet<PathBuf>,
/// What catching this host's sessions up does, when they are caught up at all.
pub catch_up: Option<CatchUp>,
/// What each fork started from.
pub forks: Arc<Mutex<Vec<ForkFrom>>>,
}
impl Default for FakeResumer {
fn default() -> Self {
Self {
missing: HashSet::from([PathBuf::from(DELETED_WORKTREE)]),
catch_up: None,
forks: Arc::default(),
}
}
}
@@ -917,18 +1029,40 @@ impl FakeResumer {
#[async_trait]
impl Resumer for FakeResumer {
/// This host's sessions are here, live while they were written to in the last two minutes;
/// other hosts' are restored.
async fn plan(&self, session: &SessionRow) -> Result<Resume, NotResumable> {
let plan = self.plan_for(session, None)?;
if session.host_id == THIS_HOST_ID {
return Ok(Resume::ready(plan));
let ago = now() - session.active_at;
return Ok(Resume {
catch_up: self.catch_up.is_some(),
live: ago < Duration::seconds(i64::try_from(super::state::LIVE_SECS).unwrap_or(0)),
..Resume::ready(plan)
});
}
let cwd = session.cwd.clone().unwrap_or_else(|| PathBuf::from(REPO));
Ok(Resume {
plan,
restore: Some(Restore { cwd, note: None }),
..Resume::ready(plan)
})
}
async fn catch_up(
&self,
_source: &dyn SessionSource,
session: &SessionRow,
_head: Option<&SourceId>,
) -> Result<CatchUp, NotResumable> {
match &self.catch_up {
Some(caught) if session.host_id == THIS_HOST_ID => Ok(caught.clone()),
_ => Ok(CatchUp::Ready {
plan: self.plan(session).await?.plan,
status: None,
}),
}
}
async fn restore(
&self,
_source: &dyn SessionSource,
@@ -987,8 +1121,9 @@ impl Resumer for FakeResumer {
&self,
_source: &dyn SessionSource,
session: &SessionRow,
_from: ForkFrom,
from: ForkFrom,
) -> Result<Forked, NotResumable> {
self.forks.lock().push(from);
let mut row = session.clone();
row.handle.session = NativeSessionId::from(format!("forked-{}", session.handle.session));
let native = PathBuf::from(format!("/forked/{}", row.handle.session));
+97 -14
View File
@@ -9,6 +9,7 @@
//! - [`SessionSource`] lists, searches and previews sessions;
//! - [`Resumer`] turns a session into a resume command, or says why it can't be resumed.
pub mod catchup;
pub mod chooser;
pub mod clock;
#[cfg(test)]
@@ -41,6 +42,7 @@ use ratatui::{Terminal, TerminalOptions, Viewport};
pub use resumer::{ResumePlan, Resumer};
pub use source::{SessionRow, SessionSource};
use self::catchup::{Branch, CatchUp};
use self::chooser::Destination;
use self::resumer::{Continued, Forked, NotResumable, Resume};
use self::state::{
@@ -282,8 +284,9 @@ fn apply_response(state: &mut State, response: Response, requests: &Requests) {
// A fork with nothing to fork isn't left selected.
state.settle_chooser();
}
// Handled by `finish_continuation` and `finish_fork`, which may end the picker.
Response::Continued(..) | Response::Forked(..) => {}
// Handled by `finish_continuation` and `finish_fork`, which may end the picker, and by
// `caught_up`, which needs the resumer.
Response::Continued(..) | Response::Forked(..) | Response::CaughtUp(..) => {}
}
}
@@ -296,7 +299,14 @@ fn respond(
resumer: &dyn Resumer,
requests: &Requests,
) -> Option<Outcome> {
apply_response(state, response, requests);
match response {
Response::CaughtUp(handle, result) => {
if caught_up(state, handle, result, resumer, requests) {
return None;
}
}
response => apply_response(state, response, requests),
}
let (row, pending) = state.pending.clone()?;
if !state.plans.contains_key(&row.handle) {
return None;
@@ -304,6 +314,42 @@ fn respond(
resume_original(state, &row, pending, resumer, requests)
}
/// The copy here of session `handle` is caught up with sync (or catching up needs a choice, or
/// failed): what's waiting on it resumes it as planned now, with what was written said once the
/// picker is gone; or the chooser opens on the choice, saying why. Returns whether it opened.
///
/// When nothing waits on it any more (the user moved on), it is dropped: the plan stays one that
/// catches the copy up first, so a later enter catches it up again, against sync as it is then.
fn caught_up(
state: &mut State,
handle: HarnessSession,
result: Result<CatchUp, NotResumable>,
resumer: &dyn Resumer,
requests: &Requests,
) -> bool {
state.requested.remove(&(handle.clone(), RESTORE));
let Some((row, action)) = state.pending.clone().filter(|(row, _)| row.handle == handle) else {
return false;
};
match result {
Ok(CatchUp::Ready { plan, status }) => {
state.note = status;
state.plans.insert(handle, Ok(Resume::ready(plan)));
}
Ok(CatchUp::Choice(held)) => {
state.pending = None;
let (targets, fork) = (resumer.continue_targets(&row), resumer.can_fork(&row));
state.open_choice(&row, held, targets, fork, action);
request_flatten(state, requests);
return true;
}
Err(why) => {
state.plans.insert(handle, Err(why));
}
}
false
}
/// Handle a user input, then drop an enter, tab or ctrl-y the user has moved on from: one whose
/// session it moved the selection off ([`moved_off_pending`]), or any, and a continuation being
/// written, when it changed the search ([`searched_again`]).
@@ -349,8 +395,9 @@ fn moved_off_pending(state: &mut State, before: Option<&HarnessSession>) {
}
/// Carry out `action` for `row` (the session acted on) once its plan is known, restoring the
/// session from sync first when its transcript isn't here. `None` keeps the picker open (the
/// plan or the restore is still coming, the session can't be resumed, or it was a copy).
/// session from sync first when its transcript isn't here, and catching it up with sync when it
/// is. `None` keeps the picker open (the plan, the restore or the catch-up is still coming, the
/// session can't be resumed, or it was a copy).
fn complete(
state: &mut State,
row: &SessionRow,
@@ -388,6 +435,14 @@ fn complete(
state.pending = Some((row, action));
return None;
}
(Ok(Resume { catch_up: true, .. }), action) => {
if state.requested.insert((row.handle.clone(), RESTORE)) {
requests.send(Request::CatchUp(Box::new(row.clone())));
}
state.status = Some(("catching up with sync…".to_owned(), Meaning::Annotation));
state.pending = Some((row, action));
return None;
}
(Ok(resume), Pending::Resume) => Some(Outcome::Resume(resume.plan)),
(Ok(resume), Pending::Edit) => Some(Outcome::Edit(resume.plan)),
};
@@ -509,21 +564,46 @@ fn pick(
supersede(state, requests);
match line {
Destination::Original => return resume_original(state, &row, action, resumer, requests),
Destination::Fork => start_fork(state, row, action, requests),
Destination::AsIs(plan) => return as_is(state, *plan, action),
Destination::Fork(branch) => start_fork(state, row, branch.map(|b| *b), action, requests),
Destination::Continue(target) => start_continuation(state, row, target, action, requests),
}
None
}
/// Fork `row`, then carry out `action` (see [`finish_fork`]). Copying writes nothing: the
/// command copied forks it when run.
fn start_fork(state: &mut State, row: SessionRow, action: Pending, requests: &Requests) {
/// Resume the copy here as it is, with `plan`, rather than catch it up: what `action` does with
/// it. Copying copies the agent's own command.
fn as_is(state: &mut State, plan: ResumePlan, action: Pending) -> Option<Outcome> {
match action {
Pending::Resume => Some(Outcome::Resume(plan)),
Pending::Edit => Some(Outcome::Edit(plan)),
Pending::Copy => {
copy(state, &resumer::shell_line(&plan));
None
}
}
}
/// Fork `row` (from `branch`'s head, when given), then carry out `action` (see [`finish_fork`]).
/// Copying writes nothing: the command copied forks it when run.
fn start_fork(
state: &mut State,
row: SessionRow,
branch: Option<Branch>,
action: Pending,
requests: &Requests,
) {
if action == Pending::Copy {
let id = resumer::quote(row.handle.session.as_ref());
copy(state, &format!("atuin ai resume {id} --fork"));
let pick = branch.map(|b| format!(" --branch {}", resumer::quote(&b.selector)));
copy(state, &format!("atuin ai resume {id} --fork{}", pick.unwrap_or_default()));
return;
}
state.status = Some(("forking…".to_owned(), Meaning::Annotation));
let status = match &branch {
Some(branch) => format!("forking from {}'s…", branch.host),
None => "forking…".to_owned(),
};
state.status = Some((status, Meaning::Annotation));
state.continued = state.continued.wrapping_add(1);
let id = state.continued;
let target = row.handle.harness;
@@ -533,7 +613,8 @@ fn start_fork(state: &mut State, row: SessionRow, action: Pending, requests: &Re
fork: true,
action,
});
requests.send(Request::Fork(Box::new(row), id));
let head = branch.map(|b| b.head.source_id);
requests.send(Request::Fork(Box::new(row), id, head));
}
/// Fork `id` is written (or failed): as [`finish_continuation`].
@@ -635,9 +716,10 @@ fn finish_continuation(
/// What ctrl-y copies to resume `row`, planned as `resume`, in its own harness. Copying writes
/// nothing, so when running the harness's own command wouldn't do (the session has to be
/// restored from sync first), it is `atuin ai resume <id>`, which does that when run.
/// restored from sync, or caught up with it, first), it is `atuin ai resume <id>`, which does
/// that when run.
fn resume_line(row: &SessionRow, resume: &Resume) -> String {
if resume.restore.is_none() {
if resume.restore.is_none() && !resume.catch_up {
return resumer::shell_line(&resume.plan);
}
format!("atuin ai resume {}", resumer::quote(row.handle.session.as_ref()))
@@ -807,6 +889,7 @@ impl Picker<'_> {
if let Some(outcome) =
respond(&mut state, response, resumer, &requests)
{
note = state.note.take();
break 'render outcome;
}
None
+329 -35
View File
@@ -10,20 +10,28 @@
//! resumed from the synced messages: the plan says so ([`Resume::restore`]), and only once the
//! user accepts it does [`Resumer::restore`] write the transcript back out
//! ([`Harness::rehydrate`](atuin_common::harnesstools::Harness::rehydrate)) and plan resuming it.
//! A session that went on separately on several machines is restored along one branch only (the
//! newest, unless one was named), never with every branch's messages at once.
//! One that is here is caught up with sync first ([`Resumer::catch_up`], [`super::catchup`]).
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use async_trait::async_trait;
use atuin_client::ai_session::HarnessKind;
use atuin_client::ai_session::{HarnessKind, SourceId};
use atuin_client::settings::AiSessionResume;
use atuin_common::harnesstools::continuation::{self, Flattened};
use atuin_common::harnesstools::rehydrate::{
ForkOf, RehydrateError, RehydrateMessage, RehydrateSession,
};
pub use atuin_common::harnesstools::resume::{ResumeError, ResumePlan, ResumeTarget, quote};
use atuin_common::harnesstools::sync::{
AppendOptions, AppendOutcome, Liveness, LocalTip, SessionSync as _, SyncError,
};
use atuin_common::harnesstools::{AnyHarness, Harness as _, fork};
use super::ResumeContext;
use super::catchup::{self, CatchUp, Held, Step, Why};
use super::source::{SessionRow, SessionSource, harness_label};
/// Why a session can be viewed but not resumed here.
@@ -44,6 +52,9 @@ pub enum NotResumable {
#[error("forking it failed: {0}")]
Fork(String),
#[error("catching it up with sync failed: {0}")]
CatchUp(String),
/// A continuation or fork of a session with nothing of the conversation in it.
#[error(transparent)]
Empty(#[from] continuation::NothingToContinue),
@@ -59,6 +70,11 @@ pub struct Resume {
/// Set when the session's transcript isn't on this machine: it is written from the synced
/// messages first ([`Resumer::restore`]), and `plan` only shows what will run.
pub restore: Option<Restore>,
/// Set when the session's transcript is here: it is caught up with sync first
/// ([`Resumer::catch_up`]), and `plan` resumes it as it is.
pub catch_up: bool,
/// Whether an agent here has the session open, or may have.
pub live: bool,
}
impl Resume {
@@ -67,6 +83,8 @@ impl Resume {
Self {
plan,
restore: None,
catch_up: false,
live: false,
}
}
}
@@ -158,6 +176,24 @@ pub trait Resumer: Send + Sync {
restore: &Restore,
) -> Result<ResumePlan, NotResumable>;
/// Bring this machine's copy of `session` to `head` (one of its heads; the newest when
/// `None`) from what `source` holds of it, and plan resuming it ([`super::catchup`]): restore
/// it when it isn't here, append what it lacks, or say what to choose. Only once the user has
/// chosen to resume it.
async fn catch_up(
&self,
source: &dyn SessionSource,
session: &SessionRow,
_head: Option<&SourceId>,
) -> Result<CatchUp, NotResumable> {
let resume = self.plan(session).await?;
let (plan, status) = match &resume.restore {
Some(restore) => (self.restore(source, session, restore).await?, restored(restore)),
None => (resume.plan, None),
};
Ok(CatchUp::Ready { plan, status })
}
/// The harnesses `session` can be continued in here: installed, and not its own.
fn continue_targets(&self, _session: &SessionRow) -> Vec<HarnessKind> {
Vec::new()
@@ -217,6 +253,45 @@ pub trait Machine: Send + Sync {
/// Whether `program` can be run here.
fn installed(&self, program: &str) -> bool;
/// What `harness`'s copy of session `id` here holds ([`SessionSync::local_tip`]).
///
/// [`SessionSync::local_tip`]: atuin_common::harnesstools::sync::SessionSync::local_tip
async fn local_tip(
&self,
_harness: AnyHarness,
_id: &str,
) -> Result<Option<LocalTip>, SyncError> {
Ok(None)
}
/// Whether a `harness` process here may be writing session `id`
/// ([`SessionSync::is_live`]).
///
/// [`SessionSync::is_live`]: atuin_common::harnesstools::sync::SessionSync::is_live
async fn is_live(&self, _harness: AnyHarness, _id: &str, _cwd: Option<&Path>) -> Liveness {
Liveness::NotLive
}
/// Append `lines` to `harness`'s copy of session `id`, read as `base`
/// ([`SessionSync::append`]).
///
/// [`SessionSync::append`]: atuin_common::harnesstools::sync::SessionSync::append
async fn append(
&self,
_harness: AnyHarness,
_id: &str,
_base: &LocalTip,
_lines: &[RehydrateMessage],
_options: &AppendOptions<'_>,
) -> Result<AppendOutcome, SyncError> {
Err(SyncError::Unsupported("this machine can't catch sessions up"))
}
}
/// What the status line says of a session restored as `restore`, when anything.
fn restored(restore: &Restore) -> Option<String> {
restore.note.as_ref().map(|note| format!("restored the session from sync; {note}"))
}
/// This machine, as the harnesses themselves see it.
@@ -239,6 +314,29 @@ impl Machine for ThisMachine {
fn installed(&self, program: &str) -> bool {
on_path(program)
}
async fn local_tip(
&self,
harness: AnyHarness,
id: &str,
) -> Result<Option<LocalTip>, SyncError> {
harness.local_tip(id).await
}
async fn is_live(&self, harness: AnyHarness, id: &str, cwd: Option<&Path>) -> Liveness {
harness.is_live(id, cwd).await
}
async fn append(
&self,
harness: AnyHarness,
id: &str,
base: &LocalTip,
lines: &[RehydrateMessage],
options: &AppendOptions<'_>,
) -> Result<AppendOutcome, SyncError> {
harness.append(id, base, lines, options).await
}
}
/// The real [`Resumer`]: the harness's own resume command (or the user's template), restoring
@@ -320,11 +418,54 @@ impl HarnessResumer {
self.check_installed(&plan)?;
Ok(plan)
}
/// [`Resumer::restore`], along `head` (see [`restored_session`]).
async fn restore_along(
&self,
source: &dyn SessionSource,
session: &SessionRow,
restore: &Restore,
head: Option<&SourceId>,
) -> Result<ResumePlan, NotResumable> {
let data = restored_session(source, session, &restore.cwd, head)
.await
.map_err(|e| NotResumable::Restore(format!("{e:#}")))?;
self.write_out(session.handle.harness, &data, NotResumable::Restore).await
}
}
#[async_trait]
impl Resumer for HarnessResumer {
async fn plan(&self, session: &SessionRow) -> Result<Resume, NotResumable> {
/// `session` as `source` holds it, to be written out and resumed in `cwd`: along `head` (or, for
/// a session that diverged, its newest head) when there is one ([`catchup::branch_rows`]), so no
/// other branch's messages come with it; else every row synced of it.
async fn restored_session(
source: &dyn SessionSource,
session: &SessionRow,
cwd: &Path,
head: Option<&SourceId>,
) -> eyre::Result<RehydrateSession> {
let mut data = source.rehydrate(&session.handle, cwd).await?;
if let Some(rows) = catchup::branch_rows(source, &session.handle, head).await? {
data.messages = rows;
}
Ok(data)
}
/// This machine's copy of a session, as [`HarnessResumer::plan_here`] read it.
struct Copy {
/// What it holds ([`Machine::local_tip`]).
tip: Result<Option<LocalTip>, SyncError>,
/// Where an agent here would have it open, as asked whether one has: where the copy records
/// working, else where the session works here ([`local_dir`]); `None` (an agent of its
/// harness anywhere counts) when neither is known.
dir: Option<PathBuf>,
}
impl HarnessResumer {
/// [`Resumer::plan`], with the copy here it planned resuming, when there is one.
async fn plan_here(
&self,
session: &SessionRow,
) -> Result<(Resume, Option<Copy>), NotResumable> {
let kind = session.handle.harness;
let harness = kind.harness().ok_or(NotResumable::Unsupported(harness_label(kind)))?;
let id = session.handle.session.as_ref();
@@ -345,21 +486,42 @@ impl Resumer for HarnessResumer {
target.cwd = Some(resolve_cwd(session.cwd.as_deref(), &self.context).cwd);
}
let target = target.with_native_path(native);
Resume::ready(harness.resume(&target, template)?.prepare()?)
// Where this copy works, as it records; else where the session works here,
// which for another host's session isn't where it was recorded.
let tip = self.machine.local_tip(harness, id).await;
let recorded = tip.as_ref().ok().and_then(|t| t.as_ref()?.cwd.clone());
let dir = recorded
.filter(|dir| dir.is_dir())
.or_else(|| local_dir(session.cwd.as_deref(), &self.context));
let live = self.machine.is_live(harness, id, dir.as_deref()).await;
let resume = Resume {
catch_up: true,
live: live != Liveness::NotLive,
..Resume::ready(harness.resume(&target, template)?.prepare()?)
};
(resume, Some(Copy { tip, dir }))
}
None => {
let restore = resolve_cwd(session.cwd.as_deref(), &self.context);
target.cwd = Some(restore.cwd.clone());
// The transcript is only written once the user accepts the restore.
Resume {
plan: self.unwritten_plan(harness, kind, &target)?,
let resume = Resume {
restore: Some(restore),
}
..Resume::ready(self.unwritten_plan(harness, kind, &target)?)
};
(resume, None)
}
};
self.check_installed(&resume.plan)?;
self.check_installed(&resume.0.plan)?;
Ok(resume)
}
}
#[async_trait]
impl Resumer for HarnessResumer {
async fn plan(&self, session: &SessionRow) -> Result<Resume, NotResumable> {
Ok(self.plan_here(session).await?.0)
}
async fn restore(
&self,
@@ -367,11 +529,107 @@ impl Resumer for HarnessResumer {
session: &SessionRow,
restore: &Restore,
) -> Result<ResumePlan, NotResumable> {
let data = source
.rehydrate(&session.handle, &restore.cwd)
.await
.map_err(|e| NotResumable::Restore(format!("{e:#}")))?;
self.write_out(session.handle.harness, &data, NotResumable::Restore).await
self.restore_along(source, session, restore, None).await
}
async fn catch_up(
&self,
source: &dyn SessionSource,
session: &SessionRow,
head: Option<&SourceId>,
) -> Result<CatchUp, NotResumable> {
// The append looks for an agent where the plan did, in the copy the plan read.
let (resume, copy) = self.plan_here(session).await?;
let ready = |plan, status| Ok(CatchUp::Ready { plan, status });
if let Some(restore) = &resume.restore {
let plan = self.restore_along(source, session, restore, head).await?;
return ready(plan, restored(restore));
}
let Some(Copy { tip, dir }) = copy else {
return ready(resume.plan, None);
};
let tip = match tip {
Ok(Some(tip)) => Ok(tip),
Ok(None) => {
// Gone since it was located: restored, as a copy that isn't here is.
let restore = resolve_cwd(session.cwd.as_deref(), &self.context);
let plan = self.restore_along(source, session, &restore, head).await?;
return ready(plan, restored(&restore));
}
Err(e) => Err(e),
};
let kind = session.handle.harness;
let harness = kind.harness().ok_or(NotResumable::Unsupported(harness_label(kind)))?;
let id = session.handle.session.as_ref();
let analysis = source.analyse(&session.handle).await;
let Some(analysis) = analysis.map_err(|e| NotResumable::CatchUp(format!("{e:#}")))? else {
return ready(resume.plan, None);
};
let heads = analysis.heads();
let chosen = match head {
Some(named) => heads.iter().position(|h| h.source_id == *named).ok_or_else(|| {
NotResumable::CatchUp(format!("{named} is no longer one of its branches' heads"))
})?,
None => 0,
};
let Some(head) = heads.get(chosen) else {
return ready(resume.plan, None);
};
let here = &self.context.host_id;
let tip = match tip {
Ok(tip) => tip,
// A copy the agent can't read, or can't catch up (opencode 2.0's): never resumed
// as if it were caught up.
Err(e) => {
let branches = catchup::branches(&analysis, None, here);
return Ok(held(Why::Refused(e.to_string()), kind, resume.plan, branches, chosen));
}
};
let branches = || catchup::branches(&analysis, Some(&tip), here);
let (rows, base) = match catchup::classify(&analysis, kind, &head.source_id, &tip) {
Step::AsIs | Step::Ahead => return ready(resume.plan, None),
Step::Choice(why) => return Ok(held(why, kind, resume.plan, branches(), chosen)),
Step::FastForward { rows, base } => (rows, base),
};
// Never under an agent that may be writing the session.
if resume.live {
return Ok(held(Why::Live, kind, resume.plan, branches(), chosen));
}
// Ids the copy must not take (pi's are short): every row synced on any branch.
let mut taken: HashSet<String> = heads
.iter()
.flat_map(|h| analysis.path_to(&h.source_id))
.map(|m| m.source_id.to_string())
.collect();
for row in &rows {
taken.remove(&row.source_id);
}
let options = AppendOptions {
taken_ids: Some(&taken),
cwd: dir.as_deref(),
};
let outcome = match self.machine.append(harness, id, &base, &rows, &options).await {
Ok(outcome) => outcome,
Err(e) => {
let why = match e {
SyncError::Live(_) | SyncError::MaybeLive => Why::Live,
e => Why::Refused(e.to_string()),
};
return Ok(held(why, kind, resume.plan, branches(), chosen));
}
};
// Resumed from where it was written, which a `{path}` template names.
let plan = if outcome.native_path == tip.native_path {
resume.plan
} else {
let mut target = ResumeTarget::new(id).with_native_path(&outcome.native_path);
target.cwd.clone_from(&resume.plan.cwd);
harness.resume(&target, self.templates.template(kind))?.prepare()?
};
let status = (!outcome.appended.is_empty()).then(|| {
catchup::caught_up(catchup::message_count(&rows), &catchup::host_label(head, here))
});
ready(plan, status)
}
fn continue_targets(&self, session: &SessionRow) -> Vec<HarnessKind> {
@@ -401,8 +659,9 @@ impl Resumer for HarnessResumer {
let harness = kind.harness().ok_or(NotResumable::Unsupported(harness_label(kind)))?;
let fail = |e: String| NotResumable::Fork(e);
let restore = resolve_cwd(session.cwd.as_deref(), &self.context);
let mut original = source
.rehydrate(&session.handle, &restore.cwd)
// Restored as a resume would restore it (along its newest head, when it diverged), should
// pi need it written out below.
let mut original = restored_session(source, session, &restore.cwd, None)
.await
.map_err(|e| fail(format!("{e:#}")))?;
// Checked before anything is written. A fork of no conversation is refused, as a
@@ -484,6 +743,24 @@ impl Resumer for HarnessResumer {
}
}
/// Catching up held for `why`: the choice between resuming the copy here with `plan` as it is and
/// forking from one of `branches`.
fn held(
why: Why,
harness: HarnessKind,
plan: ResumePlan,
branches: Vec<catchup::Branch>,
chosen: usize,
) -> CatchUp {
CatchUp::Choice(Box::new(Held {
why,
harness,
plan,
branches,
chosen,
}))
}
/// Where to resume a session that ran in `original` on this machine: there, when it exists;
/// else, when the current directory is in a git repository named like one `original` was in,
/// the same place in that repository (or its root); else the current directory.
@@ -506,7 +783,41 @@ pub fn resolve_cwd(original: Option<&Path>, context: &ResumeContext) -> Restore
note: None,
};
}
let checkout = context.git_root.as_deref().and_then(|root| {
match in_checkout(original, context) {
Some(cwd) => Restore {
note: Some(format!(
"{} isn't on this machine; resuming in {}",
original.display(),
cwd.display()
)),
cwd,
},
None => Restore {
cwd: context.cwd.clone(),
note: Some(format!(
"{} isn't on this machine; resuming in the current directory",
original.display()
)),
},
}
}
/// The directory here a session that ran in `original` works in: `original` when it exists,
/// else the same place in this checkout of its repository ([`resolve_cwd`]); `None` when neither
/// is known, as the current directory [`resolve_cwd`] falls back to says nothing of where an
/// agent may already have it open.
fn local_dir(original: Option<&Path>, context: &ResumeContext) -> Option<PathBuf> {
let original = original?;
if original.is_dir() {
return Some(original.to_owned());
}
in_checkout(original, context)
}
/// The same place as `original` in the checkout the current directory is in, or its root, when
/// `original` was in a repository named like it ([`resolve_cwd`]).
fn in_checkout(original: &Path, context: &ResumeContext) -> Option<PathBuf> {
context.git_root.as_deref().and_then(|root| {
let name = root.file_name()?;
let components: Vec<_> = original.components().collect();
let named: Vec<usize> =
@@ -527,24 +838,7 @@ pub fn resolve_cwd(original: Option<&Path>, context: &ResumeContext) -> Restore
let here = places.iter().position(|place| *place != root && context.cwd.starts_with(place));
let same_place = places.into_iter().nth(here.unwrap_or(0));
Some(same_place.unwrap_or(root))
});
match checkout {
Some(cwd) => Restore {
note: Some(format!(
"{} isn't on this machine; resuming in {}",
original.display(),
cwd.display()
)),
cwd,
},
None => Restore {
cwd: context.cwd.clone(),
note: Some(format!(
"{} isn't on this machine; resuming in the current directory",
original.display()
)),
},
}
})
}
/// Whether `program` can be run: a path to a file, or a file of that name in a `PATH` directory
@@ -15,12 +15,26 @@ use crate::resume_tui::source::{SessionFilter, SessionPreview};
/// A machine that touches nothing: transcripts are found where `found` says, restoring records
/// what it would write, and every program is installed unless `installed` says otherwise.
/// Rehydrating fails for the harness `cant_rehydrate` names.
///
/// Its copies read as `tip`, an agent has them open as `live` says (none, by default; with
/// `agent_in`, one works there, open for a session in that directory or above it, or in no
/// directory known), and appending records what it would write (from where) and the directory it was told the session
/// works in, unless `refuse` names the error to refuse with.
#[derive(Clone, Default)]
struct FakeMachine {
found: HashMap<String, PathBuf>,
written: Arc<Mutex<Vec<RehydrateSession>>>,
missing_programs: Vec<&'static str>,
cant_rehydrate: Option<HarnessKind>,
tip: Option<LocalTip>,
/// Reading the copy fails, saying this.
tip_error: Option<&'static str>,
live: Option<Liveness>,
live_cwds: Arc<Mutex<Vec<Option<PathBuf>>>>,
agent_in: Option<PathBuf>,
refuse: Option<&'static str>,
appended: Arc<Mutex<Vec<String>>>,
append_cwds: Arc<Mutex<Vec<Option<PathBuf>>>>,
}
#[async_trait]
@@ -50,6 +64,53 @@ impl Machine for FakeMachine {
fn installed(&self, program: &str) -> bool {
!self.missing_programs.contains(&program)
}
async fn local_tip(&self, _: AnyHarness, _: &str) -> Result<Option<LocalTip>, SyncError> {
match self.tip_error {
Some(why) => Err(SyncError::Unsupported(why)),
None => Ok(self.tip.clone()),
}
}
async fn is_live(&self, _: AnyHarness, _: &str, cwd: Option<&Path>) -> Liveness {
self.live_cwds.lock().push(cwd.map(Path::to_path_buf));
let agent =
self.agent_in.as_deref().is_some_and(|dir| cwd.is_none_or(|c| dir.starts_with(c)));
match self.live {
Some(live) => live,
None if agent => Liveness::Live { pid: Some(7) },
None => Liveness::NotLive,
}
}
async fn append(
&self,
harness: AnyHarness,
id: &str,
base: &LocalTip,
lines: &[RehydrateMessage],
options: &AppendOptions<'_>,
) -> Result<AppendOutcome, SyncError> {
self.append_cwds.lock().push(options.cwd.map(Path::to_path_buf));
if self.is_live(harness, id, options.cwd).await != Liveness::NotLive {
return Err(SyncError::MaybeLive);
}
match self.refuse {
Some("unsupported") => return Err(SyncError::Unsupported("not a fast-forward")),
Some("changed") => return Err(SyncError::Changed),
Some("id taken") => return Err(SyncError::IdTaken("c".to_owned())),
Some(_) => return Err(SyncError::MaybeLive),
None => {}
}
let ids: Vec<String> = lines.iter().map(|m| m.source_id.clone()).collect();
let from = base.tip_source_id.clone().unwrap_or_default();
self.appended.lock().push(format!("{}@{from}", ids.join(",")));
Ok(AppendOutcome {
native_path: base.native_path.clone(),
tip_source_id: ids.last().cloned(),
appended: ids,
})
}
}
/// A source whose every session has one message, as the sidecar would give it.
@@ -765,3 +826,371 @@ async fn a_codex_fork_reads_the_original_where_it_is(dirs: Dirs, #[case] found:
assert_eq!(fork.id, forked.id);
assert_eq!(fork.fork_of.clone().unwrap().path, found.map(PathBuf::from));
}
// --- catching up with sync ----------------------------------------------------------------------
/// What catching the synced Claude Code session up with this machine's copy (read as the
/// machine's `tip`; none: not here) says, and what it appended.
async fn catch_up(dirs: &Dirs, machine: FakeMachine, diverged: bool) -> (String, Vec<String>) {
catch_up_in(dirs, machine, diverged, &dirs.elsewhere).await
}
/// [`catch_up`], with the session recorded working in `cwd`.
async fn catch_up_in(
dirs: &Dirs,
mut machine: FakeMachine,
diverged: bool,
cwd: &Path,
) -> (String, Vec<String>) {
if machine.tip.is_some() {
machine.found.insert("s".to_owned(), PathBuf::from("/t/s.jsonl"));
}
let mut session = row(HarnessKind::ClaudeCode, "s", cwd, false);
session.handle = fake::synced_handle();
let source = fake::FakeSource::from_rows(vec![session.clone()])
.with_synced(&session.handle, fake::synced_rows(diverged));
let caught = resumer(dirs, &machine).catch_up(&source, &session, None).await.unwrap();
let said = match caught {
CatchUp::Ready { plan, status } => {
format!("{} {}", plan.native_path.unwrap().display(), status.unwrap_or_default())
}
CatchUp::Choice(held) => {
let forks: Vec<&str> = held.branches.iter().map(|b| b.selector.as_str()).collect();
format!("{} [{}]", held.status(), forks.join(" "))
}
};
(said.trim_end().to_owned(), machine.appended.lock().clone())
}
/// Resuming a session in its own agent catches its copy here up with sync first: restored when
/// it isn't here, resumed as it is when it is at the head, the rows it lacks appended when it is
/// behind; anything else is a choice, with nothing written.
#[rstest]
#[case::not_here(None, None, false, "/restored/s.jsonl", &[])]
#[case::at_the_head(Some((&["a", "b", "c", "d"][..], "d")), None, false, "/t/s.jsonl", &[])]
#[case::behind(
Some((&["a", "b"][..], "b")),
None,
false,
"/t/s.jsonl caught up: 2 messages from @00000002",
&["c,d@b"]
)]
#[case::behind_and_live(
Some((&["a", "b"][..], "b")),
Some(Liveness::Live { pid: Some(7) }),
false,
"Claude Code is running this session here [d]",
&[]
)]
#[case::behind_and_maybe_live(
Some((&["a", "b"][..], "b")),
Some(Liveness::Unknown),
false,
"Claude Code is running this session here [d]",
&[]
)]
#[case::live_at_the_head(
Some((&["a", "b", "c", "d"][..], "d")),
Some(Liveness::Unknown),
false,
"/t/s.jsonl",
&[]
)]
#[case::diverged(
Some((&["a", "b", "c", "d"][..], "d")),
None,
true,
"this copy went another way than this machine's [y d]",
&[]
)]
#[case::diverged_at_the_newest(
Some((&["a", "b", "x", "y"][..], "y")),
None,
true,
"/t/s.jsonl",
&[]
)]
#[case::unsynced(
Some((&["a", "b", "z"][..], "z")),
None,
false,
"this copy has messages sync hasn't got [d]",
&[]
)]
#[tokio::test]
async fn a_copy_here_is_caught_up_or_offered_a_choice(
dirs: Dirs,
#[case] copy: Option<(&[&str], &str)>,
#[case] live: Option<Liveness>,
#[case] diverged: bool,
#[case] said: &str,
#[case] appended: &[&str],
) {
let machine = FakeMachine {
tip: copy.map(|(known, at)| fake::local_tip(known, Some(at))),
live,
..FakeMachine::default()
};
let written = machine.written.clone();
let caught = catch_up(&dirs, machine, diverged).await;
let appended = appended.iter().map(|s| (*s).to_owned()).collect();
assert_eq!(caught, (said.to_owned(), appended));
assert_eq!(written.lock().len(), usize::from(copy.is_none()), "only a restore writes");
}
/// The agent's copy refusing the rows (it changed, they aren't a clean fast-forward, an id is
/// taken, an agent may have it open after all) leaves the choice, with nothing written.
#[rstest]
#[case::unsupported("unsupported", "couldn't catch up: not a fast-forward [d]")]
#[case::changed(
"changed",
"couldn't catch up: the session's transcript changed since it was read [d]"
)]
#[case::id_taken("id taken", "couldn't catch up: c is already taken [d]")]
#[case::maybe_live("maybe live", "Claude Code is running this session here [d]")]
#[tokio::test]
async fn a_refused_append_leaves_the_choice(
dirs: Dirs,
#[case] refuse: &'static str,
#[case] said: &str,
) {
let machine = FakeMachine {
tip: Some(fake::local_tip(&["a", "b"], Some("b"))),
refuse: Some(refuse),
..FakeMachine::default()
};
assert_eq!(catch_up(&dirs, machine, false).await, (said.to_owned(), Vec::new()));
}
/// The append checks for an agent where the session works, as the plan's check does.
#[rstest]
#[tokio::test]
async fn the_append_looks_for_an_agent_where_the_session_works(dirs: Dirs) {
let machine = FakeMachine {
tip: Some(fake::local_tip(&["a", "b"], Some("b"))),
..FakeMachine::default()
};
let cwds = machine.append_cwds.clone();
catch_up(&dirs, machine, false).await;
assert_eq!(*cwds.lock(), vec![Some(dirs.elsewhere.clone())]);
}
#[derive(Debug, Clone, Copy)]
enum Agent {
InTheMappedDir,
Unrelated,
}
/// A session recorded on another machine under a path that isn't here works in the same place
/// in this checkout: an agent there may have it open, so nothing is appended; one in an unrelated
/// directory doesn't. One whose directory here isn't known counts any agent as having it open.
#[rstest]
#[case::mapped_agent_there("/home/u/atuin/crates", Agent::InTheMappedDir, false)]
#[case::mapped_agent_elsewhere("/home/u/atuin/crates", Agent::Unrelated, true)]
#[case::unknown_agent_anywhere("/home/u/zsh", Agent::Unrelated, false)]
#[tokio::test]
async fn an_agent_is_looked_for_where_the_session_works_here(
dirs: Dirs,
#[case] recorded: &str,
#[case] agent: Agent,
#[case] appends: bool,
) {
let agent_in = match agent {
Agent::InTheMappedDir => dirs.repo.join("crates").join("atuin"),
Agent::Unrelated => dirs.elsewhere.clone(),
};
let machine = FakeMachine {
tip: Some(fake::local_tip(&["a", "b"], Some("b"))),
agent_in: Some(agent_in),
..FakeMachine::default()
};
let cwds = machine.append_cwds.clone();
let caught = catch_up_in(&dirs, machine, false, Path::new(recorded)).await;
let want = if appends {
("/t/s.jsonl caught up: 2 messages from @00000002", vec!["c,d@b".to_owned()])
} else {
("Claude Code is running this session here [d]", Vec::new())
};
assert_eq!(caught, (want.0.to_owned(), want.1));
// The append, when it gets that far, looks where the plan did.
if appends {
assert_eq!(*cwds.lock(), vec![Some(dirs.repo.join("crates"))]);
}
}
/// Planning a session whose copy is here says it is to be caught up, and whether an agent here
/// has it open, as the agent's own check says (unknown counting as open).
#[rstest]
#[case::not_live(None, false)]
#[case::live(Some(Liveness::Live { pid: None }), true)]
#[case::unknown(Some(Liveness::Unknown), true)]
#[tokio::test]
async fn a_plan_says_whether_an_agent_here_has_the_session_open(
dirs: Dirs,
#[case] live: Option<Liveness>,
#[case] want: bool,
) {
let machine = FakeMachine {
found: HashMap::from([("abc".to_owned(), PathBuf::from("/t/abc.jsonl"))]),
live,
..FakeMachine::default()
};
let session = row(HarnessKind::ClaudeCode, "abc", &dirs.elsewhere, false);
let resume = resumer(&dirs, &machine).plan(&session).await.unwrap();
assert!(resume.catch_up);
assert_eq!(resume.live, want);
}
/// The synced Claude Code session (diverged or not), with a title host 2 named last: session
/// metadata, which is no node of the tree.
fn synced_with_a_title(diverged: bool) -> Vec<atuin_client::ai_session::Message> {
let handle = fake::synced_handle();
let mut rows = fake::synced_rows(diverged);
let mut title = fake::synced_row(&handle, ("syn-title", None, 2, 6, false));
title.role = atuin_common::harnesstools::session::Role::Other("ai-title".to_owned());
title.content = Vec::new();
rows.push(title);
rows
}
/// A session with no copy here is restored along the head named, else, when it diverged, along
/// its newest head (whether `atuin ai resume` catches it up or the picker restores it), with the
/// rows off the tree that go with it; one that went one way is restored from every row synced of
/// it, as before.
#[rstest]
#[case::one_way(false, None, false, &["a", "b", "c", "d", "syn-title"])]
#[case::one_way_by_the_picker(false, None, true, &["a", "b", "c", "d", "syn-title"])]
#[case::one_way_named(false, Some("d"), false, &["a", "b", "c", "d", "syn-title"])]
#[case::diverged(true, None, false, &["a", "b", "x", "y", "syn-title"])]
#[case::diverged_by_the_picker(true, None, true, &["a", "b", "x", "y", "syn-title"])]
#[case::diverged_named(true, Some("d"), false, &["a", "b", "c", "d", "syn-title"])]
#[tokio::test]
async fn a_session_not_here_is_restored_along_one_branch(
dirs: Dirs,
#[case] diverged: bool,
#[case] head: Option<&str>,
#[case] by_the_picker: bool,
#[case] want: &[&str],
) {
let machine = FakeMachine::default();
let resumer = resumer(&dirs, &machine);
let mut session = row(HarnessKind::ClaudeCode, "s", &dirs.elsewhere, true);
session.handle = fake::synced_handle();
let source = fake::FakeSource::from_rows(vec![session.clone()])
.with_synced(&session.handle, synced_with_a_title(diverged));
if by_the_picker {
let restore = resumer.plan(&session).await.unwrap().restore.unwrap();
resumer.restore(&source, &session, &restore).await.unwrap();
} else {
let head = head.map(|h| SourceId::from(h.to_owned()));
let caught = resumer.catch_up(&source, &session, head.as_ref()).await.unwrap();
assert!(matches!(caught, CatchUp::Ready { .. }), "{caught:?}");
}
let written = machine.written.lock().clone();
let [restored] = written.as_slice() else {
panic!("one session written: {written:?}");
};
let ids: Vec<&str> = restored.messages.iter().map(|m| m.source_id.as_str()).collect();
assert_eq!(ids, want);
}
/// The plan, the catch-up and the append all look for an agent where this machine's copy
/// records working, when it records a directory that is there: not where the session row says
/// (another host's newer directory, maybe). Else where the session works here; else anywhere.
#[rstest]
#[case::where_the_copy_works(true, true)]
#[case::where_the_copy_worked_but_gone(true, false)]
#[case::no_directory_recorded(false, false)]
#[tokio::test]
async fn an_agent_is_looked_for_where_the_copy_works(
dirs: Dirs,
#[case] recorded: bool,
#[case] exists: bool,
) {
let copy_dir = dirs.repo.join("crates").join("atuin");
let copy_dir = if exists {
copy_dir
} else {
dirs.repo.join("gone")
};
let mut tip = fake::local_tip(&["a", "b"], Some("b"));
tip.cwd = recorded.then(|| copy_dir.clone());
let machine = FakeMachine {
tip: Some(tip),
..FakeMachine::default()
};
let (live, appended) = (machine.live_cwds.clone(), machine.append_cwds.clone());
let caught = catch_up(&dirs, machine, false).await;
assert_eq!(caught.1, ["c,d@b"]);
let want = if recorded && exists {
copy_dir
} else {
dirs.elsewhere.clone()
};
// The plan asks, then the append (the fake asks again as it appends).
assert_eq!(*live.lock(), vec![Some(want.clone()); 2], "the plan's check");
assert_eq!(*appended.lock(), vec![Some(want)], "the append's check");
// An agent where the copy works holds the append, wherever the session row says.
if recorded && exists {
let mut tip = fake::local_tip(&["a", "b"], Some("b"));
tip.cwd = Some(dirs.repo.join("crates").join("atuin"));
let machine = FakeMachine {
tip: Some(tip),
agent_in: Some(dirs.repo.join("crates").join("atuin")),
..FakeMachine::default()
};
let caught = catch_up(&dirs, machine, false).await;
assert_eq!(caught, ("Claude Code is running this session here [d]".to_owned(), vec![]));
}
}
/// A copy that can't be read, or caught up (opencode 2.0's), is never resumed as if it were
/// caught up: the choice says why.
#[rstest]
#[tokio::test]
async fn a_copy_that_cant_be_caught_up_leaves_the_choice(dirs: Dirs) {
let mut machine = FakeMachine {
tip_error: Some("opencode 2.0 sessions can't be caught up yet"),
..FakeMachine::default()
};
machine.found.insert("s".to_owned(), PathBuf::from("/t/s.jsonl"));
let caught = catch_up(&dirs, machine, false).await;
assert_eq!(
caught,
("couldn't catch up: opencode 2.0 sessions can't be caught up yet [d]".to_owned(), vec![])
);
}
/// A copy gone since it was located is restored from sync, as one that isn't here is.
#[rstest]
#[tokio::test]
async fn a_copy_gone_since_it_was_located_is_restored(dirs: Dirs) {
let mut machine = FakeMachine::default();
machine.found.insert("s".to_owned(), PathBuf::from("/t/s.jsonl"));
let written = machine.written.clone();
let caught = catch_up(&dirs, machine, false).await;
assert_eq!(caught, ("/restored/s.jsonl".to_owned(), vec![]));
assert_eq!(written.lock().len(), 1);
}
/// A branch named that is no longer a head is an error, not the newest head instead.
#[rstest]
#[tokio::test]
async fn a_branch_no_longer_a_head_is_not_swapped_for_another(dirs: Dirs) {
let machine = FakeMachine {
tip: Some(fake::local_tip(&["a", "b"], Some("b"))),
found: HashMap::from([("s".to_owned(), PathBuf::from("/t/s.jsonl"))]),
..FakeMachine::default()
};
let appended = machine.appended.clone();
let mut session = row(HarnessKind::ClaudeCode, "s", &dirs.elsewhere, false);
session.handle = fake::synced_handle();
let source = fake::FakeSource::from_rows(vec![session.clone()])
.with_synced(&session.handle, fake::synced_rows(false));
let gone = SourceId::from("b".to_owned());
let caught = resumer(&dirs, &machine).catch_up(&source, &session, Some(&gone)).await;
assert!(matches!(caught, Err(NotResumable::CatchUp(_))), "{caught:?}");
assert!(appended.lock().is_empty());
}
+5 -1
View File
@@ -5,7 +5,7 @@ use std::collections::HashMap;
use std::path::{Path, PathBuf};
use async_trait::async_trait;
use atuin_client::ai_session::{AiSessionDatabase, HarnessSession, SearchTerms, Session};
use atuin_client::ai_session::{AiSessionDatabase, Analysis, HarnessSession, SearchTerms, Session};
use atuin_common::harnesstools::rehydrate::RehydrateSession;
use atuin_common::harnesstools::session::{Content, ParentKind};
use atuin_common::string::highlighted::HighlightedString;
@@ -179,6 +179,10 @@ impl SessionSource for SidecarSource {
.await?
.ok_or_else(|| eyre::eyre!("the session isn't in the AI session database"))
}
async fn analyse(&self, session: &HarnessSession) -> Result<Option<Analysis>> {
Ok(Some(self.db.analyse(session).await?))
}
}
#[cfg(test)]
+7 -1
View File
@@ -9,7 +9,7 @@ use std::path::{Path, PathBuf};
use async_trait::async_trait;
use atuin_client::ai_session::{
AtuinSessionId, HarnessKind, HarnessSession, SessionFilter as DbFilter,
Analysis, AtuinSessionId, HarnessKind, HarnessSession, SessionFilter as DbFilter,
};
use atuin_common::harnesstools::rehydrate::RehydrateSession;
use atuin_common::harnesstools::session::Usage;
@@ -148,6 +148,12 @@ pub trait SessionSource: Send + Sync {
) -> eyre::Result<RehydrateSession> {
eyre::bail!("this source can't restore sessions")
}
/// The branches of `session` as its synced rows hold them now: its heads, and the rows on
/// each (see [`super::catchup`]). `None` when the source doesn't know them.
async fn analyse(&self, _session: &HarnessSession) -> eyre::Result<Option<Analysis>> {
Ok(None)
}
}
/// How a session's host shows: `this machine` when it is `here`, and another by a short form of
+5 -1
View File
@@ -35,7 +35,8 @@ pub const LIVE_SECS: u64 = 120;
pub const PREVIEW: u8 = 0;
pub const CHILDREN: u8 = 1;
pub const PLAN: u8 = 2;
/// Restoring the session's transcript from sync, once an action is waiting on it.
/// Restoring the session's transcript from sync, or catching its copy here up with sync, once an
/// action is waiting on it.
pub const RESTORE: u8 = 3;
/// How many rows a search asks for.
@@ -258,6 +259,8 @@ pub struct State {
pub continuing: Option<Continuing>,
/// The last continuation's id.
pub continued: u64,
/// What catching the session being resumed up with sync wrote, to say once the picker is gone.
pub note: Option<String>,
/// A one-line message in the status row (copied, can't resume, search failed).
pub status: Option<(String, Meaning)>,
@@ -309,6 +312,7 @@ impl State {
flattening: None,
continuing: None,
continued: 0,
note: None,
status: None,
rebuilding: None,
now: if settings.prefers_reduced_motion {
+247 -5
View File
@@ -481,6 +481,7 @@ async fn with_chooser(settings: &Settings, query: &str, action: super::state::Pe
state.results[at].active_at = fake::now() - time::Duration::days(1);
let row = state.selected().unwrap().clone();
let resumer = FakeResumer::default();
state.plans.insert(row.handle.clone(), resumer.plan(&row).await);
state.open_chooser(&row, resumer.continue_targets(&row), resumer.can_fork(&row), action);
let flattened = Flattened {
tool_calls: 42,
@@ -653,7 +654,7 @@ async fn f_selects_the_fork_under_the_original() {
let row = state.selected().unwrap().clone();
InputAction::Pick(Box::new(Picked {
row,
line: Destination::Fork,
line: Destination::Fork(None),
action,
}))
};
@@ -700,11 +701,11 @@ async fn the_fork_is_preselected_over_an_original_that_shouldnt_resume(
} else {
fake::now()
};
state.plans.insert(row.handle.clone(), FakeResumer::default().plan(&row).await);
} else {
row.updated_at = fake::now() - time::Duration::days(1);
row.active_at = row.updated_at;
}
state.plans.insert(row.handle.clone(), FakeResumer::default().plan(&row).await);
state.results[state.list.selected] = row.clone();
let resumer = FakeResumer::default();
state.open_chooser(&row, resumer.continue_targets(&row), true, Pending::Resume);
@@ -754,6 +755,7 @@ async fn a_session_with_nothing_to_fork_never_picks_the_fork(#[values(false, tru
row.active_at = fake::now();
let at = state.list.selected;
state.results[at] = row.clone();
state.plans.insert(row.handle.clone(), FakeResumer::default().plan(&row).await);
let targets = FakeResumer::default().continue_targets(&row);
state.open_chooser(&row, targets.clone(), true, Pending::Resume);
assert_eq!(state.chooser.as_ref().unwrap().selected, usize::from(!empty));
@@ -811,7 +813,7 @@ async fn a_picked_fork_is_written_then_resumed() {
let row = state.selected().unwrap().clone();
let picked = Picked {
row: row.clone(),
line: Destination::Fork,
line: Destination::Fork(None),
action: Pending::Resume,
};
assert_eq!(pick(&mut state, picked, resumer.as_ref(), &requests), None, "waits");
@@ -1400,7 +1402,7 @@ async fn a_continuation_picked_supersedes_a_waiting_original(#[values(false, tru
let picked = Picked {
row: row.clone(),
line: if fork {
Destination::Fork
Destination::Fork(None)
} else {
Destination::Continue(HarnessKind::Codex)
},
@@ -1465,7 +1467,7 @@ async fn an_original_chosen_supersedes_a_continuation(
action: Pending::Resume,
};
let first = if fork {
Destination::Fork
Destination::Fork(None)
} else {
Destination::Continue(HarnessKind::Codex)
};
@@ -2185,3 +2187,243 @@ async fn a_clipped_chooser_keeps_the_selected_line_in_sight() {
assert!(out.contains(&format!("> {}", lines[n])), "line {}: {out}", n + 1);
}
}
// --- catching up with sync ----------------------------------------------------------------------
/// A resumer whose catching up of this host's sessions does `caught`, and the selected session's
/// plan from it; and a worker over a source holding the selected session's synced rows.
async fn catching_up(
state: &mut State,
caught: super::catchup::CatchUp,
) -> (
std::sync::Arc<FakeResumer>,
super::worker::Requests,
tokio::sync::mpsc::UnboundedReceiver<super::worker::Response>,
) {
use std::sync::Arc;
let row = state.selected().unwrap().clone();
let resumer = Arc::new(FakeResumer {
catch_up: Some(caught),
..FakeResumer::default()
});
state.plans.insert(row.handle.clone(), resumer.plan(&row).await);
let source =
FakeSource::new().with_synced(&row.handle, fake::synced_rows_of(&row.handle, true));
let (requests, responses) = super::worker::spawn(Arc::new(source), resumer.clone());
(resumer, requests, responses)
}
/// The choice catching up needs for `why`: this copy is at `d`, the other host's head; this
/// host's `y` is the newest.
async fn held(why: super::catchup::Why) -> super::catchup::CatchUp {
use super::catchup::{CatchUp, Held, branches};
let source = FakeSource::from_rows(Vec::new())
.with_synced(&fake::synced_handle(), fake::synced_rows(true));
let analysis = source.analyse(&fake::synced_handle()).await.unwrap().unwrap();
let copy = fake::local_tip(&["a", "b", "c", "d"], Some("d"));
CatchUp::Choice(Box::new(Held {
why,
harness: atuin_client::ai_session::HarnessKind::ClaudeCode,
plan: as_is_plan(),
branches: branches(&analysis, Some(&copy), fake::THIS_HOST_ID),
chosen: 0,
}))
}
fn as_is_plan() -> super::ResumePlan {
super::ResumePlan {
program: "claude".to_owned(),
args: vec!["--resume".to_owned(), "as-is".to_owned()],
cwd: None,
cwd_requirement: atuin_common::harnesstools::resume::CwdRequirement::Preferred,
native_path: None,
}
}
/// Enter on a session whose copy is here catches it up with sync first, through the worker as a
/// restore goes, then resumes it; what was written is said once the picker is gone.
#[rstest]
#[tokio::test]
async fn a_copy_here_is_caught_up_then_resumed() {
use super::catchup::CatchUp;
use super::state::{Pending, RESTORE};
use super::worker::Response;
use super::{Outcome, accept, respond};
let s = settings();
let mut state = loaded(&s, "", 0).await;
let status = "caught up: 2 messages from @00000002".to_owned();
let caught = CatchUp::Ready {
plan: as_is_plan(),
status: Some(status.clone()),
};
let (resumer, requests, _responses) = catching_up(&mut state, caught).await;
let row = state.selected().unwrap().clone();
assert_eq!(accept(&mut state, Pending::Resume, resumer.as_ref(), &requests, false), None);
assert!(state.requested.contains(&(row.handle.clone(), RESTORE)));
assert_eq!(state.status.clone().unwrap().0, "catching up with sync…");
let caught = resumer.catch_up(&FakeSource::new(), &row, None).await;
let response = Response::CaughtUp(row.handle.clone(), caught);
let outcome = respond(&mut state, response, resumer.as_ref(), &requests);
assert_eq!(outcome, Some(Outcome::Resume(as_is_plan())));
assert_eq!(state.note, Some(status));
}
/// A catch-up answered after the user moved off the session is dropped: the session's plan still
/// catches it up, so the next enter on it asks again rather than resuming the copy unchecked.
#[rstest]
#[case::ready(false)]
#[case::failed(true)]
#[tokio::test]
async fn a_catch_up_nobody_waits_on_is_dropped(#[case] failed: bool) {
use super::catchup::CatchUp;
use super::resumer::NotResumable;
use super::state::{Pending, RESTORE};
use super::worker::Response;
use super::{accept, respond};
let s = settings();
let mut state = loaded(&s, "", 0).await;
let caught = CatchUp::Ready {
plan: as_is_plan(),
status: Some("caught up: 2 messages from @00000002".to_owned()),
};
let (resumer, requests, _responses) = catching_up(&mut state, caught).await;
let row = state.selected().unwrap().clone();
assert_eq!(accept(&mut state, Pending::Resume, resumer.as_ref(), &requests, false), None);
// The user moves on.
state.pending = None;
let caught = if failed {
Err(NotResumable::CatchUp("it changed".to_owned()))
} else {
resumer.catch_up(&FakeSource::new(), &row, None).await
};
let response = Response::CaughtUp(row.handle.clone(), caught);
assert_eq!(respond(&mut state, response, resumer.as_ref(), &requests), None);
assert_eq!(state.note, None);
assert!(!state.requested.contains(&(row.handle.clone(), RESTORE)));
let plan = state.plans.get(&row.handle).cloned().unwrap().unwrap();
assert!(plan.catch_up, "still to be caught up: {plan:?}");
// Enter on it again catches it up again.
assert_eq!(accept(&mut state, Pending::Resume, resumer.as_ref(), &requests, false), None);
assert!(state.requested.contains(&(row.handle.clone(), RESTORE)));
}
/// Catching up that needs a choice opens the chooser on it, saying why: this copy as it is
/// first, then a fork line per head, newest first, the newest preselected when an agent here has
/// the session open. Picking the copy resumes it as it is; picking the older head forks from
/// the rows on its branch.
#[rstest]
#[case::live(crate::resume_tui::catchup::Why::Live, "Claude Code is running this session here", 1)]
#[case::diverged(
crate::resume_tui::catchup::Why::Diverged,
"this copy went another way than this machine's",
0
)]
#[tokio::test]
async fn a_choice_offers_the_copy_as_it_is_and_a_fork_per_head(
#[case] why: super::catchup::Why,
#[case] status: &str,
#[case] selected: usize,
) {
use super::chooser::Destination;
use super::state::{InputAction, Pending, Picked};
use super::worker::Response;
use super::{Outcome, accept, finish_fork, pick, respond};
let s = settings();
let mut state = loaded(&s, "", 0).await;
let (resumer, requests, mut responses) = catching_up(&mut state, held(why).await).await;
let row = state.selected().unwrap().clone();
assert_eq!(accept(&mut state, Pending::Resume, resumer.as_ref(), &requests, false), None);
let caught = resumer.catch_up(&FakeSource::new(), &row, None).await;
let response = Response::CaughtUp(row.handle.clone(), caught);
assert_eq!(respond(&mut state, response, resumer.as_ref(), &requests), None);
assert!(state.pending.is_none());
assert_eq!(state.status.clone().unwrap().0, status);
assert_eq!(state.chooser.as_ref().expect("the chooser opens").selected, selected);
let out = text(&render(&mut state, &s, 100, 30));
assert!(out.contains("1 CC Claude Code this copy as is"), "{out}");
let newest = "2 CC Claude Code fork this machine's · +2 since they split · 55m ago";
assert!(out.contains(newest), "{out}");
let older = "3 CC Claude Code fork @00000002's · 57m ago";
assert!(out.contains(older), "{out}");
// The copy as it is.
let line = state.chooser.as_ref().unwrap().line(0);
assert_eq!(line, Destination::AsIs(Box::new(as_is_plan())));
let as_is = Picked {
row: row.clone(),
line,
action: Pending::Resume,
};
let outcome = pick(&mut state, as_is, resumer.as_ref(), &requests);
assert_eq!(outcome, Some(Outcome::Resume(as_is_plan())));
// The older head, forked from its own rows.
let InputAction::Pick(picked) = press(&mut state, &s, "3") else {
panic!("picks");
};
let Picked {
line: Destination::Fork(Some(branch)),
..
} = picked.as_ref()
else {
panic!("{picked:?}");
};
assert_eq!(branch.head.source_id.as_ref(), "d");
assert_eq!(pick(&mut state, *picked, resumer.as_ref(), &requests), None, "waits");
let id = state.continued;
let forked = loop {
let next = tokio::time::timeout(std::time::Duration::from_secs(10), responses.recv());
match next.await.expect("the worker is stuck").expect("the worker stopped") {
Response::Forked(n, forked) if n == id => break forked,
_ => {}
}
};
let rows = resumer.forks.lock()[0].rows.clone().expect("the head's rows");
let ids: Vec<&str> = rows.iter().map(|r| r.source_id.as_str()).collect();
assert_eq!(ids, ["a", "b", "c", "d"]);
let (_, status) = finish_fork(&mut state, id, forked).unwrap();
assert_eq!(status, "forked into a new Claude Code session");
}
/// The fork is preselected when an agent here has the session open, as its plan says, however
/// lately it was written to.
#[rstest]
#[case::open_but_quiet(true, false, 1)]
#[case::written_to_lately_but_closed(false, true, 0)]
#[tokio::test]
async fn a_session_open_here_preselects_the_fork(
#[case] live: bool,
#[case] recent: bool,
#[case] selected: usize,
) {
use super::resumer::Resume;
use super::state::Pending;
let s = settings();
let mut state = loaded(&s, "", 0).await;
let at = state.list.selected;
if !recent {
state.results[at].active_at = fake::now() - time::Duration::days(1);
}
let row = state.results[at].clone();
let plan = FakeResumer::default().plan(&row).await.unwrap().plan;
let resume = Resume {
catch_up: true,
live,
..Resume::ready(plan)
};
state.plans.insert(row.handle.clone(), Ok(resume));
let targets = FakeResumer::default().continue_targets(&row);
state.open_chooser(&row, targets, true, Pending::Resume);
assert_eq!(state.chooser.as_ref().unwrap().selected, selected);
}
+35 -20
View File
@@ -6,8 +6,9 @@
//! screen until then;
//! - details (preview, children, plan) for the selected session. Only the newest request of each
//! kind is kept: while the selection moves, the sessions it passed over are never loaded.
//! Restoring a session from sync, or continuing it in another harness, which only an enter or
//! tab asks for, goes first: only the newest accepted session's restore is kept, and one dropped
//! Restoring a session from sync (or catching its copy here up with sync), or continuing it in
//! another harness, which only an enter or tab asks for, goes first: only the newest accepted
//! session's restore or catch-up is kept, and one dropped
//! before it started is reported ([`Response::Abandoned`]) so that the picker asks again if the
//! user comes back to it. Likewise only the newest continuation or fork is kept, and one is
//! dropped when the picker stops waiting on it ([`Request::CancelContinue`]); once the picker is
@@ -19,14 +20,13 @@
use std::path::PathBuf;
use std::sync::Arc;
use atuin_client::ai_session::{HarnessKind, HarnessSession};
use atuin_client::ai_session::{HarnessKind, HarnessSession, SourceId};
use atuin_client::settings::AiSessionFilterMode as FilterMode;
use atuin_common::harnesstools::continuation::{self, Flattened};
use tokio::sync::mpsc;
use super::resumer::{
Continued, ForkFrom, Forked, NotResumable, Restore, Resume, ResumePlan, Resumer,
};
use super::catchup::{self, CatchUp};
use super::resumer::{Continued, Forked, NotResumable, Restore, Resume, ResumePlan, Resumer};
use super::source::{SessionFilter, SessionPreview, SessionRow, SessionSource};
#[derive(Debug)]
@@ -47,15 +47,18 @@ pub enum Request {
Accept(Box<SessionRow>),
/// Write out the transcript of a session planned with a restore, and plan resuming it.
Restore(Box<SessionRow>, Restore),
/// Catch the copy here of a session up with sync, and plan resuming it. Kept as a restore
/// is, in its place.
CatchUp(Box<SessionRow>),
/// Count what continuing a session in another harness would flatten (reads all of it).
Flatten(HarnessSession, PathBuf),
/// Write a session out as a new session of another harness, and plan resuming that. The id
/// comes back with the answer. Only the newest is kept: one a newer one replaces before it
/// started never runs (its answer would be ignored, as the picker waits on the newer id).
Continue(Box<SessionRow>, HarnessKind, u64),
/// Write a session out as a fork of it, and plan resuming that. Answered as a continuation
/// is, and in its place: a newer pick supersedes either.
Fork(Box<SessionRow>, u64),
/// Write a session out as a fork of it (from a head, when one is given), and plan resuming
/// that. Answered as a continuation is, and in its place: a newer pick supersedes either.
Fork(Box<SessionRow>, u64, Option<SourceId>),
/// The picker no longer waits on a continuation or fork (the user chose something else):
/// drop the one not yet started, so nothing is written for it.
CancelContinue,
@@ -73,8 +76,10 @@ pub enum Response {
Plan(HarnessSession, Result<Resume, NotResumable>),
/// The session is restored (or couldn't be): the plan that resumes it.
Restored(HarnessSession, Result<ResumePlan, NotResumable>),
/// The session's restore was dropped before it started, for a newer accepted session's: it
/// has to be asked for again.
/// The copy here is caught up (or catching up needs a choice, or failed).
CaughtUp(HarnessSession, Result<CatchUp, NotResumable>),
/// The session's restore or catch-up was dropped before it started, for a newer accepted
/// session's: it has to be asked for again.
Abandoned(HarnessSession),
/// What continuing the session in another harness would flatten (or why that can't be read).
Flattened(HarnessSession, Result<Flattened, String>),
@@ -174,7 +179,7 @@ impl Latest {
Request::Children(_) => &mut self.children,
Request::Plan(_) => &mut self.plan,
Request::Accept(_) => &mut self.accept,
Request::Restore(..) => &mut self.restore,
Request::Restore(..) | Request::CatchUp(_) => &mut self.restore,
Request::Flatten(..) => &mut self.flatten,
Request::Continue(..) | Request::Fork(..) => &mut self.continuation,
Request::CancelContinue => {
@@ -185,11 +190,10 @@ impl Latest {
};
let dropped = slot.replace(request)?;
match (dropped, &*slot) {
(Request::Restore(old, _), Some(Request::Restore(new, _)))
if old.handle != new.handle =>
{
Some(old.handle)
}
(
Request::Restore(old, _) | Request::CatchUp(old),
Some(Request::Restore(new, _) | Request::CatchUp(new)),
) if old.handle != new.handle => Some(old.handle),
_ => None,
}
}
@@ -263,6 +267,10 @@ async fn details(
row.handle.clone(),
resumer.restore(source.as_ref(), &row, &restore).await,
),
Request::CatchUp(row) => Response::CaughtUp(
row.handle.clone(),
resumer.catch_up(source.as_ref(), &row, None).await,
),
Request::Flatten(session, cwd) => {
let flattened =
source.rehydrate(&session, &cwd).await.map_err(|e| format!("{e:#}")).and_then(
@@ -273,8 +281,14 @@ async fn details(
Request::Continue(row, target, id) => {
Response::Continued(id, resumer.continue_in(source.as_ref(), &row, target).await)
}
Request::Fork(row, id) => {
Response::Forked(id, resumer.fork(source.as_ref(), &row, ForkFrom::default()).await)
Request::Fork(row, id, head) => {
let source = source.as_ref();
let from = catchup::fork_from(source, &row.handle, head.as_ref()).await;
let forked = match from {
Ok(from) => resumer.fork(source, &row, from).await,
Err(why) => Err(why),
};
Response::Forked(id, forked)
}
Request::Search { .. } | Request::CancelContinue => continue,
};
@@ -296,6 +310,7 @@ mod tests {
use super::*;
use crate::resume_tui::fake::{self, FakeResumer, FakeSource};
use crate::resume_tui::resumer::ForkFrom;
fn roots() -> SessionFilter {
let mut filter = SessionFilter::default();
@@ -536,7 +551,7 @@ mod tests {
// `b`, then `c`: each forked, or continued.
for ((n, id), &fork) in (1..).zip(["b", "c"]).zip(forks) {
tx.send(if fork {
Request::Fork(row(id), n)
Request::Fork(row(id), n, None)
} else {
Request::Continue(row(id), HarnessKind::Codex, n)
});
@@ -262,20 +262,91 @@ impl Analysis {
}
/// The rows from the root of the session down to `row` (a [head](Self::heads), or any row on
/// a branch), in order, following the branch tree: what a transcript on that branch holds,
/// for forking from it, or for fast-forwarding a copy holding the start of it.
/// a branch), in order, following the branch tree: the rows of the tree a transcript on that
/// branch holds, for fast-forwarding a copy holding the start of it.
///
/// Only rows that are nodes of the tree are on it (see the module docs): for Claude Code and
/// Pi, the rows with an id of their own, so not titles or other content-addressed rows. Empty
/// when `row` is not stored, or is no node.
/// Pi, the rows with an id of their own, so not titles or other content-addressed rows (see
/// [`Self::rows_for`] for those too). Empty when `row` is not stored, or is no node.
#[must_use]
pub fn path_to(&self, row: &SourceId) -> Vec<&Message> {
self.path_indices(row).into_iter().map(|i| &self.rows[i]).collect()
}
/// Every row a transcript on the branch ending at `row` holds, in order, for writing one out
/// (a fork from a head, or a restore along it): [its path](Self::path_to), with the rows that
/// are no node of the tree (Claude Code's and Pi's content-addressed rows) that go with it.
///
/// Such a row goes with a branch when it is session metadata (a title, a header: a row of
/// [`Role::Other`] with no content, so nothing the model is given, which plain restore keeps
/// too), or when it goes with a row on the path. Context the model is given under a role of
/// its own (pi's extensions' `custom` messages) is no metadata:
///
/// - a row with no row of the tree before it at all (a Pi session's prompts and replies from
/// before ids) starts every branch;
/// - else, as a row whose parent is not stored does (see the module docs), it follows its
/// host's row of the tree before it;
/// - else, when its host has none before it (a host's own rows from before ids, written after
/// another host's), it goes with its host's first row of the tree after it, which its host
/// wrote them ahead of, not with another host's branch;
/// - and when its host has no row of the tree at all, it follows the row of the tree before it.
///
/// So a row written before a branch point goes with every branch below it, and one written
/// after it goes with the branch its host was on.
///
/// The path keeps its order; each row off it comes in time order, before the first row of the
/// path after it. Empty when `row` is not stored, or is no node.
#[must_use]
pub fn rows_for(&self, row: &SourceId) -> Vec<&Message> {
let path = self.path_indices(row);
if path.is_empty() {
return Vec::new();
}
let on: HashSet<usize> = path.iter().copied().collect();
// Each host's first row of the tree.
let mut first: HashMap<Option<HostId>, usize> = HashMap::new();
for (i, m) in self.rows.iter().enumerate().filter(|(i, _)| self.in_tree[*i]) {
first.entry(m.host).or_insert(i);
}
let mut previous = None;
let mut by_host: HashMap<Option<HostId>, usize> = HashMap::new();
let mut off = Vec::new();
for (i, m) in self.rows.iter().enumerate() {
if self.in_tree[i] {
previous = Some(i);
by_host.insert(m.host, i);
continue;
}
let follows = previous.map(|previous| {
by_host.get(&m.host).or_else(|| first.get(&m.host)).copied().unwrap_or(previous)
});
// Nothing the model is given: context of another role (pi's extensions' messages)
// goes with its branch like any row.
let metadata = matches!(m.role, Role::Other(_)) && m.content.is_empty();
if metadata || follows.is_none_or(|f| on.contains(&f)) {
off.push(i);
}
}
let mut off = off.into_iter().peekable();
let mut rows = Vec::with_capacity(path.len() + off.len());
for p in path {
while let Some(i) = off.next_if(|&i| i < p) {
rows.push(&self.rows[i]);
}
rows.push(&self.rows[p]);
}
rows.extend(off.map(|i| &self.rows[i]));
rows
}
/// [`Self::path_to`], as indices into the rows.
fn path_indices(&self, row: &SourceId) -> Vec<usize> {
let found =
self.rows.iter().zip(&self.in_tree).position(|(m, t)| *t && m.source_id == *row);
let mut path = Vec::new();
let mut at = found;
while let Some(i) = at {
path.push(&self.rows[i]);
path.push(i);
at = self.parent[i];
}
path.reverse();
@@ -21,8 +21,12 @@ enum Kind {
Tool,
/// A Pi session header.
Header,
/// Session metadata: a title.
Title,
/// Context an extension gave the model (pi's `custom_message`): no metadata.
Custom,
}
use Kind::{Header, Prompt, Reply, Tool};
use Kind::{Custom, Header, Prompt, Reply, Title, Tool};
/// A row: its source id, parent, host, timestamp (ms) and kind.
type Row<'a> = (&'a str, Option<&'a str>, u8, i64, Kind);
@@ -44,6 +48,8 @@ fn message(harness: HarnessKind, (id, parent, on, at, kind): Row<'_>) -> Message
Reply => (Role::Assistant, vec![Content::Text(format!("ok {id}"))]),
Tool => (Role::Tool, vec![Content::Other(serde_json::json!(id))]),
Header => (Role::Other("session".to_owned()), Vec::new()),
Title => (Role::Other("session_info".to_owned()), Vec::new()),
Custom => (Role::Other("custom".to_owned()), vec![Content::Text(format!("ctx {id}"))]),
};
Message::builder()
.id(RecordId(atuin_common::utils::uuid_v7()))
@@ -480,3 +486,87 @@ fn a_copy_fast_forwards_only_along_the_heads_line(
};
assert_eq!(found, behind.map(|rows| rows.iter().map(|r| (*r).to_owned()).collect::<Vec<_>>()));
}
/// A Pi session from before ids (its first prompts and replies content-addressed, so no node),
/// then ids, diverged off a1: host 2's side, a reply of it with no id, and a title host 2 named.
const PRE_IDS: &[Row<'static>] = &[
("syn-p0", None, 1, 0, Prompt),
("syn-r0", None, 1, 1, Reply),
("u1", None, 1, 2, Prompt),
("a1", Some("u1"), 1, 3, Reply),
("syn-r1", None, 1, 4, Reply),
("u2", Some("a1"), 1, 5, Prompt),
("syn-r2", None, 1, 6, Reply),
("v2", Some("a1"), 2, 10, Prompt),
("b2", Some("v2"), 2, 11, Reply),
("syn-r3", None, 2, 12, Reply),
("syn-title", None, 2, 13, Title),
("syn-c3", None, 2, 14, Custom),
];
/// The rows for a branch are its path with the rows off the tree that go with it: those before
/// any row of the tree, those following on from a row on the path (by their host's row before
/// them), and session metadata wherever it is; never another branch's own, even context an
/// extension gave the model, which is no metadata.
#[rstest]
#[case::host_2s(
"b2",
&["syn-p0", "syn-r0", "u1", "a1", "syn-r1", "v2", "b2", "syn-r3", "syn-title", "syn-c3"]
)]
#[case::host_1s("u2", &["syn-p0", "syn-r0", "u1", "a1", "syn-r1", "u2", "syn-r2", "syn-title"])]
#[case::before_the_split("a1", &["syn-p0", "syn-r0", "u1", "a1", "syn-r1", "syn-title"])]
#[case::no_node("syn-r1", &[])]
#[case::not_stored("nope", &[])]
fn a_branchs_rows_hold_what_goes_with_it(#[case] head: &str, #[case] want: &[&str]) {
let analysis = analyse(HarnessKind::Pi, PRE_IDS);
assert!(analysis.diverged());
let rows: Vec<&str> = analysis
.rows_for(&SourceId::from(head.to_owned()))
.iter()
.map(|m| m.source_id.as_ref())
.collect();
assert_eq!(rows, want);
}
/// Host 2 joins a session host 1 started with rows from before ids (its own pi), then branches
/// off a1 with ids: its rows from before ids go with its own branch, not host 1's, though host 1
/// wrote the row of the tree before them.
const HOST_2_PRE_IDS: &[Row<'static>] = &[
("u1", None, 1, 0, Prompt),
("a1", Some("u1"), 1, 1, Reply),
("u2", Some("a1"), 1, 2, Prompt),
("syn-p2", None, 2, 3, Prompt),
("syn-r2", None, 2, 4, Reply),
("a2", Some("u2"), 1, 5, Reply),
("v2", Some("a1"), 2, 6, Prompt),
("b2", Some("v2"), 2, 7, Reply),
];
#[rstest]
#[case::host_2s("b2", &["u1", "a1", "syn-p2", "syn-r2", "v2", "b2"])]
#[case::host_1s("a2", &["u1", "a1", "u2", "a2"])]
fn a_hosts_rows_before_its_first_node_go_with_its_branch(
#[case] head: &str,
#[case] want: &[&str],
) {
let analysis = analyse(HarnessKind::Pi, HOST_2_PRE_IDS);
assert!(analysis.diverged());
let rows: Vec<&str> = analysis
.rows_for(&SourceId::from(head.to_owned()))
.iter()
.map(|m| m.source_id.as_ref())
.collect();
assert_eq!(rows, want);
}
/// With every row a node of the tree, a branch's rows are its path.
#[rstest]
#[case::codex(CODEX)]
#[case::opencode(OPENCODE)]
#[case::claude_code(CC)]
fn rows_for_a_branch_of_nodes_are_its_path(#[case] harness: HarnessKind) {
let analysis = analyse(harness, GROWN);
for head in analysis.heads() {
assert_eq!(analysis.rows_for(&head.source_id), analysis.path_to(&head.source_id));
}
}
+38 -3
View File
@@ -145,8 +145,8 @@ first), Atuin asks where to resume it:
grouped under the original, with an Atuin id of its own. It keeps the
original's title. Pi names the original by its file, so a Pi session that
isn't on this machine is written out first. Forking is selected instead of
the original when the original can't resume here, or looks to be running
here still. Copilot sessions, and sessions with no messages, can't be
the original when the original can't resume here, or its agent has it open
on this machine. Copilot sessions, and sessions with no messages, can't be
forked.
- Every other agent installed on this machine follows. Picking one continues
the session there as a new session: the conversation carries over, but that
@@ -162,6 +162,35 @@ first), Atuin asks where to resume it:
directory is gone, for an agent that needs it), its line is dimmed with the
reason, and the next one is selected instead.
A session keeps one id on every machine. When you resume one in its own agent
and its transcript is on this machine, Atuin first catches it up with the
synced messages:
- If the session continued on another machine since, the messages this copy
lacks are appended to it, and it resumes in place. The status line says
`caught up: 12 messages from @3f9a12bc`.
- If this copy is up to date, or went on past the newest messages along the
same branch, it resumes unchanged.
- Otherwise Atuin writes nothing and asks, with the chooser saying why: when
the agent is running this session here (`Claude Code is running this session
here`), when this copy went another way than the newest messages, when it
has messages sync hasn't got, or when this copy can't be caught up
(`couldn't catch up: ...`). Its first line resumes this copy unchanged, and
a fork line for each branch of the session, newest first, forks from that
branch (`fork @3f9a12bc's · +11 since they split · 20m ago`). With the agent
running here, the newest fork is selected.
Atuin never merges branches, and never writes to a session its agent has open.
Claude Code, and Codex where it keeps session locks, say which session they
have open. opencode, Pi and other Codex installs don't, so Atuin treats a
session as possibly open, and doesn't catch it up, while one of that agent's
processes works in the session's directory or below it: the directory this
copy records, or else where the session works on this machine. A process elsewhere
counts when its command line names the session (`opencode -s <id>`,
`codex resume <id>`, `pi --session <file>`). While the session's transcript has
changed in the last two minutes, any running process of that agent counts,
wherever it works.
A session written back out from sync, or continued in another agent, resumes in
the directory the session ran in when that exists here. Otherwise, when you're
in a checkout of a repository with the same name, it resumes in the same place
@@ -191,7 +220,13 @@ From the command line, `atuin ai resume <id>` resumes a session in its own
agent directly (an id prefix works too), writing it out from sync first when it
isn't on this machine, and `atuin ai resume <id> --in codex` continues it in
another (`claude`, `codex`, `opencode` or `pi`). `atuin ai resume <id> --fork`
forks it. `--print` prints the command instead of running it.
forks it. It catches the session up with sync as the picker does, and fails
when that needs a choice, saying which: `--as-is` resumes this copy unchanged,
and `--fork` forks it from its newest branch. `--branch` picks the branch to
catch up to, or to fork from: `this`, `@<host id>`, or the start of the id the
error lists. `--as-is` and `--branch` need an id that names a single session:
when it names several, Atuin lists them and asks you to be more specific.
`--print` prints the command instead of running it.
The id can be the agent's own, or the session's Atuin id, which Inspect and
the MCP session tools show. An Atuin id is a UUIDv7, written as 32 hex digits