diff --git a/Cargo.lock b/Cargo.lock index 9e246726b..78e5d1e5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -474,7 +474,6 @@ dependencies = [ "url", "uuid", "wiremock", - "xxhash-rust", "zstd", ] diff --git a/crates/atuin-ai/src/commands/session.rs b/crates/atuin-ai/src/commands/session.rs index 29e662507..0ea90477e 100644 --- a/crates/atuin-ai/src/commands/session.rs +++ b/crates/atuin-ai/src/commands/session.rs @@ -146,10 +146,8 @@ pub async fn run(cmd: Cmd, settings: &Settings) -> Result<()> { if reported.is_none() { eprintln!("AI sessions are being rebuilt from the record store; waiting..."); } - if let Some((replayed, pending)) = progress - && reported != Some(progress) - && pending > 0 - { + let (replayed, pending) = progress; + if reported != Some(progress) && pending > 0 { eprintln!(" {replayed} of {pending} records replayed"); } reported = Some(progress); @@ -624,8 +622,8 @@ fn write_message_text(out: &mut dyn Write, m: &Message) -> io::Result<()> { for block in &m.content { match block { Content::Text(t) => writeln!(out, "{}", sanitize(t))?, - Content::Other(json) => writeln!(out, "{}", sanitize(&json.to_string()))?, - Content::Reasoning(t) => writeln!(out, "[thinking] {}", sanitize(t))?, + // Capture never stores reasoning text or raw blocks. + Content::Reasoning(_) | Content::Other(_) => {} Content::ReasoningSummary { tokens } => { let tokens = tokens.or(m.usage.and_then(|u| u.reasoning)); writeln!(out, "[thinking] {}", reasoning_label(tokens))?; @@ -730,18 +728,8 @@ fn message_summary(m: &Message) -> Option { return Some(Summary::Text(line)); } } - Content::Other(json) => { - let line = one_line(&json.to_string(), SUMMARY_WIDTH); - if !line.is_empty() { - return Some(Summary::Text(line)); - } - } - Content::Reasoning(t) => { - let line = one_line(t, SUMMARY_WIDTH); - if !line.is_empty() { - return Some(Summary::Thinking(line)); - } - } + // Capture never stores reasoning text or raw blocks. + Content::Reasoning(_) | Content::Other(_) => {} Content::ReasoningSummary { tokens } => { let tokens = tokens.or(m.usage.and_then(|u| u.reasoning)); return Some(Summary::Thinking(reasoning_label(tokens))); @@ -1196,13 +1184,11 @@ fn session_json(s: &Session) -> SessionJson { } /// `reasoning` is the message's reasoning token count, for a summary block that lacks its own. -fn content_json(block: &Content, reasoning: Option) -> ContentJson { - match block { +/// `None` for what capture never stores (reasoning text, raw blocks). +fn content_json(block: &Content, reasoning: Option) -> Option { + Some(match block { Content::Text(t) => ContentJson::Text { text: t.clone() }, - Content::Other(json) => ContentJson::Text { - text: json.to_string(), - }, - Content::Reasoning(t) => ContentJson::Thinking { text: t.clone() }, + Content::Reasoning(_) | Content::Other(_) => return None, Content::ReasoningSummary { tokens } => ContentJson::Thinking { text: reasoning_label(tokens.or(reasoning)), }, @@ -1237,7 +1223,7 @@ fn content_json(block: &Content, reasoning: Option) -> ContentJson { }) .collect(), }, - } + }) } fn message_json(m: &Message) -> MessageJson { @@ -1249,7 +1235,7 @@ fn message_json(m: &Message) -> MessageJson { model: m.model.clone(), cwd: m.cwd.as_ref().map(|cwd| cwd.to_string_lossy().into_owned()), git_branch: m.git_branch.clone(), - content: m.content.iter().map(|block| content_json(block, reasoning)).collect(), + content: m.content.iter().filter_map(|block| content_json(block, reasoning)).collect(), tokens: m.usage, stop_reason: match &m.stop_reason { Some(StopReason::Other(label)) => label.clone(), @@ -1279,7 +1265,9 @@ mod tests { fn session(harness: HarnessKind, id: &str) -> Session { Session::builder() .handle(handle(harness, id)) - .atuin_id(handle(harness, id).atuin_id(OffsetDateTime::UNIX_EPOCH)) + .atuin_id(AtuinSessionId::from(uuid::Uuid::from_u128(xxhash_rust::xxh3::xxh3_128( + format!("{harness:?}:{id}").as_bytes(), + )))) .started_at(OffsetDateTime::UNIX_EPOCH) .updated_at(OffsetDateTime::UNIX_EPOCH) .usage(Usage::default()) @@ -1540,8 +1528,8 @@ mod tests { #[rstest] fn summary_marks_thinking() { - let m = msg(Role::Assistant, vec![Content::Reasoning("pondering".to_owned())]); - assert_eq!(message_summary(&m).unwrap().render(false), "» pondering"); + let m = msg(Role::Assistant, vec![Content::ReasoningSummary { tokens: Some(7) }]); + assert_eq!(message_summary(&m).unwrap().render(false), "» Reasoning · 7 tokens"); } #[rstest] diff --git a/crates/atuin-ai/src/resume_tui/fake.rs b/crates/atuin-ai/src/resume_tui/fake.rs index 9b73a7a66..d904f0337 100644 --- a/crates/atuin-ai/src/resume_tui/fake.rs +++ b/crates/atuin-ai/src/resume_tui/fake.rs @@ -10,7 +10,8 @@ use std::sync::Arc; use async_trait::async_trait; use atuin_client::ai_session::{ - AiSessionDatabase, Analysis, HarnessKind, HarnessSession, Message, NativeSessionId, SourceId, + AiSessionDatabase, Analysis, AtuinSessionId, HarnessKind, HarnessSession, Message, + NativeSessionId, SourceId, }; use atuin_common::harnesstools::Harness as _; use atuin_common::harnesstools::continuation::Flattened; @@ -54,7 +55,7 @@ pub fn context() -> ResumeContext { pub fn row(harness: HarnessKind, id: &str, title: &str) -> SessionRow { let started_at = now() - Duration::hours(1); SessionRow { - atuin_id: handle(harness, id).atuin_id(started_at), + atuin_id: atuin_id(harness, id, started_at), handle: handle(harness, id), parent: None, relation: Relation::Root, @@ -81,6 +82,16 @@ fn handle(harness: HarnessKind, id: &str) -> HarnessSession { } } +/// A fake session's atuin id: a UUIDv7 at its start, like a minted one, but with its other bits +/// from its handle, so renders are deterministic. +fn atuin_id(harness: HarnessKind, id: &str, started_at: OffsetDateTime) -> AtuinSessionId { + let millis = u64::try_from(started_at.unix_timestamp_nanos() / 1_000_000).unwrap_or(0); + let hash = xxhash_rust::xxh3::xxh3_128(format!("{harness:?}:{id}").as_bytes()).to_le_bytes(); + let mut bits = [0; 10]; + bits.copy_from_slice(&hash[..10]); + uuid::Builder::from_unix_timestamp_millis(millis, &bits).into_uuid().into() +} + #[derive(Clone, Copy, PartialEq, Eq)] enum Role { User, @@ -210,7 +221,7 @@ fn build(spec: Spec, relation: Relation, parent: Option<&SessionRow>) -> FakeSes }; FakeSession { row: SessionRow { - atuin_id: handle(spec.harness, spec.id).atuin_id(updated_at - spec.duration), + atuin_id: atuin_id(spec.harness, spec.id, updated_at - spec.duration), handle: handle(spec.harness, spec.id), parent: parent.map(|p| p.handle.clone()), relation, diff --git a/crates/atuin-ai/src/resume_tui/rebuild.rs b/crates/atuin-ai/src/resume_tui/rebuild.rs index 7819b1bb4..896be821d 100644 --- a/crates/atuin-ai/src/resume_tui/rebuild.rs +++ b/crates/atuin-ai/src/resume_tui/rebuild.rs @@ -21,20 +21,22 @@ const PROBE_TIMEOUT: Duration = Duration::from_millis(300); /// The daemon is rebuilding the session index. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct Rebuilding { - /// Records replayed, and roughly how many there are to replay, when the daemon says. - pub progress: Option<(u64, u64)>, + /// Records replayed, and roughly how many there are to replay. + pub progress: (u64, u64), } impl Rebuilding { /// The picker's status line while it lasts. #[must_use] pub fn status(&self) -> String { - match self.progress { - Some((replayed, pending)) if pending > 0 => format!( + let (replayed, pending) = self.progress; + if pending > 0 { + format!( "rebuilding the session index: {replayed} of {pending} records; results may be \ incomplete" - ), - _ => "rebuilding the session index; results may be incomplete".to_owned(), + ) + } else { + "rebuilding the session index; results may be incomplete".to_owned() } } } @@ -117,7 +119,7 @@ mod tests { fn at(replayed: u64, pending: u64) -> Rebuilding { Rebuilding { - progress: Some((replayed, pending)), + progress: (replayed, pending), } } @@ -141,10 +143,6 @@ mod tests { "rebuilding the session index: 1200 of 5000 records; results may be incomplete" )] #[case::nothing_to_replay(at(0, 0), "rebuilding the session index; results may be incomplete")] - #[case::unknown( - Rebuilding { progress: None }, - "rebuilding the session index; results may be incomplete" - )] fn the_status_says_how_far_it_has_got(#[case] rebuilding: Rebuilding, #[case] want: &str) { assert_eq!(rebuilding.status(), want); } diff --git a/crates/atuin-ai/src/resume_tui/sidecar.rs b/crates/atuin-ai/src/resume_tui/sidecar.rs index 64f13a31b..d7b9b8cab 100644 --- a/crates/atuin-ai/src/resume_tui/sidecar.rs +++ b/crates/atuin-ai/src/resume_tui/sidecar.rs @@ -52,10 +52,12 @@ impl SidecarSource { /// A picker row for `s`. fn row(&self, s: Session) -> SessionRow { let relation = match s.inferred_parent_kind() { + // A child of unknown kind is a newer build's: taken for a subagent, as group sizes + // take it. Some(ParentKind::Subagent) => Relation::Subagent, + None if s.parent.is_some() => Relation::Subagent, Some(ParentKind::Fork) => Relation::Fork, Some(ParentKind::Continuation) => Relation::Continuation, - None if s.parent.is_some() => Relation::Child, None => Relation::Root, }; // A session with no recorded host predates host tracking: it can only be this host's. @@ -206,12 +208,20 @@ mod tests { } } + /// A row of session `id`: like Claude Code's, an `agent-*` child is a subagent and any + /// other a fork. fn message(id: &str, parent: Option<&str>, role: Role, text: &str, minutes: i64) -> Message { + let kind = if id.starts_with("agent-") { + ParentKind::Subagent + } else { + ParentKind::Fork + }; Message::builder() .id(RecordId(uuid_v7())) .session(handle(id)) .source_id(SourceId::from(format!("{id}-{minutes}"))) .parent(parent.map(handle)) + .parent_kind(parent.map(|_| kind)) .timestamp(OffsetDateTime::UNIX_EPOCH + Duration::minutes(minutes)) .role(role) .content(vec![Content::Text(text.to_owned())]) diff --git a/crates/atuin-ai/src/resume_tui/source.rs b/crates/atuin-ai/src/resume_tui/source.rs index 608f11b41..2ce4a3539 100644 --- a/crates/atuin-ai/src/resume_tui/source.rs +++ b/crates/atuin-ai/src/resume_tui/source.rs @@ -37,22 +37,18 @@ pub struct SessionFilter { pub enum Relation { #[default] Root, - /// Spawned by its parent to do part of its work. + /// Spawned by its parent to do part of its work. Also a child of a kind this build does not + /// know (a newer build's). Subagent, /// Branches off its parent's conversation, or copies it. Fork, /// Carries its parent's conversation on under a new session (maybe in another harness). Continuation, - /// Has a parent, but no kind was recorded and the harness doesn't say which kind of child it - /// is. - Child, } impl Relation { /// Whether the picker shows sessions of this kind: roots, forks and continuations (which - /// resume like any session). Subagents never resume, so they are left out everywhere, with - /// the children whose kind is unknown: before capture recorded kinds, Codex and opencode - /// linked their spawned agents (by far the most of their children) and forks alike. + /// resume like any session). Subagents never resume, so they are left out everywhere. pub fn is_listed(self) -> bool { matches!(self, Self::Root | Self::Fork | Self::Continuation) } diff --git a/crates/atuin-ai/src/resume_tui/tests.rs b/crates/atuin-ai/src/resume_tui/tests.rs index 9e9330e68..f2f9462d7 100644 --- a/crates/atuin-ai/src/resume_tui/tests.rs +++ b/crates/atuin-ai/src/resume_tui/tests.rs @@ -2123,7 +2123,7 @@ async fn a_rebuild_in_progress_is_said_in_the_status_row() { assert!(!before.contains("rebuilding"), "{before}"); state.rebuilding = Some(Rebuilding { - progress: Some((1200, 5000)), + progress: (1200, 5000), }); let out = text(&render(&mut state, &s, 100, 30)); let line = "rebuilding the session index: 1200 of 5000 records; results may be incomplete"; diff --git a/crates/atuin-ai/src/tools/session/mod.rs b/crates/atuin-ai/src/tools/session/mod.rs index 05d168877..b8550b36a 100644 --- a/crates/atuin-ai/src/tools/session/mod.rs +++ b/crates/atuin-ai/src/tools/session/mod.rs @@ -160,20 +160,19 @@ fn elsewhere_note<'a>( /// A subagent: a fragment of the session that spawned it, only meaningful as part of it. Not any /// session with a parent: forks and continuations have one too, and those are sessions a person -/// ran. A parent of unknown kind (records captured before the kind was) counts as a subagent, -/// the cautious reading, since most sessions with a parent are subagents. +/// ran. A child of a kind this build does not know (a newer build's) is taken for one, as the +/// sidecar does. fn is_subagent(s: &Session) -> bool { - s.parent.is_some() - && !matches!(s.parent_kind, Some(ParentKind::Fork | ParentKind::Continuation)) + s.parent.is_some() && matches!(s.parent_kind, Some(ParentKind::Subagent) | None) } -/// How a session relates to its parent, phrased to precede the parent's id. +/// How a session relates to its parent, phrased to precede the parent's id. A kind this build +/// does not know reads as a subagent, as in [`is_subagent`]. const fn relation(kind: Option) -> &'static str { match kind { - Some(ParentKind::Subagent) => "subagent of", + Some(ParentKind::Subagent) | None => "subagent of", Some(ParentKind::Fork) => "forked from", Some(ParentKind::Continuation) => "continues from", - None => "started from", } } @@ -313,7 +312,6 @@ mod tests { #[rstest] #[case::subagent(Some(ParentKind::Subagent), true)] - #[case::unknown_kind(None, true)] #[case::fork(Some(ParentKind::Fork), false)] #[case::continuation(Some(ParentKind::Continuation), false)] fn subagents_are_children_not_known_to_be_forks( @@ -335,7 +333,6 @@ mod tests { #[case::subagent(Some(ParentKind::Subagent), "subagent of")] #[case::fork(Some(ParentKind::Fork), "forked from")] #[case::continuation(Some(ParentKind::Continuation), "continues from")] - #[case::unknown_kind(None, "started from")] fn relation_names_the_parent_kind(#[case] kind: Option, #[case] want: &str) { assert_eq!(relation(kind), want); } diff --git a/crates/atuin-ai/src/tools/session/read.rs b/crates/atuin-ai/src/tools/session/read.rs index 3a3ecf0f5..b23264051 100644 --- a/crates/atuin-ai/src/tools/session/read.rs +++ b/crates/atuin-ai/src/tools/session/read.rs @@ -34,24 +34,17 @@ const WINDOW_CHARS: usize = 20_000; /// volume, and a reader mostly needs what was asked, said and decided. A single-message read /// (`full`) cuts nothing per part; it is windowed as a whole instead. const TEXT_CHARS: usize = 2_000; -const THINKING_CHARS: usize = 600; const TOOL_INPUT_CHARS: usize = 300; const TOOL_RESULT_CHARS: usize = 400; -/// Text a harness wrote into the conversation (standing context such as AGENTS.md or a sandbox -/// policy, but also one-off notes like a subagent's report) gets a line, not its bulk, on a page. -const HARNESS_CHARS: usize = 200; /// How much of a message around a query match a filtered page shows, when the abridged message /// leaves the match out. const SNIPPET_CHARS: usize = 240; /// The labels a page puts on blocks, so a query match can tell them from what was said. -const HARNESS_LABEL: &str = "(harness)"; -const THINKING_LABEL: &str = "(thinking)"; const SUMMARY_LABEL: &str = "(summary of earlier conversation)"; const MODEL_ERROR_LABEL: &str = "(model error)"; const TOOL_ERROR_MARK: &str = "← error"; -const LABELS: [&str; 5] = - [HARNESS_LABEL, THINKING_LABEL, SUMMARY_LABEL, MODEL_ERROR_LABEL, TOOL_ERROR_MARK]; +const LABELS: [&str; 3] = [SUMMARY_LABEL, MODEL_ERROR_LABEL, TOOL_ERROR_MARK]; // Doc comments on the fields are the descriptions the model reads in the tool schema. #[derive(Debug, Clone, Deserialize, JsonSchema)] @@ -574,20 +567,7 @@ fn render_message(m: &Message, full: bool) -> Option<(String, String)> { } } (Block::Readable, Content::Text(text)) => { - // Only people and the model write user and assistant text; the parsers file - // what a harness wrote (Codex developer prompts, Claude Code reminders, subagent - // reports) under other roles. Often bulky and repeated every session, it gets a - // line on a page. - let text = text.trim(); - if full || matches!(m.role, Role::User | Role::Assistant) { - let _ = writeln!(body, "{}", clip(text, TEXT_CHARS, full)); - } else { - let _ = writeln!(body, "{HARNESS_LABEL} {}", one_line(text, HARNESS_CHARS)); - } - } - (Block::Readable, Content::Reasoning(text)) => { - let _ = - writeln!(body, "{THINKING_LABEL} {}", clip(text.trim(), THINKING_CHARS, full)); + let _ = writeln!(body, "{}", clip(text.trim(), TEXT_CHARS, full)); } // A compaction summary stands in for the conversation before it, so it is often the // best account of what an earlier stretch of a long session did. @@ -630,7 +610,11 @@ fn render_message(m: &Message, full: bool) -> Option<(String, String)> { } } } - (Block::Readable, Content::ReasoningSummary { .. } | Content::Other(_)) => {} + // Empty (see `classify`). + ( + Block::Readable, + Content::Reasoning(_) | Content::ReasoningSummary { .. } | Content::Other(_), + ) => {} } } if body.is_empty() { @@ -750,10 +734,7 @@ enum Block<'a> { fn classify(block: &Content) -> Block<'_> { match block { - Content::Text(text) - | Content::Reasoning(text) - | Content::Summary(text) - | Content::Error(text) => { + Content::Text(text) | Content::Summary(text) | Content::Error(text) => { if text.trim().is_empty() { Block::Empty } else { @@ -773,8 +754,11 @@ fn classify(block: &Content) -> Block<'_> { } } Content::ToolUse(_) | Content::ToolResult(_) | Content::Patch(_) => Block::Readable, - // Capture keeps only that reasoning happened, which says nothing to a reader. - Content::ReasoningSummary { .. } | Content::Other(_) => Block::Empty, + // Capture keeps only that reasoning happened, which says nothing to a reader, and never + // stores reasoning text or raw blocks. + Content::ReasoningSummary { .. } | Content::Reasoning(_) | Content::Other(_) => { + Block::Empty + } } } @@ -1034,30 +1018,6 @@ mod tests { assert!(one.contains("(model error) overloaded\ndetail"), "{one}"); } - #[rstest] - fn harness_text_gets_a_line_unless_full() { - let msgs = vec![ - text( - Role::Other("developer".to_owned()), - &format!("sandbox rules{}", " policy".repeat(2_000)), - ), - // The Codex parser files `` under the system role. - text(Role::System, "/x"), - text(Role::User, "fix this"), - ]; - let abridged = render(json!({"session_id": "abc"}), &msgs); - assert!(abridged.contains("#0 developer"), "{abridged}"); - assert!( - abridged.contains("(harness) sandbox rules"), - "{abridged}" - ); - assert!(abridged.len() < 2_000, "a long harness blob is one line: {}", abridged.len()); - assert!(abridged.contains("#2 user 00:00\nfix this"), "{abridged}"); - - let full = render(json!({"session_id": "abc", "limit": 1}), &msgs); - assert!(full.matches(" policy").count() >= 2_000, "a single-message read shows it all"); - } - #[rstest] fn a_page_ends_early_at_its_character_budget() { let msgs: Vec<_> = diff --git a/crates/atuin-client/Cargo.toml b/crates/atuin-client/Cargo.toml index a7e359acc..c6b441077 100644 --- a/crates/atuin-client/Cargo.toml +++ b/crates/atuin-client/Cargo.toml @@ -71,7 +71,6 @@ unicode-normalization = { workspace = true } typed-builder = { workspace = true } url = { workspace = true } uuid = { workspace = true } -xxhash-rust = { workspace = true } # encryption rusty_paserk = { workspace = true } diff --git a/crates/atuin-client/src/ai_session/database.rs b/crates/atuin-client/src/ai_session/database.rs index f1b36682c..d02d039ab 100644 --- a/crates/atuin-client/src/ai_session/database.rs +++ b/crates/atuin-client/src/ai_session/database.rs @@ -92,17 +92,15 @@ macro_rules! message_columns { }; } -/// Whether the session `c` is a subagent, or a child of unknown kind: what +/// Whether the session `c` is a subagent, or a child of unknown kind, as /// [`Session::inferred_parent_kind`] tells from its parent kind (`ParentKind` as -/// [`parent_kind_repr`] stores it), else from its harness and id, in SQL. Such a session is -/// grouped under its root but not counted in its size. +/// [`parent_kind_repr`] stores it), in SQL. Such a session is grouped under its root but not +/// counted in its size. /// /// [`parent_kind_repr`]: AiSessionDatabase::parent_kind_repr macro_rules! subagent_like { () => { - "(c.parent_session_id IS NOT NULL AND (c.parent_kind IS 0 OR (c.parent_kind IS NULL AND \ - c.parent_harness = c.harness AND (c.harness NOT IN (1, 5) OR (c.harness = 1 AND \ - substr(c.session_id, 1, 6) = 'agent-')))))" + "(c.parent_session_id IS NOT NULL AND (c.parent_kind IS 0 OR c.parent_kind IS NULL))" }; } @@ -626,7 +624,7 @@ impl AiSessionDatabase { .await?; Ok(row.map(|(at, digest)| Checkpoint { - at: u64::try_from(at).unwrap_or(0), + at: at.cast_unsigned(), digest: digest.cast_unsigned(), generation: 0, })) @@ -643,7 +641,9 @@ impl AiSessionDatabase { ) .bind(session.harness as i64) .bind(session.session.as_ref()) - .bind(i64::try_from(checkpoint.at).unwrap_or(i64::MAX)) + // Both are kept bit for bit: a checkpoint may use every bit of either (opencode's packs + // two places into `at`, its top bit set). + .bind(checkpoint.at.cast_signed()) .bind(checkpoint.digest.cast_signed()) .execute(self.db.pool()) .await?; @@ -1869,9 +1869,9 @@ mod tests { }) .await .unwrap(); - // A digest with its top bit set survives the signed column. + // An offset and a digest with their top bits set survive the signed columns. let latest = Checkpoint { - at: 4096, + at: 1 << 63 | 4096, digest: u64::MAX - 1, generation: 0, }; @@ -2428,13 +2428,9 @@ mod tests { let session = sample_handle(); let parent = handle(HarnessKind::ClaudeCode, "parent"); let mut first = message_in(&session, 0, "copied from the parent"); - first.parent = Some(parent.clone()); + first.parent = Some(parent); first.parent_kind = Some(ParentKind::Fork); db.append(&first).await.unwrap(); - // A record from an older build carries the parent but not the kind: it must not erase it. - let mut older = message_in(&session, 1, "carried on"); - older.parent = Some(parent); - db.append(&older).await.unwrap(); let s = db.get_session(&session).await.unwrap().unwrap(); assert_eq!(s.parent_kind, Some(ParentKind::Fork)); @@ -3210,11 +3206,19 @@ mod tests { // --- grouping ------------------------------------------------------------------------------- - /// A row of session `id` with parent `parent`, at `seconds`, saying `text`. + /// A row of session `id` with parent `parent`, at `seconds`, saying `text`. Like Claude + /// Code's, an `agent-*` child is a subagent and any other a fork. fn tree_row(id: &str, parent: Option<&str>, seconds: i64, text: &str) -> Message { let mut m = message_in(&handle(HarnessKind::ClaudeCode, id), seconds, text); m.source_id = SourceId::from(format!("{id}-{seconds}")); m.parent = parent.map(|p| handle(HarnessKind::ClaudeCode, p)); + m.parent_kind = m.parent.as_ref().map(|_| { + if id.starts_with("agent-") { + ParentKind::Subagent + } else { + ParentKind::Fork + } + }); m } @@ -3306,19 +3310,17 @@ mod tests { /// Sessions continued in other harnesses group under the session /// they continue, in whichever order their rows arrive (as a reprojection from the synced - /// records replays them), and are its continuations: by the kind capture recorded, else - /// (in records from before kinds were) because they name another harness's session. + /// records replays them), and are its continuations. #[rstest] #[tokio::test] async fn continuations_in_other_harnesses_group_under_the_original( #[values(false, true)] reversed: bool, - #[values(None, Some(ParentKind::Continuation))] kind: Option, ) { let row = |harness, id: &str, parent: Option<(HarnessKind, &str)>, seconds| { let mut m = message_in(&handle(harness, id), seconds, "words"); m.source_id = SourceId::from(format!("{id}-{seconds}")); m.parent = parent.map(|(h, p)| handle(h, p)); - m.parent_kind = m.parent.as_ref().and(kind); + m.parent_kind = m.parent.as_ref().map(|_| ParentKind::Continuation); m }; let mut rows = vec![ @@ -3340,8 +3342,7 @@ mod tests { for (harness, id) in [(HarnessKind::ClaudeCode, "in-claude"), (HarnessKind::Pi, "in-pi")] { let s = db.get_session(&handle(harness, id)).await.unwrap().unwrap(); assert_eq!(s.group(), &original, "{id}"); - assert_eq!(s.parent_kind, kind, "{id}"); - assert_eq!(s.inferred_parent_kind(), Some(ParentKind::Continuation), "{id}"); + assert_eq!(s.parent_kind, Some(ParentKind::Continuation), "{id}"); } let children: Vec = db .children(&original) @@ -3873,9 +3874,9 @@ mod tests { ]); } - /// Where capture recorded how a child relates to its parent, that decides, whatever the - /// harness or id suggest: a Codex subagent is not counted in its root's group size, a Codex - /// fork is, and so is a Claude Code child named like a subagent that is a fork. + /// How capture recorded a child relates to its parent decides, whatever its harness or id: a + /// Codex subagent is not counted in its root's group size, a Codex fork is, and so is a + /// Claude Code child named like a subagent that is a fork. #[rstest] #[case::codex_subagent(HarnessKind::Codex, "t2", ParentKind::Subagent)] #[case::codex_fork(HarnessKind::Codex, "t2", ParentKind::Fork)] @@ -3903,36 +3904,24 @@ mod tests { assert_eq!(groups[0].child_count, u64::from(kind != ParentKind::Subagent)); } - /// Without a recorded kind, a group's size (counted in SQL) counts a child exactly when - /// [`Session::inferred_parent_kind`] makes it a fork or a continuation. + /// A child of a kind this build does not know (a newer build's) is grouped under its root + /// but, like a subagent, not counted in its size. #[rstest] - #[case::claude_subagent(HarnessKind::ClaudeCode, "agent-z", HarnessKind::ClaudeCode)] - #[case::claude_fork(HarnessKind::ClaudeCode, "sub", HarnessKind::ClaudeCode)] - #[case::pi_branch(HarnessKind::Pi, "sub", HarnessKind::Pi)] - #[case::codex_child(HarnessKind::Codex, "sub", HarnessKind::Codex)] - #[case::opencode_child(HarnessKind::Opencode, "ses_2", HarnessKind::Opencode)] - #[case::continued_elsewhere(HarnessKind::Codex, "agent-z", HarnessKind::ClaudeCode)] #[tokio::test] - async fn without_a_kind_the_group_size_agrees_with_the_inferred_kind( - #[case] harness: HarnessKind, - #[case] child: &str, - #[case] parent_harness: HarnessKind, - ) { + async fn a_child_of_unknown_kind_is_not_counted() { let db = AiSessionDatabase::in_memory().await.unwrap(); - let root = handle(parent_harness, "root"); + let root = handle(HarnessKind::Codex, "root"); db.append(&message_in(&root, 0, "root words")).await.unwrap(); - let mut m = message_in(&handle(harness, child), 1, "child words"); - m.parent = Some(root); + let mut m = message_in(&handle(HarnessKind::Codex, "child"), 1, "child words"); + m.parent = Some(root.clone()); db.append(&m).await.unwrap(); - let stored = db.get_session(&handle(harness, child)).await.unwrap().unwrap(); - let counted = matches!( - stored.inferred_parent_kind(), - Some(ParentKind::Fork | ParentKind::Continuation) - ); + let stored = db.get_session(&handle(HarnessKind::Codex, "child")).await.unwrap().unwrap(); + assert_eq!(stored.inferred_parent_kind(), None); + assert_eq!(stored.root, Some(root)); let groups = db.list_sessions(&roots_only()).await.unwrap(); assert_eq!(groups.len(), 1); - assert_eq!(groups[0].child_count, u64::from(counted)); + assert_eq!(groups[0].child_count, 0); } /// A match in a nested child is its root's, snippet included. @@ -4206,15 +4195,19 @@ mod tests { async fn an_atuin_id_or_its_prefix_finds_a_session(#[case] len: usize, #[case] found: usize) { let db = AiSessionDatabase::in_memory().await.unwrap(); let wanted = handle(HarnessKind::ClaudeCode, "wanted"); - // Same start, so the ids share their timestamp digits. - let started = OffsetDateTime::from_unix_timestamp(1_790_000_000).unwrap(); - for session in [&wanted, &handle(HarnessKind::Pi, "other")] { + // The same start, so the ids share their timestamp digits (the first 12). + let id = |n: u128| { + AtuinSessionId::from(Uuid::from_u128( + 0x0199_6dc4_0800_7000_8000_0000_0000_0000 | n << 64, + )) + }; + for (session, n) in [(&wanted, 1), (&handle(HarnessKind::Pi, "other"), 2)] { let mut msg = message_in(session, 1_790_000_000, "words"); - msg.atuin_id = Some(session.atuin_id(started)); + msg.atuin_id = Some(id(n)); db.append(&msg).await.unwrap(); } - let id = wanted.atuin_id(started); + let id = id(1); let sessions = db.sessions_with_id_prefix(&id.to_string()[..len]).await.unwrap(); assert_eq!(sessions.len(), found); assert!(sessions.iter().any(|s| s.handle == wanted)); @@ -4243,36 +4236,6 @@ mod tests { ); } - #[rstest] - #[tokio::test] - async fn rows_without_the_id_stand_for_a_derived_one(#[values(false, true)] fixed_first: bool) { - let db = AiSessionDatabase::in_memory().await.unwrap(); - let session = sample_handle(); - let old = message_in(&session, 5, "one"); - db.append(&old).await.unwrap(); - assert_eq!( - db.get_session(&session).await.unwrap().unwrap().atuin_id, - session.atuin_id(old.timestamp) - ); - - let mut fixed = message_in(&session, 6, "two"); - fixed.atuin_id = Some(AtuinSessionId::from(Uuid::from_u128(1))); - let older = message_in(&session, 0, "three"); - let rows = if fixed_first { - [fixed, older] - } else { - [older, fixed] - }; - for row in &rows { - db.append(row).await.unwrap(); - } - - assert_eq!( - db.get_session(&session).await.unwrap().unwrap().atuin_id, - AtuinSessionId::from(Uuid::from_u128(1)) - ); - } - #[rstest] #[tokio::test] async fn a_duplicate_row_settles_on_the_smaller_id() { @@ -4302,7 +4265,10 @@ mod tests { handle(HarnessKind::Codex, "child"), handle(HarnessKind::Codex, "grandchild"), ); - let id = |s: &HarnessSession| s.atuin_id(OffsetDateTime::UNIX_EPOCH); + let id = |s: &HarnessSession| { + let n = ["root", "child", "grandchild"].iter().position(|n| *n == s.session.as_ref()); + AtuinSessionId::from(Uuid::from_u128(n.unwrap() as u128 + 1)) + }; let store = |session: &HarnessSession, parent: Option<&HarnessSession>, index| { let mut m = message_in(session, index, "words"); m.parent = parent.cloned(); diff --git a/crates/atuin-client/src/ai_session/database/write.rs b/crates/atuin-client/src/ai_session/database/write.rs index 48c9f70e4..0420fc6f7 100644 --- a/crates/atuin-client/src/ai_session/database/write.rs +++ b/crates/atuin-client/src/ai_session/database/write.rs @@ -19,7 +19,7 @@ use super::watermark::BUMP_GENERATION; use super::{ AiSessionDatabase, Appended, CallRow, Claimant, DbError, Generation, SessionKey, Tokens, }; -use crate::ai_session::Message; +use crate::ai_session::{AtuinSessionId, Message}; /// Long enough to wait out the other connection's write, as sqlx's default is. const BUSY_TIMEOUT: Duration = Duration::from_secs(5); @@ -270,9 +270,9 @@ impl AiSessionDatabase { let cwd = msg.cwd.as_ref().map(|p| p.to_string_lossy().into_owned()); let host = msg.host.map(Self::host_repr); - // v0 rows carry no id. The earliest row's derived id is the smallest, so MIN settles on - // the one from the session's start. - let atuin_id = msg.atuin_id.unwrap_or_else(|| msg.session.atuin_id(msg.timestamp)); + // Capture gives every row it stores its session's id. A row appended without one (only + // ever directly, as tests do) gets a fresh one, which MIN settles like any other. + let atuin_id = msg.atuin_id.unwrap_or_else(|| AtuinSessionId::mint(msg.timestamp)); let atuin_id = atuin_id.as_bytes().as_slice(); let before = Self::session_key(conn, harness, session_id)?; diff --git a/crates/atuin-client/src/ai_session/model.rs b/crates/atuin-client/src/ai_session/model.rs index f049b5786..a6e1b2469 100644 --- a/crates/atuin-client/src/ai_session/model.rs +++ b/crates/atuin-client/src/ai_session/model.rs @@ -132,49 +132,6 @@ impl std::str::FromStr for AtuinSessionId { } } -impl HarnessSession { - /// The id for a session whose records predate stored ids (v0), so every host derives the same - /// one. A native UUIDv7 (Codex) is reused. Otherwise it's a UUIDv7 at the session's start - /// (from an opencode `ses_` id, else `started_at`), its other bits the little-endian - /// xxh3-128 of `"{harness repr}:{native id}"`. - #[must_use] - pub fn atuin_id(&self, started_at: OffsetDateTime) -> AtuinSessionId { - let native = self.session.as_ref(); - - if let Ok(id) = Uuid::parse_str(native) - && id.get_version() == Some(uuid::Version::SortRand) - && id.get_variant() == uuid::Variant::RFC4122 - { - return AtuinSessionId(id); - } - - let started = i64::try_from(started_at.unix_timestamp_nanos() / 1_000_000).unwrap_or(0); - let millis = match self.harness { - HarnessKind::Opencode => opencode_millis(native, started), - _ => None, - } - .unwrap_or(started); - let hash = - xxhash_rust::xxh3::xxh3_128(format!("{}:{native}", self.harness as u8).as_bytes()) - .to_le_bytes(); - let mut bits = [0; 10]; - bits.copy_from_slice(&hash[..10]); - let millis = u64::try_from(millis).unwrap_or(0); - AtuinSessionId(Builder::from_unix_timestamp_millis(millis, &bits).into_uuid()) - } -} - -/// opencode's `ses_` ids hold the inverted low 48 bits of `millis * 0x1000 + counter`, so only -/// the time's low 36 bits survive; the rest come from the match nearest `started`. -fn opencode_millis(id: &str, started: i64) -> Option { - const WRAP: i64 = 1 << 36; - - let encoded = u64::from_str_radix(id.strip_prefix("ses_")?.get(..12)?, 16).ok()?; - let low = i64::try_from((!encoded & 0xffff_ffff_ffff) >> 12).ok()?; - let near = started - started.rem_euclid(WRAP) + low; - [near - WRAP, near, near + WRAP].into_iter().min_by_key(|t| (t - started).abs()) -} - /// Records encode these fields by position: adding, removing or reordering one needs a new /// [record version](super::AiSessionRecord::VERSION). #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TypedBuilder)] @@ -184,11 +141,11 @@ pub struct Message { pub source_id: SourceId, #[builder(default)] pub parent: Option, - /// How this session relates to [`Self::parent`]. Absent in records written before it was - /// captured, and whenever the harness did not say. A kind this build does not know (a newer - /// build's) reads as absent, rather than failing the whole record. + /// How this session relates to [`Self::parent`]. Absent whenever the harness did not say. A + /// kind this build does not know (a newer build's) reads as absent, rather than failing the + /// whole record. #[builder(default)] - #[serde(default, deserialize_with = "known_parent_kind")] + #[serde(deserialize_with = "known_parent_kind")] pub parent_kind: Option, #[builder(default)] pub parent_source_id: Option, @@ -213,26 +170,23 @@ pub struct Message { /// survives a reproject from the synced record store (records carry messages only, not the /// separate `Started` metadata). `None` once a title is cleared: the newest row decides. #[builder(default)] - #[serde(default)] pub session_title: Option, /// Where [`Self::session_title`] came from, so a resumed capture keeps ranking it. #[builder(default)] - #[serde(default)] pub session_title_source: Option, /// The title this row's own line set or cleared. Replayed on resume, so every source's /// title is known again and a cleared one can fall back to the next. #[builder(default)] - #[serde(default)] pub title_change: Option, /// The model call this row came from, unique within the harness and the same in every /// session a harness copies the row into. Groups the rows one response is split into, so /// their usage counts once. #[builder(default)] - #[serde(default)] pub turn_id: Option, - /// `None` until capture sets it, and in v0 records. + /// The session's id. `None` until capture sets it, which it does before storing the row, + /// and in rows read back from the sidecar or over the daemon's API, which keep it on the + /// session. #[builder(default)] - #[serde(default)] pub atuin_id: Option, /// The host that captured this row. Never part of the record body: the record envelope /// already carries it, so a reproject takes it from there and live capture from the local @@ -243,8 +197,7 @@ pub struct Message { } /// [`Message::parent_kind`] as a record holds it: `None` for a kind this build does not know, -/// so a newer build can add kinds without its records failing to decode here, and for a value -/// that names no kind at all (some development builds' records hold another field there). +/// so a newer build can add kinds without its records failing to decode here. fn known_parent_kind<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, @@ -252,23 +205,12 @@ where use serde::de::IntoDeserializer; use serde::de::value::{Error, StringDeserializer}; - /// A kind's name, or anything else a record may hold there. - #[derive(Deserialize)] - #[serde(untagged)] - enum Raw { - Name(String), - Other(serde::de::IgnoredAny), - } - // Parse the name with `ParentKind`'s own `Deserialize`, so a new variant is known here // without listing it twice. - Ok(match Option::::deserialize(deserializer)? { - Some(Raw::Name(name)) => { - let name: StringDeserializer = name.into_deserializer(); - ParentKind::deserialize(name).ok() - } - Some(Raw::Other(_)) | None => None, - }) + Ok(Option::::deserialize(deserializer)?.and_then(|name| { + let name: StringDeserializer = name.into_deserializer(); + ParentKind::deserialize(name).ok() + })) } impl From for RehydrateMessage { @@ -398,39 +340,21 @@ impl Session { self.root.as_ref().unwrap_or(&self.handle) } - /// How this session relates to its parent: the [`Self::parent_kind`] capture recorded, else - /// the kind its harness and id tell. A parentless copy of another session ([`Self::copy_of`]) - /// is a [fork](ParentKind::Fork) of it. + /// How this session relates to its parent: the [`Self::parent_kind`] capture recorded. A + /// parentless copy of another session ([`Self::copy_of`]) is a [fork](ParentKind::Fork) of + /// it. /// - /// `None` for a session with no parent (and not a copy), and for a child whose kind cannot be - /// told. Only forks and continuations count toward a group's [`Self::child_count`]: subagents - /// are fragments of their parent's work, and so are taken to be the children of unknown kind - /// (most of Codex's and opencode's are spawned agents). + /// `None` for a session with no parent (and not a copy), and for a child of a kind this build + /// does not know (a newer build's). Only forks and continuations count toward a group's + /// [`Self::child_count`]: subagents are fragments of their parent's work, and so are taken to + /// be the children of unknown kind. #[must_use] pub fn inferred_parent_kind(&self) -> Option { match &self.parent { - Some(parent) => self.parent_kind.or_else(|| self.guess_parent_kind(parent)), + Some(_) => self.parent_kind, None => self.copy_of.as_ref().map(|_| ParentKind::Fork), } } - - /// The kind of a parent recorded without one (by a build from before kinds were captured), as - /// well as the harness and id tell: another harness's session is only ever named by a - /// continuation (`atuin ai resume --in`), and a Claude Code `agent-*` session is a subagent - /// while its other children are forks, as are all of pi's. Codex and opencode link subagents - /// and forks alike: `None`. - fn guess_parent_kind(&self, parent: &HarnessSession) -> Option { - if parent.harness != self.handle.harness { - return Some(ParentKind::Continuation); - } - match self.handle.harness { - HarnessKind::ClaudeCode if self.handle.session.as_ref().starts_with("agent-") => { - Some(ParentKind::Subagent) - } - HarnessKind::ClaudeCode | HarnessKind::Pi => Some(ParentKind::Fork), - _ => None, - } - } } /// Which sessions a listing or search returns. An absent field is not a filter; every present @@ -518,34 +442,11 @@ pub struct MatchedSession { #[cfg(test)] mod tests { use atuin_domain::record::RecordVersion; - use proptest::prelude::*; use rstest::rstest; use super::*; use crate::ai_session::{AiSessionRecord, DecodeError}; - fn arb_content() -> impl Strategy { - "[a-zA-Z0-9 ]{0,16}".prop_map(Content::Text) - } - - fn arb_message() -> impl Strategy { - ("[a-z0-9]{1,8}", "[a-z0-9]{1,8}", proptest::collection::vec(arb_content(), 0..3)).prop_map( - |(native_session, source_id, content)| { - Message::builder() - .id(RecordId(atuin_common::utils::uuid_v7())) - .session(HarnessSession { - harness: HarnessKind::ClaudeCode, - session: NativeSessionId::from(native_session), - }) - .source_id(SourceId::from(source_id)) - .timestamp(OffsetDateTime::UNIX_EPOCH) - .role(Role::User) - .content(content) - .build() - }, - ) - } - /// Every kind with harness tools maps back to itself; the rest have none to resume with. #[rstest] #[case::claude(HarnessKind::ClaudeCode, true)] @@ -565,8 +466,6 @@ mod tests { } } - /// Records are named-field msgpack, so a host whose build predates a field (here `turn_id`) - /// still decodes a newer host's records, skipping the field it does not know. fn record_message() -> Message { Message::builder() .id(RecordId(Uuid::from_u128(1))) @@ -594,38 +493,10 @@ mod tests { } #[rstest] - fn a_v0_record_still_decodes() { - #[derive(Serialize)] - struct V0<'a> { - id: RecordId, - session: &'a HarnessSession, - source_id: &'a SourceId, - timestamp: OffsetDateTime, - role: Role, - content: &'a [Content], - turn_id: Option<&'a str>, - } - let msg = record_message(); - let mut record = vec![0]; - record.extend( - rmp_serde::to_vec_named(&V0 { - id: msg.id, - session: &msg.session, - source_id: &msg.source_id, - timestamp: msg.timestamp, - role: msg.role.clone(), - content: &msg.content, - turn_id: msg.turn_id.as_deref(), - }) - .unwrap(), - ); - - let AiSessionRecord::Message(back) = - AiSessionRecord::deserialize(&record, &RecordVersion::V0).unwrap(); - assert_eq!(back, Message { - atuin_id: None, - ..msg - }); + fn a_v0_record_is_no_longer_read() { + let record = AiSessionRecord::Message(record_message()).serialize(); + let err = AiSessionRecord::deserialize(&record, &RecordVersion::V0).unwrap_err(); + assert!(matches!(err, DecodeError::Retired(RecordVersion::V0))); } #[rstest] @@ -675,98 +546,25 @@ mod tests { assert_eq!(back.parent_kind, kind); } - /// A kind this build does not know (a newer build's), or a value naming none, reads as no - /// kind: the rest of the record still decodes. + /// A kind this build does not know (a newer build's) reads as no kind: the rest of the record + /// still decodes. #[rstest] - #[case::a_newer_kind("Handoff")] - #[case::not_a_kind("784ad9be-9b3d-48e2-a3d7-a7cf227fd86e")] - fn an_unknown_parent_kind_reads_as_none(#[case] kind: &str) { - /// A record as a build that writes `kind` would. - #[derive(Serialize)] - struct Written<'a> { - id: RecordId, - session: HarnessSession, - source_id: SourceId, - parent_kind: &'a str, - timestamp: OffsetDateTime, - role: Role, - content: Vec, - } - let written = Written { - id: RecordId(atuin_common::utils::uuid_v7()), - session: HarnessSession { - harness: HarnessKind::ClaudeCode, - session: NativeSessionId::from("s".to_owned()), - }, - source_id: SourceId::from("x".to_owned()), - parent_kind: kind, - timestamp: OffsetDateTime::UNIX_EPOCH, - role: Role::User, - content: vec![], - }; - let body = rmp_serde::to_vec_named(&written).unwrap(); - let back: Message = rmp_serde::from_slice(&body).unwrap(); - assert_eq!(back.parent_kind, None); - assert_eq!(back.source_id, written.source_id); - assert_eq!(back.atuin_id, None); + fn an_unknown_parent_kind_reads_as_none() { + let msg = kinded(Some(ParentKind::Fork)); + let body = rmp_serde::to_vec(&msg).unwrap(); + let (known, newer) = + (rmp_serde::to_vec("Fork").unwrap(), rmp_serde::to_vec("Handoff").unwrap()); + let at = body.windows(known.len()).position(|w| w == known).unwrap(); + let written = [&body[..at], &newer, &body[at + known.len()..]].concat(); + + let back: Message = rmp_serde::from_slice(&written).unwrap(); + assert_eq!(back, Message { + parent_kind: None, + ..msg + }); } - /// Without a recorded kind (records from before capture recorded one), the kind is told from - /// the harness and id. - #[rstest] - #[case::no_parent(HarnessKind::ClaudeCode, "s", false, None)] - #[case::claude_subagent(HarnessKind::ClaudeCode, "agent-a1", true, Some(ParentKind::Subagent))] - #[case::claude_fork(HarnessKind::ClaudeCode, "0b3c", true, Some(ParentKind::Fork))] - #[case::pi_branch(HarnessKind::Pi, "s", true, Some(ParentKind::Fork))] - #[case::codex_child(HarnessKind::Codex, "s", true, None)] - #[case::opencode_child(HarnessKind::Opencode, "ses_1", true, None)] - fn without_a_kind_it_is_told_from_the_harness_and_id( - #[case] harness: HarnessKind, - #[case] id: &str, - #[case] has_parent: bool, - #[case] expected: Option, - ) { - let handle = |id: &str| HarnessSession { - harness, - session: NativeSessionId::from(id.to_owned()), - }; - let session = Session::builder() - .handle(handle(id)) - .parent(has_parent.then(|| handle("parent"))) - .started_at(OffsetDateTime::UNIX_EPOCH) - .updated_at(OffsetDateTime::UNIX_EPOCH) - .usage(Usage::default()) - .build(); - assert_eq!(session.inferred_parent_kind(), expected); - } - - /// A session continued in another harness (`atuin ai resume --in`) is a continuation of the - /// one it continues, whatever its own harness calls its children, even in records from - /// before capture recorded the kind. - #[rstest] - fn a_continuation_in_another_harness_is_a_continuation( - #[values(HarnessKind::Codex, HarnessKind::Opencode, HarnessKind::ClaudeCode)] - harness: HarnessKind, - #[values(None, Some(ParentKind::Continuation))] kind: Option, - ) { - let session = Session::builder() - .handle(HarnessSession { - harness, - session: NativeSessionId::from("agent-new".to_owned()), - }) - .parent(Some(HarnessSession { - harness: HarnessKind::Pi, - session: NativeSessionId::from("original".to_owned()), - })) - .parent_kind(kind) - .started_at(OffsetDateTime::UNIX_EPOCH) - .updated_at(OffsetDateTime::UNIX_EPOCH) - .usage(Usage::default()) - .build(); - assert_eq!(session.inferred_parent_kind(), Some(ParentKind::Continuation)); - } - - /// A recorded kind decides, whatever the harness and id suggest. + /// A recorded kind is the session's kind, whatever its harness and id. #[rstest] #[case::codex_subagent(HarnessKind::Codex, "s", ParentKind::Subagent)] #[case::codex_fork(HarnessKind::Codex, "s", ParentKind::Fork)] @@ -817,25 +615,6 @@ mod tests { } } - #[rstest] - fn message_msgpack_roundtrips() { - proptest!(|(m in arb_message())| { - let bytes = rmp_serde::to_vec_named(&m).unwrap(); - let back: Message = rmp_serde::from_slice(&bytes).unwrap(); - prop_assert_eq!(m, back); - }); - } - - /// From a real opencode export. - const OPENCODE: (&str, i64) = ("ses_f2b668796ffe7eBzqVGu7k59OB", 1_790_273_222_761); - - fn session(harness: HarnessKind, id: &str) -> HarnessSession { - HarnessSession { - harness, - session: NativeSessionId::from(id.to_owned()), - } - } - fn at(millis: i64) -> OffsetDateTime { OffsetDateTime::from_unix_timestamp_nanos(i128::from(millis) * 1_000_000).unwrap() } @@ -844,18 +623,6 @@ mod tests { i64::try_from(Uuid::from(id).as_u128() >> 80).unwrap() } - #[rstest] - #[case::claude(HarnessKind::ClaudeCode, "0b7c2a7e-3f53-4b8e-9f43-6c1f6f0f2a10")] - #[case::codex_v4(HarnessKind::Codex, "5e0f2c1a-6a5b-4c3d-8e9f-0a1b2c3d4e5f")] - #[case::opencode(HarnessKind::Opencode, OPENCODE.0)] - #[case::pi(HarnessKind::Pi, "pi-session")] - #[case::copilot(HarnessKind::Copilot, "")] - fn the_id_is_a_v7(#[case] harness: HarnessKind, #[case] id: &str) { - let id = session(harness, id).atuin_id(at(1_790_000_000_000)); - assert_eq!(Uuid::from(id).get_version(), Some(uuid::Version::SortRand)); - assert_eq!(Uuid::from(id).get_variant(), uuid::Variant::RFC4122); - } - #[rstest] fn a_minted_id_is_a_fresh_v7_at_the_start() { let (a, b) = (AtuinSessionId::mint(at(1_234)), AtuinSessionId::mint(at(1_234))); @@ -865,72 +632,6 @@ mod tests { assert_eq!(Uuid::from(a).get_variant(), uuid::Variant::RFC4122); } - #[rstest] - fn the_id_is_stable_and_names_one_session() { - let started = at(1_790_000_000_000); - let claude = session(HarnessKind::ClaudeCode, "s"); - assert_eq!(claude.atuin_id(started), claude.atuin_id(started)); - assert_ne!( - claude.atuin_id(started), - session(HarnessKind::ClaudeCode, "t").atuin_id(started) - ); - // The same native id in another harness is another session. - assert_ne!(claude.atuin_id(started), session(HarnessKind::Pi, "s").atuin_id(started)); - } - - #[rstest] - fn a_native_v7_is_the_id() { - let native = "01a0d147-e745-7ae2-8000-0123456789ab"; - let id = session(HarnessKind::Codex, native).atuin_id(at(0)); - assert_eq!(Uuid::from(id).to_string(), native); - } - - #[rstest] - fn any_native_id_gives_a_v7() { - proptest!(|(native in any::(), harness in 0u8..6)| { - let harness = match harness { - 0 => HarnessKind::Unknown, - 1 => HarnessKind::ClaudeCode, - 2 => HarnessKind::Codex, - 3 => HarnessKind::Copilot, - 4 => HarnessKind::Opencode, - _ => HarnessKind::Pi, - }; - let native = Uuid::from_u128(native).to_string(); - let id = session(harness, &native).atuin_id(at(1_790_000_000_000)); - prop_assert_eq!(Uuid::from(id).get_version(), Some(uuid::Version::SortRand)); - prop_assert_eq!(Uuid::from(id).get_variant(), uuid::Variant::RFC4122); - }); - } - - #[rstest] - fn a_later_start_moves_only_the_timestamp() { - let s = session(HarnessKind::ClaudeCode, "s"); - let (full, partial) = (s.atuin_id(at(1_000)), s.atuin_id(at(9_000))); - assert_eq!((millis_of(full), millis_of(partial)), (1_000, 9_000)); - assert_eq!(Uuid::from(full).as_u128() << 48, Uuid::from(partial).as_u128() << 48); - } - - #[rstest] - #[case::at_creation(0)] - #[case::a_partial_capture(3_600_000)] - #[case::clock_skew(-5_000)] - #[case::most_of_a_year_off(300 * 86_400_000)] - #[case::most_of_a_year_early(-300 * 86_400_000)] - fn an_opencode_id_gives_its_own_time(#[case] off_by: i64) { - let (native, created) = OPENCODE; - let id = session(HarnessKind::Opencode, native).atuin_id(at(created + off_by)); - assert_eq!(millis_of(id), created); - } - - #[rstest] - #[case::not_hex("ses_zzzzzzzzzzzzabc")] - #[case::too_short("ses_f2b6")] - fn an_opencode_id_without_a_time_uses_the_start(#[case] native: &str) { - let id = session(HarnessKind::Opencode, native).atuin_id(at(1_234)); - assert_eq!(millis_of(id), 1_234); - } - #[rstest] #[case::empty("", true)] #[case::odd_length("01a0d147e", true)] diff --git a/crates/atuin-client/src/ai_session/store.rs b/crates/atuin-client/src/ai_session/store.rs index 163332119..1ce1c1b14 100644 --- a/crates/atuin-client/src/ai_session/store.rs +++ b/crates/atuin-client/src/ai_session/store.rs @@ -6,7 +6,7 @@ use atuin_domain::record::{ DecryptedData, Host, HostId, Record, RecordId, RecordIdx, RecordSeriesKey, RecordTag, RecordVersion, }; -use tracing::warn; +use tracing::{debug, warn}; use typed_builder::TypedBuilder; use crate::ai_session::Message; @@ -48,6 +48,9 @@ pub enum DecodeError { UnknownKind(u8), #[error("unknown ai-session record version {}", .0.as_str())] UnknownVersion(RecordVersion), + /// Only written by builds from before records stored the session's id. + #[error("ai-session record version {} is no longer read", .0.as_str())] + Retired(RecordVersion), #[error("failed to decode ai-session record body: {0}")] Body(#[from] rmp_serde::decode::Error), #[error("failed to decompress ai-session record body: {0}")] @@ -81,9 +84,9 @@ pub enum BuildError { impl AiSessionRecord { const MESSAGE_KIND: u8 = 0; - /// v0: named msgpack, without [`Message::atuin_id`]. v1: positional, so any change to - /// [`Message`]'s fields needs a new version. v2: v1's body, compressed with zstd (captured - /// tool payloads compress about 2.4 times). + /// v1: positional msgpack, so any change to [`Message`]'s fields needs a new version. v2: + /// v1's body, compressed with zstd (captured tool payloads compress about 2.4 times). v0 + /// (named msgpack, without [`Message::atuin_id`]) is no longer read. pub const VERSION: RecordVersion = RecordVersion::V2; /// A kind byte, then the zstd-compressed positional msgpack body. @@ -108,11 +111,9 @@ impl AiSessionRecord { return Err(DecodeError::UnknownKind(kind)); } - // from_slice reads a map by name (v0) or an array by position (v1, v2). match version { - RecordVersion::V0 | RecordVersion::V1 => { - Ok(Self::Message(rmp_serde::from_slice(body)?)) - } + RecordVersion::V0 => Err(DecodeError::Retired(RecordVersion::V0)), + RecordVersion::V1 => Ok(Self::Message(rmp_serde::from_slice(body)?)), RecordVersion::V2 => { let body = zstd::stream::decode_all(body).map_err(DecodeError::Decompress)?; Ok(Self::Message(rmp_serde::from_slice(&body)?)) @@ -177,6 +178,8 @@ impl AiSessionStore { // series' later records on each reprojection, but never stops them projecting. // - One that decrypts but whose kind or version this build does not know is held until // an upgrade. + // - One of a version this build no longer reads (from before records stored the + // session's id) is skipped: no later build reads it either. // - One that decrypts to a known kind but fails to decode is skipped: decryption // authenticates it, so it is exactly what its writer wrote, and no retry reads it. let decrypted = match record.decrypt(key) { @@ -195,6 +198,10 @@ impl AiSessionStore { warn!(?err, id = %id.0, "unknown ai-session record kind or version, holding it back"); return Ok(Projected::Held); } + Err(DecodeError::Retired(version)) => { + debug!(id = %id.0, version = version.as_str(), "skipping retired ai-session record"); + return Ok(Projected::Skipped); + } Err(err) => { warn!(?err, id = %id.0, "failed to deserialize ai-session record, skipping"); return Ok(Projected::Skipped); @@ -1360,7 +1367,7 @@ mod tests { let record = Record::builder() .id(RecordId(atuin_common::utils::uuid_v7())) .host(Host::new(s.host_id)) - .version(RecordVersion::V0) + .version(AiSessionRecord::VERSION) .tag(RecordTag::AiSession) .idx(1) .data(DecryptedData(vec![99])) @@ -1498,7 +1505,7 @@ mod tests { let record = Record::builder() .id(RecordId(atuin_common::utils::uuid_v7())) .host(Host::new(s.host_id)) - .version(RecordVersion::V0) + .version(RecordVersion::V1) .tag(RecordTag::AiSession) .idx(1) // A message kind, then a msgpack byte that is never valid. @@ -1513,6 +1520,35 @@ mod tests { assert_eq!(mark(&db, &s).await.map(|m| m.idx), Some(3)); } + /// A v0 record (from before records stored the session's id) is passed over, not held: it + /// would hold the watermark forever. + #[rstest] + #[tokio::test] + async fn a_v0_record_is_skipped() { + let store = SqliteStore::in_memory(test_local_timeout()).await.unwrap(); + let [s] = <[_; 1]>::try_from(writers(&store, 1)).ok().unwrap(); + let handle = sample_handle(); + push_range(&s, &handle, 0..1).await; + let msg = message_in(&session_named("v0"), 1, "text"); + let record = Record::builder() + .id(msg.id) + .host(Host::new(s.host_id)) + .version(RecordVersion::V0) + .tag(RecordTag::AiSession) + .idx(1) + .data(DecryptedData(AiSessionRecord::Message(msg).serialize())) + .build(); + store.push(&record.encrypt(&key())).await.unwrap(); + push_range(&s, &handle, 1..3).await; + + let db = AiSessionDatabase::in_memory().await.unwrap(); + assert_eq!(s.reproject(&db).await.unwrap().replayed, 4); + assert_eq!(count(&db, &handle).await, Some(3)); + assert_eq!(count(&db, &session_named("v0")).await, None); + assert_eq!(mark(&db, &s).await.map(|m| m.idx), Some(3)); + assert_eq!(s.reproject(&db).await.unwrap().replayed, 0); + } + /// Startup reprojection cost on a large store, before (a full replay into an already /// populated sidecar, what every startup used to do) and after (watermarks in place, a few /// new records). `cargo test -p atuin-client --release -- --ignored --nocapture reproject_timing` diff --git a/crates/atuin-common/src/harnesstools/ccode/rehydrate.rs b/crates/atuin-common/src/harnesstools/ccode/rehydrate.rs index 6a9312198..00ebb9b11 100644 --- a/crates/atuin-common/src/harnesstools/ccode/rehydrate.rs +++ b/crates/atuin-common/src/harnesstools/ccode/rehydrate.rs @@ -15,8 +15,8 @@ //! result as an ordinary tool call and result, losing the block types the API needs to replay //! them, and a bare `tool_use` with no `tool_result` after it is invalid. Both are dropped //! (one captured without its input becomes a note, as below). -//! - **Tool calls captured without their input** (capture keeps only a call's name now): the API -//! rejects a `tool_use` whose `input` is not an object, so each becomes a note in its turn's +//! - **Tool calls captured without their input** (`ai.capture_tools` off, or a policy withheld +//! it): the API rejects a `tool_use` whose `input` is not an object, so each becomes a note in its turn's //! text (`[ran a shell command]`; see [`tool_note`](crate::harnesstools::note::tool_note)) and its `tool_result` is dropped. //! The run of assistant lines the dropped results stood between is merged into its first line, //! so user and assistant turns still alternate ([`Flatten::Runs`]). Re-captured, that line keeps @@ -24,13 +24,12 @@ //! results are not there to capture again. The line records which synced rows went into it //! ([`MERGED_FIELD`](crate::harnesstools::rehydrate::MERGED_FIELD)), as does the line a row //! with nothing to write hangs from, so the transcript says which synced rows it holds. -//! - **Tool output**: written back as captured (`ai.capture_tools`). Capture keeps none with -//! it off; a call kept with its input (older records) gets [`UNCAPTURED_OUTPUT`] as its +//! - **Tool output**: written back as captured. A call kept with its input but not its output +//! (withheld by a policy, or over the size limit) gets [`UNCAPTURED_OUTPUT`] as its //! `tool_result` content, which says so to the model. An edit's or write's result gets back //! the record Claude Code shows its diff from (`toolUseResult.structuredPatch`) where capture //! kept the patch, without the file as it was before, which is never synced. -//! - **Pasted images and documents**: capture keeps what they were, not their bytes. Each becomes -//! a text placeholder saying so. Other blocks capture kept raw are dropped. +//! - **Pasted images and documents**: capture keeps none of them, so they are not written back. //! - **Empty lines**: rows with nothing left to write (attachments, hook records, turn timings) //! are skipped, and lines under them are linked to their nearest written ancestor. //! - **Line kinds**: harness-written user-role text (captured as [`Role::System`]) is written @@ -365,19 +364,16 @@ impl<'a> Writer<'a> { match &m.role { Role::User => user_line(&m.content), Role::Tool => tool_line(&m.content, &self.calls), - Role::System if self.boundaries.contains(m.source_id.as_str()) => { - let text = joined_text(&m.content); - Some(json!({ - "parentUuid": null, - "logicalParentUuid": parent, - "type": "system", - "subtype": "compact_boundary", - "content": text.as_deref().unwrap_or("Conversation compacted"), - "isMeta": false, - "level": "info", - "compactMetadata": {"trigger": "manual", "preTokens": 0}, - })) - } + Role::System if self.boundaries.contains(m.source_id.as_str()) => Some(json!({ + "parentUuid": null, + "logicalParentUuid": parent, + "type": "system", + "subtype": "compact_boundary", + "content": "Conversation compacted", + "isMeta": false, + "level": "info", + "compactMetadata": {"trigger": "manual", "preTokens": 0}, + })), Role::System => system_line(&m.content), Role::Assistant => self.assistant_line(m), Role::Other(_) => None, @@ -461,17 +457,12 @@ fn joined_text(content: &[Content]) -> Option { (!texts.is_empty()).then(|| texts.join("\n")) } -/// A user-role content block for `c`, if it has one: text, or a placeholder for media whose -/// bytes were never captured. +/// A user-role text block for `c`, if it is text. fn user_block(c: &Content) -> Option { match c { Content::Text(t) | Content::Summary(t) if !t.is_empty() => { Some(json!({"type": "text", "text": t})) } - Content::Other(raw) => { - let kind = raw["type"].as_str().filter(|k| matches!(*k, "image" | "document"))?; - Some(json!({"type": "text", "text": format!("[{kind} not restored]")})) - } _ => None, } } @@ -524,7 +515,7 @@ fn tool_line(content: &[Content], calls: &HashMap<&str, &ToolUse>) -> Option Some(tool_result_block(r)), - other => user_block(other), + _ => None, }) .collect(); if blocks.is_empty() { diff --git a/crates/atuin-common/src/harnesstools/ccode/rehydrate/tests.rs b/crates/atuin-common/src/harnesstools/ccode/rehydrate/tests.rs index 0c2d15a2b..2e4bda8cd 100644 --- a/crates/atuin-common/src/harnesstools/ccode/rehydrate/tests.rs +++ b/crates/atuin-common/src/harnesstools/ccode/rehydrate/tests.rs @@ -81,12 +81,8 @@ fn carried(messages: &[RehydrateMessage]) -> Vec<(String, Role, Vec)> { Content::Reasoning(_) | Content::ReasoningSummary { .. } => None, Content::ToolUse(u) if server.contains(&u.id) => None, Content::ToolResult(_) if m.role == Role::Assistant => None, - Content::Other(raw) => match raw["type"].as_str() { - Some(kind @ ("image" | "document")) => { - Some(Content::Text(format!("[{kind} not restored]"))) - } - _ => None, - }, + // Capture keeps no raw block. + Content::Other(_) => None, other => Some(other.clone()), }) .collect(); @@ -363,9 +359,7 @@ fn thinking_and_server_tools_are_dropped_and_the_tree_relinked(projects: TempDir Content::Text("found it".to_owned()), ]), message("x1", Some("a2"), Role::Other("attachment".to_owned()), vec![]), - message("u2", Some("x1"), Role::User, vec![Content::Other( - json!({"type": "image", "source": {"type": "base64", "media_type": "image/png"}}), - )]), + message("u2", Some("x1"), Role::User, vec![Content::Text("and now?".to_owned())]), ]); let path = rehydrate_into(projects.path(), &session).unwrap(); let lines = lines(&path); @@ -375,9 +369,9 @@ fn thinking_and_server_tools_are_dropped_and_the_tree_relinked(projects: TempDir let reply = by_uuid("a2").unwrap(); assert_eq!(reply["parentUuid"], "u1"); assert_eq!(reply["message"]["content"], json!([{"type": "text", "text": "found it"}])); - let image = by_uuid("u2").unwrap(); - assert_eq!(image["parentUuid"], "a2"); - assert_eq!(image["message"]["content"], "[image not restored]"); + let prompt = by_uuid("u2").unwrap(); + assert_eq!(prompt["parentUuid"], "a2"); + assert_eq!(prompt["message"]["content"], "and now?"); assert!(!std::fs::read_to_string(&path).unwrap().contains("thinking")); } diff --git a/crates/atuin-common/src/harnesstools/codex/rehydrate.rs b/crates/atuin-common/src/harnesstools/codex/rehydrate.rs index 8a367499b..903a15b96 100644 --- a/crates/atuin-common/src/harnesstools/codex/rehydrate.rs +++ b/crates/atuin-common/src/harnesstools/codex/rehydrate.rs @@ -5,9 +5,8 @@ //! Every row capture keys on a native id is written back under that id, so re-capturing the //! rollout finds the rows already synced: //! -//! - messages (`response_item` `message`), under their `id`; a system row is written as a -//! `developer` message, or as the `user` message Codex injected it as when its text is one of -//! Codex's contextual fragments (AGENTS.md, ``, ...); +//! - messages (`response_item` `message`), under their `id`; a system row (a continuation's +//! marker) is written as a `developer` message; //! - tool calls and their outputs (`function_call`, `custom_tool_call`, `local_shell_call`, //! `web_search_call`, `image_generation_call`, `tool_search_call` and their `*_output`s), under //! their `id`, or with none when capture keyed them on their `call_id` (`#out` for an @@ -46,7 +45,8 @@ //! API by `id`, which a store-less request cannot resolve. The model's earlier reasoning is not //! something it needs to continue, so these items are left out, and re-capture has no row to //! compare them with. -//! - **Tool calls captured without their input** (capture keeps only a call's name now). A +//! - **Tool calls captured without their input** (`ai.capture_tools` off, or a policy withheld +//! it). A //! `function_call`'s `arguments` must be a string of JSON, a `custom_tool_call`'s `input` a //! string, a `local_shell_call`'s `action` an object: Codex cannot read the line back, let alone //! send it. Each becomes a note (`[ran a shell command]`; see [`tool_note`](crate::harnesstools::note::tool_note)) and its @@ -60,9 +60,9 @@ //! is left out. The line of the written row before them records the rows written as no line //! ([`MERGED_FIELD`](crate::harnesstools::rehydrate::MERGED_FIELD)), so the rollout says which //! synced rows it holds. -//! - **Tool output**: written back as captured (`ai.capture_tools`). Capture keeps none with -//! it off; a call kept with its input (older records) gets [`UNCAPTURED_OUTPUT`] as its -//! output (an empty `tools` list for a tool search). +//! - **Tool output**: written back as captured. A call kept with its input but not its output +//! (withheld by a policy, or over the size limit) gets [`UNCAPTURED_OUTPUT`] as its output (an +//! empty `tools` list for a tool search). //! - Rows whose line carries nothing Codex needs back: thread names (Codex keeps those in its //! `session_index.jsonl` now) and other events. //! - Content kinds a line of the kind cannot carry (text inside a tool call, and so on). @@ -74,7 +74,7 @@ use serde_json::{Map, Value, json}; use time::format_description::well_known::Rfc3339; use time::{OffsetDateTime, UtcOffset}; -use super::session::{self, archive_of, is_contextual_user_text, session_id_of}; +use super::session::{self, archive_of, session_id_of}; use crate::harnesstools::rehydrate::{ Flatten, RehydrateError, RehydrateMessage, RehydrateSession, UNCAPTURED_OUTPUT, flatten_uncaptured_calls, record_merged, @@ -586,25 +586,15 @@ fn file_change( } fn is_message_block(content: &Content) -> bool { - matches!(content, Content::Text(_) | Content::Other(_)) + matches!(content, Content::Text(_)) } /// A `message` item, and for what the user said or the model answered, the event Codex shows it /// from on resume. fn text_message(message: &RehydrateMessage, at: &str) -> Vec { - let texts: Vec<&str> = message - .content - .iter() - .filter_map(|content| match content { - Content::Text(text) => Some(text.as_str()), - _ => None, - }) - .collect(); let (role, block) = match &message.role { Role::User => ("user", "input_text"), Role::Assistant => ("assistant", "output_text"), - // Context Codex injected as a user message reads back as the system's again. - _ if texts.iter().any(|text| is_contextual_user_text(text)) => ("user", "input_text"), _ => ("developer", "input_text"), }; let blocks: Vec = message @@ -612,8 +602,6 @@ fn text_message(message: &RehydrateMessage, at: &str) -> Vec { .iter() .filter_map(|content| match content { Content::Text(text) => Some(json!({"type": block, "text": text})), - // A block capture kept whole (an image, say): back as it was. - Content::Other(value) => Some(value.clone()), _ => None, }) .collect(); @@ -637,7 +625,6 @@ fn said(message: &RehydrateMessage) -> Option { .iter() .map(|content| match content { Content::Text(text) => Some(text.as_str()), - Content::Other(_) => Some(""), _ => None, }) .collect::>()?; diff --git a/crates/atuin-common/src/harnesstools/continuation.rs b/crates/atuin-common/src/harnesstools/continuation.rs index d5b100d90..b7e850a60 100644 --- a/crates/atuin-common/src/harnesstools/continuation.rs +++ b/crates/atuin-common/src/harnesstools/continuation.rs @@ -18,7 +18,7 @@ //! other; it just can't make it again. //! - **Other calls, as notes**: a short line of text in the assistant's turn saying what was //! called (see [`tool_note`]): `[ran a shell command]`, ``[edited `src/store.rs`]``. That is a -//! call captured without its input (by default capture keeps the tool's name only), one with no +//! call captured without its input (`ai.capture_tools` off, or a policy withheld it), one with no //! result (interrupted, or a Codex web search), and one the model's provider ran itself (a //! Claude Code web search), whose result only that provider reads. //! - Summaries (compactions, abandoned branches), as text of the user's turn. @@ -609,11 +609,6 @@ fn turns(messages: &[RehydrateMessage]) -> (Vec>, Flattened) { let note = format!("[error: {}]", clip(why, NOTE_ERROR)); assistant.push(Item::Part(Part::Note(note))); } - Content::Other(raw) if m.role == Role::User => { - if let Some(kind) = raw["type"].as_str().filter(|k| is_media(k)) { - user.push(Part::Note(format!("[{kind} not carried over]"))); - } - } _ => {} } } @@ -664,10 +659,6 @@ fn turns(messages: &[RehydrateMessage]) -> (Vec>, Flattened) { (turns, flattened) } -fn is_media(kind: &str) -> bool { - matches!(kind, "image" | "input_image" | "document" | "file" | "input_file") -} - /// A new session's id for `harness`, in its own format, minted at `now`. pub(crate) fn new_session_id(harness: AnyHarness, now: OffsetDateTime) -> String { Ids::new(harness, now).session() diff --git a/crates/atuin-common/src/harnesstools/continuation/tests.rs b/crates/atuin-common/src/harnesstools/continuation/tests.rs index 0d5c844df..03b194688 100644 --- a/crates/atuin-common/src/harnesstools/continuation/tests.rs +++ b/crates/atuin-common/src/harnesstools/continuation/tests.rs @@ -13,6 +13,7 @@ use crate::harnesstools::codex::session::CodexSession; use crate::harnesstools::opencode::Opencode; use crate::harnesstools::pi::Pi; use crate::harnesstools::pi::session::PiSession; +use crate::harnesstools::rehydrate::testing; use crate::harnesstools::session::{Session, SessionId, ToolCallId, ToolResult, ToolUse}; use crate::harnesstools::{ccode, codex, opencode, pi}; use crate::sync::BlockingPool; @@ -71,8 +72,15 @@ fn session_of(id: &str, messages: Vec) -> RehydrateSession { } } -/// A real session recorded by `source`, as capture holds it. +/// A real session recorded by `source`, as capture holds it (with `ai.capture_tools` on). pub async fn recorded(source: AnyHarness) -> RehydrateSession { + let mut session = read(source).await; + session.messages = testing::synced_as(session.messages, true); + session +} + +/// A real session recorded by `source`, as its harness wrote it. +async fn read(source: AnyHarness) -> RehydrateSession { match source { AnyHarness::ClaudeCode(_) => { let text = std::fs::read_to_string(fixture("ccode/session1.jsonl")).unwrap(); diff --git a/crates/atuin-common/src/harnesstools/fork.rs b/crates/atuin-common/src/harnesstools/fork.rs index 0a6d45f62..4be915632 100644 --- a/crates/atuin-common/src/harnesstools/fork.rs +++ b/crates/atuin-common/src/harnesstools/fork.rs @@ -2,9 +2,9 @@ //! id, linked to the original as its fork. The original is left as it is. //! //! A fork is written out from the captured rows by the harness's own writer, as a restore is -//! ([`Harness::rehydrate`](crate::harnesstools::Harness::rehydrate)): tool calls and reasoning -//! carry over as far as a restore carries them, never flattened into notes as a continuation's -//! are ([`continuation`]). What differs is its id, and that it +//! ([`Harness::rehydrate`](crate::harnesstools::Harness::rehydrate)): tool calls carry over as a +//! restore writes them (as calls where capture kept their input), not mapped onto another +//! harness's tools as a continuation's are ([`continuation`]). What differs is its id, and that it //! names the session it was forked from ([`RehydrateSession::fork_of`]), which each writer puts //! where its harness puts a fork's origin, out of the model's sight: //! diff --git a/crates/atuin-common/src/harnesstools/fork/tests.rs b/crates/atuin-common/src/harnesstools/fork/tests.rs index 749a45f91..2c169c607 100644 --- a/crates/atuin-common/src/harnesstools/fork/tests.rs +++ b/crates/atuin-common/src/harnesstools/fork/tests.rs @@ -109,19 +109,9 @@ async fn written( } } -/// What `rows` say, without the ids a part kept whole carries (a fork's are its own). +/// What `rows` say. fn said(rows: &[RehydrateMessage]) -> Vec<(Role, Vec)> { - let unkeyed = |c: &Content| match c { - Content::Other(serde_json::Value::Object(part)) => { - let mut part = part.clone(); - for key in ["id", "messageID", "sessionID"] { - part.remove(key); - } - Content::Other(serde_json::Value::Object(part)) - } - c => c.clone(), - }; - rows.iter().map(|r| (r.role.clone(), r.content.iter().map(unkeyed).collect())).collect() + rows.iter().map(|r| (r.role.clone(), r.content.clone())).collect() } /// A fork, written by its harness's writer next to the original and read back by its reader, is diff --git a/crates/atuin-common/src/harnesstools/note.rs b/crates/atuin-common/src/harnesstools/note.rs index 2ba72b122..6cc89b286 100644 --- a/crates/atuin-common/src/harnesstools/note.rs +++ b/crates/atuin-common/src/harnesstools/note.rs @@ -71,9 +71,9 @@ pub(crate) fn render(parts: &[Part]) -> String { /// | Pi | `bash`, `edit`, `write`, `read`, `grep`, `find`, `ls` | as Claude Code's (`path`) | /// | any | anything else | ``called `` `` | /// -/// Without the input (capture keeps none by default), each says only what was done: `ran a shell -/// command`, `edited a file`, `wrote a file`, `read a file`, `applied a patch`, `searched the -/// files`, `listed files`, `handed work to a subagent`, `fetched a web page`, `searched the +/// Without the input (`ai.capture_tools` off, or withheld), each says only what was done: `ran a +/// shell command`, `edited a file`, `wrote a file`, `read a file`, `applied a patch`, `searched +/// the files`, `listed files`, `handed work to a subagent`, `fetched a web page`, `searched the /// web`; and anything else ``called `` ``. Inputs are cut to their first line and /// 120 characters. #[must_use] @@ -95,8 +95,8 @@ pub fn tool_note(name: &str, input: &Value) -> String { }) }; let path = || field(&["file_path", "filePath", "path", "notebook_path", "file"]); - // What was done, on what when the input says (capture keeps none by default, only with - // `ai.capture_tools` and in older records), else only what. + // What was done, on what when the input says (unless `ai.capture_tools` is off or a policy + // withheld it), else only what. let on = |what: Option, detailed: &dyn Fn(&str) -> String, bare: &str| { Some(what.map_or_else(|| bare.to_owned(), |w| detailed(&w))) }; diff --git a/crates/atuin-common/src/harnesstools/opencode/rehydrate.rs b/crates/atuin-common/src/harnesstools/opencode/rehydrate.rs index bf089ff3b..0b9dd8f72 100644 --- a/crates/atuin-common/src/harnesstools/opencode/rehydrate.rs +++ b/crates/atuin-common/src/harnesstools/opencode/rehydrate.rs @@ -8,20 +8,21 @@ //! //! # What survives //! -//! Every part capture delivered, under its own id: text (the user's, the model's, and what -//! opencode injected, `synthetic`), reasoning, tool calls with their result or error, retries, -//! each model call's `step-finish` with its token counts and finish reason, and the parts capture -//! kept whole (`step-start`, files, compaction markers, subtasks...) as they were. A failed -//! model call is its assistant message's `error`, under the message id capture keyed it on. +//! Every part capture keeps, under its own id: text (the user's, the model's, and a continuation's +//! or fork's marker, `synthetic` or `ignored`), tool calls with their result or error, retries, +//! and each model call's `step-finish` with its token counts and finish reason. A failed model +//! call is its assistant message's `error`, under the message id capture keyed it on. Reasoning +//! (kept only as a marker) and the parts capture keeps nothing of (`step-start`, files, +//! compaction markers, subtasks...) are left out. //! //! What capture keeps of a message is less than opencode's info for it, so the rest is rebuilt: //! //! - **Which message a part belongs to.** Rows carry no message id, so parts are grouped back //! into messages: an assistant's by the user message it answers and the model call it was //! (`MessageInfo::turn_of`, which also gives back its creation time, provider and model), the -//! rest in the order they came. A message's id is recovered where a row names it -- a part -//! kept whole carries its `messageID`, a failure is keyed on it, an assistant message names the -//! user message it answers -- and minted otherwise. Only part and failure ids are row keys, so a +//! rest in the order they came. A message's id is recovered where a row names it -- an +//! opencode 2.0 row and a failure are keyed on it, an assistant message names the user message +//! it answers -- and minted otherwise. Only part and failure ids are row keys, so a //! minted message id costs nothing on re-capture. //! - **Agent and mode** (`build`), cost (`0`), and a tool call's title (empty) and metadata: //! empty, but for a call whose patch capture kept, which gets back the diff opencode shows an @@ -31,20 +32,21 @@ //! //! # What is not the same //! -//! - **Tool calls captured without their input** (capture keeps only a call's name now) are not -//! written as tool parts: opencode would send the model a call on no input (`{"input": null}`, -//! the object its state needs) with an output it never had. Each becomes a text part under the -//! call's own part id, reading as a [note](crate::harnesstools::note::tool_note) (`[ran a shell command]`), or joins a -//! text part of the same message before it, with nothing but reasoning or other notes between -//! (the same note several times in a row counted, `×3`; see [`Flatten::Notes`]). Re-captured, +//! - **Tool calls captured without their input** (`ai.capture_tools` off, or a policy withheld +//! it) are not written as tool parts: opencode would send the model a call on no input +//! (`{"input": null}`, the object its state needs) with an output it never had. Each becomes a +//! text part under the call's own part id, reading as a +//! [note](crate::harnesstools::note::tool_note) (`[ran a shell command]`), or joins a text part +//! of the same message before it, with nothing but reasoning or other notes between (the same +//! note several times in a row counted, `×3`; see [`Flatten::Notes`]). Re-captured, //! the part reads back under a part id capture already holds, so nothing is pushed; a part //! merged away is not there to capture again. -//! - **Tool output**: written back as captured (`ai.capture_tools`). Capture keeps none with -//! it off; a call kept with its input (older records) is written `completed` (or `error`) with +//! - **Tool output**: written back as captured. A call kept with its input but not its output +//! (withheld by a policy, or over the size limit) is written `completed` (or `error`) with //! [`UNCAPTURED_OUTPUT`] as its output. -//! - opencode stamps each part row with the time of the import. Text, reasoning, tool and retry -//! parts carry their own clock and read back with the captured timestamp; the rest (`step-start`, -//! `step-finish`, whole parts) read back with the import's. +//! - opencode stamps each part row with the time of the import. Text, tool and retry parts carry +//! their own clock and read back with the captured timestamp; a `step-finish` reads back with +//! the import's. //! - opencode reads a message's parts in id order, where capture delivered them as they finished: //! parallel tool calls that finished out of order come back in id order. //! - A tool call captured without a result (one opencode stopped mid-way) is written `pending`, @@ -402,16 +404,9 @@ fn turn_base(row: &RehydrateMessage) -> Option { Some(turn.split_once('#').map_or(turn, |(turn, _)| turn).to_owned()) } -/// The message a row names as its own: a part kept whole carries its `messageID`, a row of -/// opencode 2.0 is keyed `/`, and a failure is keyed on its message. +/// The message a row names as its own: a row of opencode 2.0 is keyed `/`, and a +/// failure is keyed on its message. fn named_message(row: &RehydrateMessage) -> Option { - for content in &row.content { - if let Content::Other(part) = content - && let Some(id) = part["messageID"].as_str() - { - return Some(id.to_owned()); - } - } if let Some((message, _)) = row.source_id.split_once('/') { return Some(message.to_owned()); } @@ -944,9 +939,6 @@ fn part_of(row: &RehydrateMessage, draft: &mut Draft) -> Option { draft.summary = true; Some(json!({"type": "text", "text": text, "time": {"start": at, "end": at}})) } - [Content::Reasoning(text)] => { - Some(json!({"type": "reasoning", "text": text, "time": {"start": at, "end": at}})) - } [Content::ToolUse(call)] => Some(tool_part(call, None, None, at)), [Content::ToolUse(call), Content::ToolResult(result)] => { Some(tool_part(call, Some(result), None, at)) @@ -961,7 +953,6 @@ fn part_of(row: &RehydrateMessage, draft: &mut Draft) -> Option { "error": {"name": "APIError", "data": {"message": why, "isRetryable": true}}, "time": {"created": at}, })), - [Content::Other(part)] if part["type"].is_string() => Some(part.clone()), _ => None, } } @@ -1189,14 +1180,14 @@ pub(crate) mod tests { } } - /// Captures `original` from `original_db`, writes it back as `opencode import` would and - /// captures it again: the rows (bar the session's own) and the session's title row. + /// Captures `original` from `original_db` as capture syncs it (tool payloads kept), writes it + /// back as `opencode import` would and captures it again. async fn round_trip( original_db: &Path, id: &str, title: Option<&str>, ) -> (Vec, Vec) { - let original = captured(original_db).await; + let original = testing::synced_as(captured(original_db).await, true); let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("opencode.db"); let mut conn = database(&path).await; @@ -1276,6 +1267,13 @@ pub(crate) mod tests { let title = s["title"].as_str(); let (original, again) = round_trip(&path, id, title).await; + // A part capture kept nothing of is not written, but for a model call's `step-finish`; + // nor is reasoning capture kept no text of. + let original: Vec = original + .into_iter() + .filter(|m| !m.content.is_empty() || m.usage.is_some() || is_session(id, m)) + .filter(|m| !matches!(m.content.as_slice(), [Content::ReasoningSummary { .. }])) + .collect(); assert!(original.len() > 5); let rows = |rows: &[RehydrateMessage]| -> Vec { rows.iter().filter(|r| !is_session(id, r)).cloned().collect() @@ -1358,8 +1356,6 @@ pub(crate) mod tests { let mut failure = answer(A2, 1_790_217_663, U2, TURN2, vec![Content::Error("mock bad request".into())]); failure.stop_reason = Some(StopReason::Error); - let step_start = json!({"type": "step-start", "id": "prt_0d148db28001TznIjdIVqXAvJ3", - "messageID": "msg_0d148d743001qhQZH5hIpZVrlb", "sessionID": SES}); vec![ row(&format!("{SES}:title:Fix the build"), 1_790_217_606, Role::System, vec![]), row("prt_0d148d4bf001JaxA9TwaMgU5vE", 1_790_217_606, Role::User, vec![Content::Text( @@ -1368,12 +1364,6 @@ pub(crate) mod tests { row("prt_0d148d4c0001JaxA9TwaMgU5vF", 1_790_217_606, Role::System, vec![ Content::Text("Called the Read tool".into()), ]), - answer("prt_0d148db28001TznIjdIVqXAvJ3", 1_790_217_607, U1, TURN1, vec![ - Content::Other(step_start), - ]), - answer("prt_0d148db29001TznIjdIVqXAvJ4", 1_790_217_607, U1, TURN1, vec![ - Content::Reasoning("thinking it over".into()), - ]), answer("prt_0d148db2a001TznIjdIVqXAvJ5", 1_790_217_607, U1, TURN1, vec![ Content::Error("overloaded".into()), ]), @@ -1413,8 +1403,7 @@ pub(crate) mod tests { } /// Every kind of row comes back under its own id, in its message: the user message the - /// assistant's answers is named for it, the assistant's for the part that carried its id - /// or the failure keyed on it. + /// assistant's answers is named for it, a failed assistant's for the failure keyed on it. #[rstest] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn every_kind_of_row_recaptures_as_itself() { @@ -1427,9 +1416,9 @@ pub(crate) mod tests { conn.close().await.unwrap(); let again = captured(&path).await; pretty_assertions::assert_eq!(keys(&again), keys(&rows)); - assert_eq!(again[9].usage, rows[9].usage); - assert_eq!(again[9].turn_id, rows[9].turn_id); - assert_eq!(again[11].stop_reason, Some(StopReason::Error)); + assert_eq!(again[7].usage, rows[7].usage); + assert_eq!(again[7].turn_id, rows[7].turn_id); + assert_eq!(again[9].stop_reason, Some(StopReason::Error)); let ids: Vec<&str> = exported["messages"] .as_array() @@ -1437,7 +1426,8 @@ pub(crate) mod tests { .iter() .map(|m| m["info"]["id"].as_str().unwrap()) .collect(); - assert_eq!(ids, [U1, "msg_0d148d743001qhQZH5hIpZVrlb", U2, A2]); + assert_eq!([ids[0], ids[2], ids[3]], [U1, U2, A2]); + assert!(ids[1].starts_with("msg_"), "the answer's id is minted: {ids:?}"); let assistant = &exported["messages"][1]["info"]; assert_eq!(assistant["parentID"], U1); assert_eq!(assistant["providerID"], "openai"); diff --git a/crates/atuin-common/src/harnesstools/opencode/session/v2.rs b/crates/atuin-common/src/harnesstools/opencode/session/v2.rs index a51941cce..305a12a16 100644 --- a/crates/atuin-common/src/harnesstools/opencode/session/v2.rs +++ b/crates/atuin-common/src/harnesstools/opencode/session/v2.rs @@ -99,120 +99,74 @@ const SUBAGENT_PROMPT: &str = "You are a subagent spawned by another session."; /// Light rows a pass reads at a time. const ROWS: usize = 256; -/// A place in one layout of a session: the `seq` of the last row passed and a digest of its id. +/// A place in one layout of a session: the `seq` of the last row passed and the low half of its +/// id's xxh3, which is what a [`checkpoint`] holding both layouts has room for. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) struct Mark { pub(super) seq: i64, - digest: Digest, -} - -/// How much of an id's xxh3 a mark keeps: all of it, or the low half a checkpoint holding both -/// layouts has room for. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum Digest { - Full(u64), - Low(u32), + digest: u32, } impl Mark { pub(super) fn of(seq: i64, id: &str) -> Self { Self { seq, - digest: Digest::Full(xxh3_64(id.as_bytes())), + digest: Self::digest_of(id), } } /// Whether `id` is the id of the row this mark was taken at. pub(super) fn names(&self, id: &str) -> bool { - let hash = xxh3_64(id.as_bytes()); - match self.digest { - Digest::Full(digest) => hash == digest, - Digest::Low(low) => Self::low_of(hash) == low, - } - } - - const fn low(&self) -> u32 { - match self.digest { - Digest::Full(hash) => Self::low_of(hash), - Digest::Low(low) => low, - } + Self::digest_of(id) == self.digest } #[allow(clippy::cast_possible_truncation, reason = "the low half is the point")] - const fn low_of(hash: u64) -> u32 { - hash as u32 + fn digest_of(id: &str) -> u32 { + xxh3_64(id.as_bytes()) as u32 } } -/// The bit a checkpoint holding both layouts sets. -const BOTH: u64 = 1 << 63; -/// How many bits of such a checkpoint hold the 2.0 layout's place. +/// The top bit, which every checkpoint sets above the places it packs. +const PACKED: u64 = 1 << 63; +/// How many bits of a checkpoint hold the 2.0 layout's place. const NEXT_BITS: u32 = 31; /// The checkpoint of a session whose 1.x layout was read up to `legacy` and whose 2.0 layout up -/// to `next`. -/// -/// A session read in its 1.x layout alone keeps the checkpoint it always had: the row's `seq` -/// and the full digest of its id, so a checkpoint stored before 2.0 support still resumes. Once -/// the 2.0 layout has a place, both are packed into one: the top bit set, then each place's `seq` -/// plus one (zero for none), the 1.x one above the 2.0 one, so that the checkpoint grows as -/// either place does; the digest holds the low halves of both ids' digests. A place too far on -/// to pack -- two billion events into one session -- is left out and read again. +/// to `next`: the top bit set, then each place's `seq` plus one (zero for none), the 1.x one above +/// the 2.0 one, so that the checkpoint grows as either place does; the digest holds both marks' +/// digests. A place too far on to pack -- two billion events into one session -- is left out and +/// read again. pub(super) fn checkpoint(legacy: Option<(i64, &str)>, next: Option) -> Checkpoint { - let legacy = legacy.map(|(seq, id)| Mark::of(seq, id)); - let Some(next) = next else { - return match legacy { - Some(mark) => Checkpoint { - at: u64::try_from(mark.seq).unwrap_or_default(), - digest: match mark.digest { - Digest::Full(digest) => digest, - Digest::Low(low) => u64::from(low), - }, - generation: 0, - }, - None => Checkpoint { - at: BOTH, - digest: 0, - generation: 0, - }, - }; - }; let plus_one = |mark: Option, bits: u32| { mark.and_then(|mark| u64::try_from(mark.seq).ok()) .map(|seq| seq + 1) .filter(|place| *place < 1 << bits) .zip(mark) }; - let (legacy_at, legacy_low) = - plus_one(legacy, 32).map_or((0, 0), |(at, mark)| (at, mark.low())); - let (next_at, next_low) = - plus_one(Some(next), NEXT_BITS).map_or((0, 0), |(at, mark)| (at, mark.low())); + let legacy = legacy.map(|(seq, id)| Mark::of(seq, id)); + let (legacy_at, legacy_digest) = + plus_one(legacy, 32).map_or((0, 0), |(at, mark)| (at, mark.digest)); + let (next_at, next_digest) = + plus_one(next, NEXT_BITS).map_or((0, 0), |(at, mark)| (at, mark.digest)); Checkpoint { - at: BOTH | legacy_at << NEXT_BITS | next_at, - digest: u64::from(legacy_low) << 32 | u64::from(next_low), + at: PACKED | legacy_at << NEXT_BITS | next_at, + digest: u64::from(legacy_digest) << 32 | u64::from(next_digest), generation: 0, } } /// The places a checkpoint names in each layout (see [`checkpoint`]). pub(super) fn unpack(from: Checkpoint) -> (Option, Option) { - if from.at & BOTH == 0 { - let legacy = i64::try_from(from.at).ok().map(|seq| Mark { - seq, - digest: Digest::Full(from.digest), - }); - return (legacy, None); - } let place = |at: u64, low: u64| { let seq = i64::try_from(at.checked_sub(1)?).ok()?; Some(Mark { seq, - digest: Digest::Low(u32::try_from(low & 0xffff_ffff).unwrap_or_default()), + digest: u32::try_from(low & 0xffff_ffff).unwrap_or_default(), }) }; let next_mask = (1 << NEXT_BITS) - 1; ( - place((from.at & !BOTH) >> NEXT_BITS, from.digest >> 32), + place((from.at & !PACKED) >> NEXT_BITS, from.digest >> 32), place(from.at & next_mask, from.digest), ) } diff --git a/crates/atuin-common/src/harnesstools/opencode/session/v2/tests.rs b/crates/atuin-common/src/harnesstools/opencode/session/v2/tests.rs index d6224f0f1..f4344324b 100644 --- a/crates/atuin-common/src/harnesstools/opencode/session/v2/tests.rs +++ b/crates/atuin-common/src/harnesstools/opencode/session/v2/tests.rs @@ -661,13 +661,6 @@ fn a_session_s_info_names_its_title_and_where_it_came_from() { assert_eq!(titled.cwd(), Some(PathBuf::from("/work/proj"))); } -#[rstest] -fn a_1x_checkpoint_keeps_its_old_shape() { - assert_eq!(checkpoint(Some((7, "evt_7")), None), Checkpoint::new(7, b"evt_7")); - assert_eq!(unpack(Checkpoint::new(7, b"evt_7")).1, None); - assert!(unpack(Checkpoint::new(7, b"evt_7")).0.unwrap().names("evt_7")); -} - proptest! { /// Both places come back out of the checkpoint they were packed into. #[test] @@ -1220,7 +1213,8 @@ async fn a_session_in_both_layouts_is_one_session(#[future] mixed: Db) { legacy_event("evt_3", "message.part.updated.1", text_event("prt_2", "one point x again")).await; db.append("ses_m", "msg_newer", "user", user("two point oh again")).await; let at = stored.lock()[&SessionId::from("ses_m".to_owned())]; - assert_ne!(at.at & BOTH, 0, "the checkpoint holds both places"); + let (legacy, next) = unpack(at); + assert!(legacy.is_some() && next.is_some(), "the checkpoint holds both places"); let mut stream = events(&db.path, ReplayBehavior::All, &stored); let mut got = lines(&next_n(&mut stream, &stored, 3).await); got.sort(); diff --git a/crates/atuin-common/src/harnesstools/pi/rehydrate.rs b/crates/atuin-common/src/harnesstools/pi/rehydrate.rs index c21cdc233..d5d6dcd44 100644 --- a/crates/atuin-common/src/harnesstools/pi/rehydrate.rs +++ b/crates/atuin-common/src/harnesstools/pi/rehydrate.rs @@ -8,8 +8,8 @@ //! //! - **Thinking.** Capture keeps only a marker ([`Content::ReasoningSummary`]), no text or //! signature, so it is dropped; pi itself drops a thinking block it can't replay. -//! - **Pasted images**: capture keeps their type, not their bytes; each becomes a text -//! placeholder saying so. +//! - **Pasted images** and extension-injected messages (`custom_message`): capture keeps +//! neither, so they are not written back. //! - **Compactions** become `compaction` entries keeping every entry since the one before //! (capture doesn't record which entries pi kept): the model sees the summary and the full //! history since the previous compaction. A branch summary is written the same way. @@ -17,19 +17,19 @@ //! content and are skipped, so pi starts on its configured thinking level; the model is named //! by each assistant turn, its provider guessed from the model's name, and pi falls back to its //! default model when it can't use it. A system-role text entry is skipped too. -//! - **Tool calls captured without their input** (capture keeps only a call's name now): pi-ai -//! sends a `toolCall` without `arguments` as an empty input to Anthropic and as `"null"` to -//! OpenAI, a call on nothing either way, so each becomes a note in its turn's text (`[ran a -//! shell command]`; see [`tool_note`](crate::harnesstools::note::tool_note)) and its `toolResult` is dropped. The run of -//! assistant messages the dropped results stood between is merged into its first one, so the +//! - **Tool calls captured without their input** (`ai.capture_tools` off, or a policy withheld +//! it): pi-ai sends a `toolCall` without `arguments` as an empty input to Anthropic and as +//! `"null"` to OpenAI, a call on nothing either way, so each becomes a note in its turn's text +//! (`[ran a shell command]`; see [`tool_note`](crate::harnesstools::note::tool_note)) and its +//! `toolResult` is dropped. The run of assistant messages the dropped results stood between is merged into its first one, so the //! turns still alternate ([`Flatten::Runs`]). Re-captured, that entry keeps its id (capture //! already holds it: nothing is pushed), and the entries merged away and the results are not //! there to capture again. The entry records which synced rows went into it //! ([`MERGED_FIELD`](crate::harnesstools::rehydrate::MERGED_FIELD)), as does the entry a row //! with nothing to write hangs from, so the file says which synced rows it holds. //! - **Tool results** keep their output; their tool's name comes from the call they answer. With -//! none captured (capture keeps none by default), a result says [`UNCAPTURED_OUTPUT`], as does a -//! `!command`'s output. An `edit`'s result whose patch capture kept gets back the `details` pi +//! none captured (withheld by a policy, or over the size limit), a result says +//! [`UNCAPTURED_OUTPUT`], as does a `!command`'s output. An `edit`'s result whose patch capture kept gets back the `details` pi //! shows its diff from: the unified diff (`patch`), pi's own line-numbered one (`diff`) and the //! first line it changed. //! - A `!command` keeps its command, output and whether it failed, not its exit code; one from a @@ -368,24 +368,12 @@ impl<'a> Writer<'a> { } Some(entry) } - Role::Other(kind) if kind == "custom" => { - let content = user_blocks(&m.content); - (!content.is_empty()).then(|| { - json!({ - "type": "custom_message", - "customType": "atuin-restored", - "content": content, - "display": true, - }) - }) - } Role::Other(_) => None, } } } -/// Text blocks for user-role content, with a placeholder for media whose bytes were never -/// captured. +/// Text blocks for user-role content. fn user_blocks(content: &[Content]) -> Vec { content .iter() @@ -393,9 +381,6 @@ fn user_blocks(content: &[Content]) -> Vec { Content::Text(t) | Content::Summary(t) if !t.is_empty() => { Some(json!({"type": "text", "text": t})) } - Content::Other(raw) if raw["type"] == "image" => { - Some(json!({"type": "text", "text": "[image not restored]"})) - } _ => None, }) .collect() diff --git a/crates/atuin-common/src/harnesstools/pi/rehydrate/tests.rs b/crates/atuin-common/src/harnesstools/pi/rehydrate/tests.rs index 9ec8345fb..8c9a3404a 100644 --- a/crates/atuin-common/src/harnesstools/pi/rehydrate/tests.rs +++ b/crates/atuin-common/src/harnesstools/pi/rehydrate/tests.rs @@ -61,7 +61,7 @@ fn carried(messages: &[RehydrateMessage]) -> Vec<(String, Role, Vec)> { messages .iter() .filter(|m| match &m.role { - Role::Other(kind) => kind == "custom", + Role::Other(_) => false, Role::System => m.content.iter().any(|c| matches!(c, Content::Summary(_))), _ => true, }) @@ -72,10 +72,6 @@ fn carried(messages: &[RehydrateMessage]) -> Vec<(String, Role, Vec)> { .filter_map(|c| match c { Content::Reasoning(_) | Content::ReasoningSummary { .. } => None, Content::Text(t) if t.is_empty() && m.role != Role::Assistant => None, - Content::Other(raw) if raw["type"] == "image" => { - Some(Content::Text("[image not restored]".to_owned())) - } - Content::Other(_) => None, // A v1 `!command` had no id to name its result after; it has now. Content::ToolResult(r) if r.call.as_ref().starts_with("bash:") => { Some(Content::ToolResult(ToolResult { @@ -111,7 +107,8 @@ async fn read_back(path: &Path) -> Vec { #[tokio::test] async fn a_session_reads_back_as_it_was_captured(sessions: TempDir, #[case] jsonl: &str) { let cwd = sessions.path().join("here"); - let session = session("0199aaaa-bbbb-7ccc-8ddd-eeeeffff0000", &cwd, captured(jsonl)); + let synced = testing::synced_as(captured(jsonl), true); + let session = session("0199aaaa-bbbb-7ccc-8ddd-eeeeffff0000", &cwd, synced); let dir = sessions.path().join("--here--"); let path = rehydrate_into(sessions.path(), &dir, &session).unwrap(); assert_eq!(path.parent(), Some(dir.as_path())); diff --git a/crates/atuin-common/src/harnesstools/rehydrate.rs b/crates/atuin-common/src/harnesstools/rehydrate.rs index 7d4b3721c..13037d914 100644 --- a/crates/atuin-common/src/harnesstools/rehydrate.rs +++ b/crates/atuin-common/src/harnesstools/rehydrate.rs @@ -70,9 +70,9 @@ pub enum RehydrateError { Io(#[from] std::io::Error), } -/// What a restored tool result says when capture kept no output (by default it keeps none): the -/// formats all want some output for a result, and an empty one would tell the model the tool -/// printed nothing. +/// What a restored tool result says when capture kept no output (`ai.capture_tools` off, a policy +/// withheld it, or it was over the size limit): the formats all want some output for a result, and +/// an empty one would tell the model the tool printed nothing. pub const UNCAPTURED_OUTPUT: &str = "(output not captured)"; /// How a writer lays out the calls [`flatten_uncaptured_calls`] turns into notes. @@ -95,7 +95,8 @@ pub enum Flatten<'f> { }, } -/// A tool call capture kept without its input: by default, capture syncs a call's name only. +/// A tool call capture kept without its input (`ai.capture_tools` off, a policy withheld it, or it +/// was over the size limit): only its name is synced. #[must_use] pub fn is_uncaptured(content: &Content) -> bool { matches!(content, Content::ToolUse(call) if call.input.is_null()) @@ -107,8 +108,7 @@ pub fn is_uncaptured(content: &Content) -> bool { /// /// No harness's API takes a call without its input (Claude's `tool_use.input` must be an object, /// a Codex `function_call`'s `arguments` a JSON string, ...), and one made up would tell the model -/// it called a tool on nothing. Calls kept with their input (`ai.capture_tools`, older records) -/// stay calls. +/// it called a tool on nothing. Calls kept with their input stay calls. /// /// Rows keep their source ids: a row whose content changed reads back under its own id, which /// capture already holds, so re-capturing the transcript pushes nothing for it; a row merged away @@ -449,15 +449,24 @@ pub(crate) mod testing { use super::*; use crate::harnesstools::session::{ToolResult, ToolUse}; - /// `content` of a `role` row as capture syncs it (the daemon's `sanitize`): calls - /// without their input, results without their output, reasoning as a marker, text only of - /// what the user or the model said, nothing kept raw. + /// `content` of a `role` row as capture syncs it with `ai.capture_tools` off (the daemon's + /// `sanitize`): calls without their input, results without their output, no patches, + /// reasoning as a marker, text only of what the user or the model said, nothing kept raw. pub fn sanitize(role: &Role, content: &[Content]) -> Vec { + sanitize_as(role, content, false) + } + + /// [`sanitize`], keeping tool inputs, outputs and patches when `tools` (`ai.capture_tools` + /// on, the default). + pub fn sanitize_as(role: &Role, content: &[Content], tools: bool) -> Vec { let conversation = matches!(role, Role::User | Role::Assistant); content .iter() .filter_map(|c| match c { Content::Text(_) if conversation => Some(c.clone()), + Content::ToolUse(_) | Content::ToolResult(_) | Content::Patch(_) if tools => { + Some(c.clone()) + } Content::ToolUse(call) => Some(Content::ToolUse(ToolUse { input: serde_json::Value::Null, ..call.clone() @@ -476,9 +485,14 @@ pub(crate) mod testing { } /// `rows` as capture syncs them (see [`sanitize`]). - pub fn synced(mut rows: Vec) -> Vec { + pub fn synced(rows: Vec) -> Vec { + synced_as(rows, false) + } + + /// `rows` as capture syncs them, keeping tool payloads when `tools` (see [`sanitize_as`]). + pub fn synced_as(mut rows: Vec, tools: bool) -> Vec { for row in &mut rows { - row.content = sanitize(&row.role, &row.content); + row.content = sanitize_as(&row.role, &row.content, tools); } rows } diff --git a/crates/atuin-common/src/harnesstools/sync/pi/tests.rs b/crates/atuin-common/src/harnesstools/sync/pi/tests.rs index dc79befde..b1c5ae650 100644 --- a/crates/atuin-common/src/harnesstools/sync/pi/tests.rs +++ b/crates/atuin-common/src/harnesstools/sync/pi/tests.rs @@ -423,8 +423,8 @@ fn content_row(id: &str, role: Role, content: Vec) -> RehydrateMessage } } -/// A branch holds the rows beside the tree that go with it (pi's prompts from before ids, -/// extensions' messages, titles: keyed on their content, with no parent): the switch passes over +/// A branch holds the rows beside the tree that go with it (pi's prompts from before ids, titles: +/// keyed on their content, with no parent): the switch passes over /// those before where the branch leaves the copy, writes those after it it hasn't got (each /// hanging from the entry before it, as a restore writes them; metadata merged into it), and pi /// resumes from the last of them. @@ -436,8 +436,7 @@ fn a_switch_writes_the_rows_beside_the_tree_that_go_with_the_branch(dir: TempDir let mut branch = their_branch(); let said = |text: &str| vec![Content::Text(text.to_owned())]; branch.insert(0, content_row("syn-0000000000000001", Role::User, said("before ids"))); - let custom = Role::Other("custom".into()); - branch.insert(5, content_row("syn-0000000000000002", custom, said("an extension's"))); + branch.insert(5, content_row("syn-0000000000000002", Role::User, said("also before ids"))); let title = Role::Other("session_info".into()); branch.push(content_row("syn-0000000000000003", title, Vec::new())); diff --git a/crates/atuin-daemon/proto/ai/agent.proto b/crates/atuin-daemon/proto/ai/agent.proto index 4e2d3f3dd..2d3ab587c 100644 --- a/crates/atuin-daemon/proto/ai/agent.proto +++ b/crates/atuin-daemon/proto/ai/agent.proto @@ -143,39 +143,36 @@ message Message { // The session this message belongs to. HarnessSession session = 2; // Session this one was handed off from, if any. - optional HarnessSession parent = 4; - // Was a harness-native thread identifier that no harness supplied. - reserved 5; - reserved "thread"; + optional HarnessSession parent = 3; // Harness-native message identifier, unique within the session. - string source_id = 6; + string source_id = 4; // Harness-native identifier of the message this one replies to. - optional string parent_source_id = 7; + optional string parent_source_id = 5; // When the message was produced. - google.protobuf.Timestamp timestamp = 8; + google.protobuf.Timestamp timestamp = 6; // Who or what produced this message. - Role role = 9; + Role role = 7; // The message's content, in order. - repeated ContentBlock content = 10; + repeated ContentBlock content = 8; // Working directory at the time of the message. - optional string cwd = 11; + optional string cwd = 9; // Git branch checked out at the time of the message. - optional string git_branch = 12; + optional string git_branch = 10; // Model that produced the message. - optional string model = 13; + optional string model = 11; // Token usage attributed to this message, if the harness reported any. - Tokens tokens = 14; + Tokens tokens = 12; // Why an assistant turn ended. Absent when the harness did not say. - optional StopReason stop_reason = 15; + optional StopReason stop_reason = 13; // The harness's own role string (e.g. codex "developer"); set exactly when role is // ROLE_OTHER. - optional string role_label = 16; + optional string role_label = 14; // The harness's own stop reason; set exactly when stop_reason is STOP_REASON_OTHER. - optional string stop_reason_label = 17; + optional string stop_reason_label = 15; // How this message's session relates to `parent`, when known. - optional ParentKind parent_kind = 18; + optional ParentKind parent_kind = 16; // The host that captured the message (a hyphenated host id), when known. - optional string host_id = 19; + optional string host_id = 17; } message Session { @@ -206,7 +203,7 @@ message Session { // calls they share); always of the same harness. optional HarnessSession copy_of = 19; // Atuin's id for the session. - optional common.Uuid atuin_id = 20; + common.Uuid atuin_id = 20; // The atuin ids of `parent` and `root`, when those sessions are stored: links are resolved // when read, never recorded. optional common.Uuid parent_atuin_id = 21; diff --git a/crates/atuin-daemon/proto/ai/session.proto b/crates/atuin-daemon/proto/ai/session.proto index 59e269d03..08d156c2a 100644 --- a/crates/atuin-daemon/proto/ai/session.proto +++ b/crates/atuin-daemon/proto/ai/session.proto @@ -28,12 +28,7 @@ message SessionFilter { } message ListSessionsRequest { - // Superseded by filter.harness, which wins when both are set. - optional ai.agent.HarnessKind harness = 1; - // Only sessions active at or after this time. - // Superseded by filter.updated_since, which wins when both are set. - google.protobuf.Timestamp updated_since = 2; - SessionFilter filter = 3; + SessionFilter filter = 1; } message GetSessionRequest { @@ -55,16 +50,10 @@ message SearchSessionsRequest { // unbounded: every matching session is ranked and streamed. Ranking is eager // (the full match set is scored before the first result), so prefer a bound. uint32 limit = 2; - // Only search sessions from this harness. Absent searches every harness. - // Superseded by filter.harness, which wins when both are set. - optional ai.agent.HarnessKind harness = 3; - // Only match sessions whose directory is this one or beneath it. Absent searches everywhere. - // Superseded by filter.workspace, which wins when both are set. - optional string cwd = 4; // Match messages containing any of the query's terms, each as a prefix, instead of all of them // as whole words. - bool any_term = 5; - SessionFilter filter = 6; + bool any_term = 3; + SessionFilter filter = 4; } message SearchSessionsMatch { diff --git a/crates/atuin-daemon/src/client.rs b/crates/atuin-daemon/src/client.rs index f94d486ed..288ebe27b 100644 --- a/crates/atuin-daemon/src/client.rs +++ b/crates/atuin-daemon/src/client.rs @@ -534,13 +534,10 @@ impl AiClient { /// Wait while the daemon is still rebuilding AI sessions after starting: until then every /// session read but a tail is refused rather than answered partially. Calls `on_wait` each /// time it finds the daemon rebuilding, with how far it has got (records replayed, and - /// roughly how many there are to replay) when the daemon says, and returns at once when it - /// is not. After this the daemon serves reads until a store command has it rebuild again - /// (`atuin store rebuild ai-session`, a purge or a forced pull), which is rare. - pub async fn wait_for_sessions( - &mut self, - mut on_wait: impl FnMut(Option<(u64, u64)>), - ) -> Result<()> { + /// roughly how many there are to replay), and returns at once when it is not. After this the + /// daemon serves reads until a store command has it rebuild again (`atuin store rebuild + /// ai-session`, a purge or a forced pull), which is rare. + pub async fn wait_for_sessions(&mut self, mut on_wait: impl FnMut((u64, u64))) -> Result<()> { let mut delay = REBUILD_POLL_START; while let Some(progress) = self.rebuild_status().await? { on_wait(progress); @@ -552,22 +549,24 @@ impl AiClient { /// Whether the daemon is rebuilding AI sessions right now (see [`Self::wait_for_sessions`]), /// without waiting for it to finish: `Some` while it is, with how far it has got (records - /// replayed, and roughly how many there are to replay) when the daemon says; `None` when it - /// serves reads. - pub async fn rebuild_status(&mut self) -> Result>> { + /// replayed, and roughly how many there are to replay); `None` when it serves reads. + pub async fn rebuild_status(&mut self) -> Result> { // A listing filtered to the future matches nothing, so the probe costs little beyond the // rebuild check every read makes first. let future = OffsetDateTime::now_utc() + time::Duration::days(365); - let probe = list_sessions_request(&SessionFilter { + let filter = SessionFilter { updated_since: Some(future), ..SessionFilter::default() - }); + }; + let probe = ListSessionsRequest { + filter: Some((&filter).into()), + }; match self.client.list_sessions(probe).await { - Err(status) if crate::grpc::ai::session::is_rebuilding(&status) => { - Ok(Some(crate::grpc::ai::session::rebuild_progress(&status))) - } - Err(status) => Err(status.into()), Ok(_) => Ok(None), + Err(status) => match crate::grpc::ai::session::rebuild_progress(&status) { + Some(progress) => Ok(Some(progress)), + None => Err(status.into()), + }, } } @@ -579,7 +578,9 @@ impl AiClient { &mut self, filter: &SessionFilter, ) -> Result> { - let request = list_sessions_request(filter); + let request = ListSessionsRequest { + filter: Some(filter.into()), + }; Ok(self.client.list_sessions(request).await?.into_inner()) } @@ -618,7 +619,12 @@ impl AiClient { filter: &SessionFilter, limit: u32, ) -> Result> { - let request = search_sessions_request(query, terms, filter, limit); + let request = SearchSessionsRequest { + query: query.to_owned(), + limit, + any_term: terms == SearchTerms::Any, + filter: Some(filter.into()), + }; Ok(self.client.search_sessions(request).await?.into_inner()) } @@ -641,69 +647,6 @@ impl AiClient { } } -/// A listing of the sessions passing `filter`. The harness and `updated_since` go in the bare -/// fields as well as the filter, for a daemon from before `filter` (still running after an -/// upgrade), which ignores it; a newer one reads the filter's first, so the two never disagree. -fn list_sessions_request(filter: &SessionFilter) -> ListSessionsRequest { - ListSessionsRequest { - harness: filter.harness.map(|h| h as i32), - updated_since: filter.updated_since.map(|ts| prost_types::Timestamp { - seconds: ts.unix_timestamp(), - nanos: ts.nanosecond().cast_signed(), - }), - filter: Some(filter.into()), - } -} - -/// A search for `query` over the sessions passing `filter`, with the harness and workspace in the -/// bare fields too (see [`list_sessions_request`]). -fn search_sessions_request( - query: &str, - terms: SearchTerms, - filter: &SessionFilter, - limit: u32, -) -> SearchSessionsRequest { - SearchSessionsRequest { - query: query.to_owned(), - limit, - harness: filter.harness.map(|h| h as i32), - cwd: filter.workspace.as_ref().map(|w| w.to_string_lossy().into_owned()), - any_term: terms == SearchTerms::Any, - filter: Some(filter.into()), - } -} - -#[cfg(test)] -mod request_tests { - use atuin_client::ai_session::{HarnessKind, SearchTerms, SessionFilter}; - use rstest::rstest; - - use super::{list_sessions_request, search_sessions_request}; - - /// An older daemon reads only the legacy `harness` field, so it must carry the filter's. - #[rstest] - #[case::none(None)] - #[case::codex(Some(HarnessKind::Codex))] - fn requests_carry_the_harness_in_the_legacy_field(#[case] harness: Option) { - let filter = SessionFilter { - harness, - branch: Some("main".to_owned()), - ..SessionFilter::default() - }; - let expected = harness.map(|h| h as i32); - - let list = list_sessions_request(&filter); - assert_eq!(list.harness, expected); - assert_eq!(list.filter.as_ref().and_then(|f| f.harness), expected); - assert_eq!(list.filter.and_then(|f| f.branch).as_deref(), Some("main")); - - let search = search_sessions_request("words", SearchTerms::All, &filter, 7); - assert_eq!(search.harness, expected); - assert_eq!(search.filter.as_ref().and_then(|f| f.harness), expected); - assert_eq!((search.query.as_str(), search.limit), ("words", 7)); - } -} - #[cfg(all(test, unix))] mod tests { use std::path::Path; @@ -794,7 +737,7 @@ mod rebuild_tests { client .wait_for_sessions(|progress| { // The test's store replays nothing: the daemon reports that it has none to. - assert_eq!(progress, Some((0, 0))); + assert_eq!(progress, (0, 0)); counter.fetch_add(1, Ordering::SeqCst); }) .await @@ -833,7 +776,7 @@ mod rebuild_tests { async fn reads_the_rebuild_status_without_waiting() { let (mut client, state, _dir) = serve(StoreState::Recovering).await; // The test's store replays nothing: the daemon reports that it has none to. - assert_eq!(client.rebuild_status().await.unwrap(), Some(Some((0, 0)))); + assert_eq!(client.rebuild_status().await.unwrap(), Some((0, 0))); state.send_replace(StoreState::Ready); assert_eq!(client.rebuild_status().await.unwrap(), None); } diff --git a/crates/atuin-daemon/src/grpc/ai/agent/pb.rs b/crates/atuin-daemon/src/grpc/ai/agent/pb.rs index 1f979c287..dd24ff24a 100644 --- a/crates/atuin-daemon/src/grpc/ai/agent/pb.rs +++ b/crates/atuin-daemon/src/grpc/ai/agent/pb.rs @@ -363,12 +363,7 @@ impl TryFrom for DomainSession { }; let handle: DomainHarnessSession = value.handle.ok_or(ParseError::Missing("handle"))?.try_into()?; - let started_at = at(value.started_at, "started_at")?; - // Older daemons don't send it. - let atuin_id = match value.atuin_id { - Some(id) => atuin_id_domain(&id)?, - None => handle.atuin_id(started_at), - }; + let atuin_id = value.atuin_id.as_ref().ok_or(ParseError::Missing("atuin_id"))?; Ok(Self { handle, @@ -376,7 +371,7 @@ impl TryFrom for DomainSession { cwd: value.cwd.map(PathBuf::from), git_branch: value.git_branch, model: value.model, - started_at, + started_at: at(value.started_at, "started_at")?, updated_at: at(value.updated_at, "updated_at")?, message_count: value.message_count, usage: value.tokens.ok_or(ParseError::Missing("tokens"))?, @@ -391,7 +386,7 @@ impl TryFrom for DomainSession { child_count: value.child_count, // Not carried on the wire. group_updated_at: None, - atuin_id, + atuin_id: atuin_id_domain(atuin_id)?, parent_atuin_id: value.parent_atuin_id.as_ref().map(atuin_id_domain).transpose()?, root_atuin_id: value.root_atuin_id.as_ref().map(atuin_id_domain).transpose()?, child_atuin_ids: value diff --git a/crates/atuin-daemon/src/grpc/ai/session/mod.rs b/crates/atuin-daemon/src/grpc/ai/session/mod.rs index 5f9ffe990..7c9111515 100644 --- a/crates/atuin-daemon/src/grpc/ai/session/mod.rs +++ b/crates/atuin-daemon/src/grpc/ai/session/mod.rs @@ -3,7 +3,7 @@ pub mod pb; use std::pin::Pin; use std::sync::Arc; -use atuin_client::ai_session::SearchTerms; +use atuin_client::ai_session::{SearchTerms, SessionFilter}; use futures::StreamExt; use tokio_stream::Stream; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; @@ -16,8 +16,8 @@ use crate::grpc::ai::session::pb::{ GetSessionEvent, GetSessionRequest, HarnessFilterRequest, ImportSessionsEvent, ImportSessionsProgress, ImportSessionsRequest, ImportSessionsSummary, ListSessionsRequest, RebuildSessionsReply, RebuildSessionsRequest, SearchSessionsMatch, SearchSessionsRequest, - SessionFilterRequest, SessionRefRequest, TailSessionsEvent, TailSessionsRequest, - get_session_event, import_sessions_event, tail_sessions_event, + SessionRefRequest, TailSessionsEvent, TailSessionsRequest, get_session_event, + import_sessions_event, tail_sessions_event, }; use crate::grpc::common::pb as common; use crate::grpc::common::pb::Lagged; @@ -47,35 +47,30 @@ impl Service { } /// Metadata key marking the `Unavailable` status returned while startup recovery rebuilds AI -/// sessions. A dropped connection is `Unavailable` too; the marker is what tells a client this one -/// is worth waiting out. +/// sessions, carrying how far the rebuild has got as `/` records. A dropped +/// connection is `Unavailable` too; the marker is what tells a client this one is worth waiting +/// out. const REBUILDING_METADATA: &str = "atuin-ai-sessions-rebuilding"; -/// Metadata key on the rebuilding status carrying how far the rebuild has got, as -/// `/` records. -const REBUILD_PROGRESS_METADATA: &str = "atuin-ai-sessions-rebuild-progress"; - fn rebuilding_status((replayed, pending): (u64, u64)) -> Status { let mut status = Status::unavailable(format!( "AI sessions are being rebuilt from the record store ({replayed} of {pending} records); \ try again shortly" )); - let metadata = status.metadata_mut(); - metadata.insert(REBUILDING_METADATA, MetadataValue::from_static("1")); - if let Ok(progress) = MetadataValue::try_from(format!("{replayed}/{pending}")) { - metadata.insert(REBUILD_PROGRESS_METADATA, progress); - } + let progress = MetadataValue::try_from(format!("{replayed}/{pending}")) + .expect("digits and a slash are valid metadata"); + status.metadata_mut().insert(REBUILDING_METADATA, progress); status } -/// How far the rebuild `status` reports has got: records replayed, and roughly how many there are -/// to replay. `None` for any other status, or a daemon that does not say. +/// How far the daemon has got rebuilding AI sessions after starting, when `status` says it still +/// is: records replayed, and roughly how many there are to replay. `None` for any other status. #[must_use] pub fn rebuild_progress(status: &Status) -> Option<(u64, u64)> { - if !is_rebuilding(status) { + if status.code() != Code::Unavailable { return None; } - let value = status.metadata().get(REBUILD_PROGRESS_METADATA)?.to_str().ok()?; + let value = status.metadata().get(REBUILDING_METADATA)?.to_str().ok()?; let (replayed, pending) = value.split_once('/')?; Some((replayed.parse().ok()?, pending.parse().ok()?)) } @@ -83,7 +78,7 @@ pub fn rebuild_progress(status: &Status) -> Option<(u64, u64)> { /// Whether `status` says the daemon is still rebuilding AI sessions after starting. #[must_use] pub fn is_rebuilding(status: &Status) -> bool { - status.code() == Code::Unavailable && status.metadata().contains_key(REBUILDING_METADATA) + rebuild_progress(status).is_some() } #[tonic::async_trait] @@ -101,7 +96,7 @@ impl GrpcService for Service { request: Request, ) -> Result, Status> { self.ensure_recovered()?; - let filter = request.into_inner().filter()?; + let filter = SessionFilter::try_from(request.into_inner().filter.unwrap_or_default())?; let sessions = self .capture @@ -153,19 +148,23 @@ impl GrpcService for Service { request: Request, ) -> Result, Status> { self.ensure_recovered()?; - let request = request.into_inner(); - let filter = request.filter()?; + let SearchSessionsRequest { + query, + limit, + any_term, + filter, + } = request.into_inner(); + let filter = SessionFilter::try_from(filter.unwrap_or_default())?; - let terms = if request.any_term { + let terms = if any_term { SearchTerms::Any } else { SearchTerms::All }; - let stream = - self.capture.search(&request.query, terms, &filter, request.limit).map(|result| { - result.map(SearchSessionsMatch::from).map_err(|e| Status::internal(e.to_string())) - }); + let stream = self.capture.search(&query, terms, &filter, limit).map(|result| { + result.map(SearchSessionsMatch::from).map_err(|e| Status::internal(e.to_string())) + }); Ok(Response::new(Box::pin(stream))) } @@ -310,8 +309,6 @@ mod tests { .search_sessions(Request::new(SearchSessionsRequest { query: "anything".to_owned(), limit: 0, - harness: None, - cwd: None, any_term: false, filter: None, })) @@ -332,10 +329,11 @@ mod tests { .search_sessions(Request::new(SearchSessionsRequest { query: "x".to_owned(), limit: 0, - harness: Some(9999), - cwd: None, any_term: false, - filter: None, + filter: Some(pb::SessionFilter { + harness: Some(9999), + ..pb::SessionFilter::default() + }), })) .await; @@ -347,19 +345,11 @@ mod tests { async fn reads_are_refused_until_recovery_finishes() { let (cap, state) = AiHarnessSessionCapture::with_state(StoreState::Recovering).await; let svc = Service::new(Arc::new(cap)); - let list = || { - svc.list_sessions(Request::new(ListSessionsRequest { - harness: None, - updated_since: None, - filter: None, - })) - }; + let list = || svc.list_sessions(Request::new(ListSessionsRequest { filter: None })); let search = || { svc.search_sessions(Request::new(SearchSessionsRequest { query: "x".to_owned(), limit: 0, - harness: None, - cwd: None, any_term: false, filter: None, })) @@ -385,13 +375,7 @@ mod tests { drop(state); assert!( - svc.list_sessions(Request::new(ListSessionsRequest { - harness: None, - updated_since: None, - filter: None, - })) - .await - .is_ok() + svc.list_sessions(Request::new(ListSessionsRequest { filter: None })).await.is_ok() ); } diff --git a/crates/atuin-daemon/src/grpc/ai/session/pb.rs b/crates/atuin-daemon/src/grpc/ai/session/pb.rs index 08be86deb..61620f652 100644 --- a/crates/atuin-daemon/src/grpc/ai/session/pb.rs +++ b/crates/atuin-daemon/src/grpc/ai/session/pb.rs @@ -31,24 +31,12 @@ pub(crate) trait HarnessFilterRequest { } } -impl HarnessFilterRequest for ListSessionsRequest { - fn harness_filter(&self) -> Option { - self.harness - } -} - impl HarnessFilterRequest for TailSessionsRequest { fn harness_filter(&self) -> Option { self.harness } } -impl HarnessFilterRequest for SearchSessionsRequest { - fn harness_filter(&self) -> Option { - self.harness - } -} - impl From<&DomainSessionFilter> for SessionFilter { fn from(value: &DomainSessionFilter) -> Self { let path = |p: &Option| p.as_ref().map(|p| p.to_string_lossy().into_owned()); @@ -99,58 +87,6 @@ impl TryFrom for DomainSessionFilter { } } -/// A request carrying a [`SessionFilter`], and the older bare filters it supersedes (`harness`, -/// and a listing's `updated_since` or a search's `cwd`): each applies only where the filter leaves -/// its field unset, so an older client's request still filters as it asked. -pub(crate) trait SessionFilterRequest: HarnessFilterRequest { - fn session_filter(&self) -> Option; - - /// The bare `updated_since`, for a request that has one. - fn bare_updated_since(&self) -> Option { - None - } - - /// The bare `cwd` (the filter's `workspace`), for a request that has one. - fn bare_cwd(&self) -> Option<&str> { - None - } - - fn filter(&self) -> Result { - let mut filter: DomainSessionFilter = - self.session_filter().map(TryInto::try_into).transpose()?.unwrap_or_default(); - if filter.harness.is_none() { - filter.harness = self.harness()?; - } - if filter.updated_since.is_none() { - filter.updated_since = self.bare_updated_since().map(from_timestamp).transpose()?; - } - if filter.workspace.is_none() { - filter.workspace = self.bare_cwd().map(PathBuf::from); - } - Ok(filter) - } -} - -impl SessionFilterRequest for ListSessionsRequest { - fn session_filter(&self) -> Option { - self.filter.clone() - } - - fn bare_updated_since(&self) -> Option { - self.updated_since - } -} - -impl SessionFilterRequest for SearchSessionsRequest { - fn session_filter(&self) -> Option { - self.filter.clone() - } - - fn bare_cwd(&self) -> Option<&str> { - self.cwd.as_deref() - } -} - impl From for SearchSessionsMatch { fn from(value: SessionMatch) -> Self { Self { @@ -298,76 +234,6 @@ mod tests { assert_eq!(decoded, original); } - /// The filter's harness wins over the older bare field, which still applies alone. - #[rstest] - #[case::bare_field_alone(Some(HarnessKind::Pi), None, Some(HarnessKind::Pi))] - #[case::filter_wins(Some(HarnessKind::Pi), Some(HarnessKind::Codex), Some(HarnessKind::Codex))] - #[case::neither(None, None, None)] - fn the_filter_harness_supersedes_the_bare_one( - #[case] bare: Option, - #[case] filtered: Option, - #[case] expected: Option, - ) { - let request = SearchSessionsRequest { - query: String::new(), - limit: 0, - harness: bare.map(|h| h as i32), - cwd: None, - any_term: false, - filter: Some(SessionFilter { - harness: filtered.map(|h| h as i32), - ..SessionFilter::default() - }), - }; - assert_eq!(request.filter().unwrap().harness, expected); - } - - /// A search's bare `cwd` is the filter's workspace, which wins when both are set. - #[rstest] - #[case::bare_field_alone(Some("/a"), None, Some("/a"))] - #[case::filter_wins(Some("/a"), Some("/b"), Some("/b"))] - #[case::neither(None, None, None)] - fn the_filter_workspace_supersedes_the_bare_cwd( - #[case] bare: Option<&str>, - #[case] filtered: Option<&str>, - #[case] expected: Option<&str>, - ) { - let request = SearchSessionsRequest { - query: String::new(), - limit: 0, - harness: None, - cwd: bare.map(str::to_owned), - any_term: false, - filter: Some(SessionFilter { - workspace: filtered.map(str::to_owned), - ..SessionFilter::default() - }), - }; - assert_eq!(request.filter().unwrap().workspace, expected.map(PathBuf::from)); - } - - /// A listing's bare `updated_since` applies unless the filter sets its own. - #[rstest] - #[case::bare_field_alone(Some(10), None, Some(10))] - #[case::filter_wins(Some(10), Some(20), Some(20))] - #[case::neither(None, None, None)] - fn the_filter_updated_since_supersedes_the_bare_one( - #[case] bare: Option, - #[case] filtered: Option, - #[case] expected: Option, - ) { - let at = |secs: i64| OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(secs); - let request = ListSessionsRequest { - harness: None, - updated_since: bare.map(|s| timestamp(at(s))), - filter: Some(SessionFilter { - updated_since: filtered.map(|s| timestamp(at(s))), - ..SessionFilter::default() - }), - }; - assert_eq!(request.filter().unwrap().updated_since, expected.map(at)); - } - #[rstest] fn a_session_match_without_its_session_is_rejected() { let wire = SearchSessionsMatch { diff --git a/crates/atuin-daemon/src/session_capture/mod.rs b/crates/atuin-daemon/src/session_capture/mod.rs index a9ef39116..4f0447f30 100644 --- a/crates/atuin-daemon/src/session_capture/mod.rs +++ b/crates/atuin-daemon/src/session_capture/mod.rs @@ -434,11 +434,11 @@ struct UnsafeCalls { cwds: HashMap, } -/// Retain conversation text, and of tool calls what `tools` keeps: by default payload-free -/// breadcrumbs, never execution payloads (null payloads, which keep the wire format without -/// storing arguments or results). Text is redacted as `policy` does, and text that takes too long -/// to redact is not kept. This only affects new captures; existing synced records are not -/// rewritten. +/// Retain conversation text, and of tool calls what `tools` keeps: their inputs and outputs, +/// unless `ai.capture_tools` is off, `policy` withholds them, or they are over the size limit +/// (then null, which keeps the wire format without storing them). Reasoning is kept only as a +/// marker; raw blocks, and text on roles other than user and assistant, are dropped. Text is +/// redacted as `policy` does, and text that takes too long to redact is not kept. fn sanitize(msg: &mut Message, tools: ToolCapture, policy: &CapturePolicy) { let redact = |text: &mut String| match policy.redact(text) { Some(Cow::Borrowed(_)) => {} @@ -952,8 +952,6 @@ mod tests { } } assert_eq!(carried, [Some(fixed), Some(fixed)]); - // Minted, not derived from the session. - assert_ne!(fixed, first.session.atuin_id(first.timestamp)); assert_eq!(uuid::Uuid::from(fixed).get_version(), Some(uuid::Version::SortRand)); }