refactor(ai-session): drop compatibility with records, checkpoints and daemons from earlier builds (atuinsh/atuin#4373)

This commit is contained in:
Ellie Huxtable
2026-10-07 15:43:12 +00:00
committed by Ellie Huxtable
parent ca22c7b422
commit 0abaa3ec5a
36 changed files with 455 additions and 1128 deletions
Generated
-1
View File
@@ -474,7 +474,6 @@ dependencies = [
"url",
"uuid",
"wiremock",
"xxhash-rust",
"zstd",
]
+17 -29
View File
@@ -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<Summary> {
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<u64>) -> ContentJson {
match block {
/// `None` for what capture never stores (reasoning text, raw blocks).
fn content_json(block: &Content, reasoning: Option<u64>) -> Option<ContentJson> {
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<u64>) -> 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]
+14 -3
View File
@@ -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,
+9 -11
View File
@@ -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);
}
+11 -1
View File
@@ -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())])
+3 -7
View File
@@ -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)
}
+1 -1
View File
@@ -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";
+6 -9
View File
@@ -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<ParentKind>) -> &'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<ParentKind>, #[case] want: &str) {
assert_eq!(relation(kind), want);
}
+13 -53
View File
@@ -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!("<permissions instructions>sandbox rules{}", " policy".repeat(2_000)),
),
// The Codex parser files `<environment_context>` under the system role.
text(Role::System, "<environment_context><cwd>/x</cwd></environment_context>"),
text(Role::User, "fix this"),
];
let abridged = render(json!({"session_id": "abc"}), &msgs);
assert!(abridged.contains("#0 developer"), "{abridged}");
assert!(
abridged.contains("(harness) <permissions instructions>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<_> =
-1
View File
@@ -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 }
+49 -83
View File
@@ -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<ParentKind>,
) {
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<String> = 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();
@@ -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)?;
+40 -339
View File
@@ -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<i64> {
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<HarnessSession>,
/// 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<ParentKind>,
#[builder(default)]
pub parent_source_id: Option<SourceId>,
@@ -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<String>,
/// Where [`Self::session_title`] came from, so a resumed capture keeps ranking it.
#[builder(default)]
#[serde(default)]
pub session_title_source: Option<TitleSource>,
/// 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<TitleChange>,
/// 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<String>,
/// `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<AtuinSessionId>,
/// 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<Option<ParentKind>, 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::<Raw>::deserialize(deserializer)? {
Some(Raw::Name(name)) => {
let name: StringDeserializer<Error> = name.into_deserializer();
ParentKind::deserialize(name).ok()
}
Some(Raw::Other(_)) | None => None,
})
Ok(Option::<String>::deserialize(deserializer)?.and_then(|name| {
let name: StringDeserializer<Error> = name.into_deserializer();
ParentKind::deserialize(name).ok()
}))
}
impl From<Message> 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<ParentKind> {
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<ParentKind> {
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<Value = Content> {
"[a-zA-Z0-9 ]{0,16}".prop_map(Content::Text)
}
fn arb_message() -> impl Strategy<Value = Message> {
("[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<Content>,
}
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<ParentKind>,
) {
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<ParentKind>,
) {
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::<u128>(), 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)]
+46 -10
View File
@@ -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`
@@ -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<String> {
(!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<Value> {
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<Val
.iter()
.filter_map(|c| match c {
Content::ToolResult(r) => Some(tool_result_block(r)),
other => user_block(other),
_ => None,
})
.collect();
if blocks.is_empty() {
@@ -81,12 +81,8 @@ fn carried(messages: &[RehydrateMessage]) -> Vec<(String, Role, Vec<Content>)> {
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"));
}
@@ -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, `<environment_context>`, ...);
//! - 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` (`<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<Value> {
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<Value> = message
@@ -612,8 +602,6 @@ fn text_message(message: &RehydrateMessage, at: &str) -> Vec<Value> {
.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<String> {
.iter()
.map(|content| match content {
Content::Text(text) => Some(text.as_str()),
Content::Other(_) => Some(""),
_ => None,
})
.collect::<Option<_>>()?;
@@ -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<Turn<'_>>, 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<Turn<'_>>, 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()
@@ -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<RehydrateMessage>) -> 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();
+3 -3
View File
@@ -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:
//!
@@ -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<Content>)> {
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
+5 -5
View File
@@ -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 `<name>` <compact arguments>`` |
///
/// 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 `<name>` ``. 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<String>, detailed: &dyn Fn(&str) -> String, bare: &str| {
Some(what.map_or_else(|| bare.to_owned(), |w| detailed(&w)))
};
@@ -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<String> {
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 `<message>/<part>`, and a failure is keyed on its message.
/// The message a row names as its own: a row of opencode 2.0 is keyed `<message>/<part>`, and a
/// failure is keyed on its message.
fn named_message(row: &RehydrateMessage) -> Option<String> {
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<Value> {
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<Value> {
"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<RehydrateMessage>, Vec<RehydrateMessage>) {
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<RehydrateMessage> = 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<RehydrateMessage> {
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");
@@ -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<Mark>) -> 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<Mark>, 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<Mark>, Option<Mark>) {
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),
)
}
@@ -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();
@@ -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<Value> {
content
.iter()
@@ -393,9 +381,6 @@ fn user_blocks(content: &[Content]) -> Vec<Value> {
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()
@@ -61,7 +61,7 @@ fn carried(messages: &[RehydrateMessage]) -> Vec<(String, Role, Vec<Content>)> {
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<Content>)> {
.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<PiMessage> {
#[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()));
@@ -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<Content> {
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<Content> {
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<RehydrateMessage>) -> Vec<RehydrateMessage> {
pub fn synced(rows: Vec<RehydrateMessage>) -> Vec<RehydrateMessage> {
synced_as(rows, false)
}
/// `rows` as capture syncs them, keeping tool payloads when `tools` (see [`sanitize_as`]).
pub fn synced_as(mut rows: Vec<RehydrateMessage>, tools: bool) -> Vec<RehydrateMessage> {
for row in &mut rows {
row.content = sanitize(&row.role, &row.content);
row.content = sanitize_as(&row.role, &row.content, tools);
}
rows
}
@@ -423,8 +423,8 @@ fn content_row(id: &str, role: Role, content: Vec<Content>) -> 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()));
+16 -19
View File
@@ -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;
+3 -14
View File
@@ -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 {
+26 -83
View File
@@ -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<Option<Option<(u64, u64)>>> {
/// replayed, and roughly how many there are to replay); `None` when it serves reads.
pub async fn rebuild_status(&mut self) -> Result<Option<(u64, u64)>> {
// 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<tonic::Streaming<AiSession>> {
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<tonic::Streaming<SearchSessionsMatch>> {
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<HarnessKind>) {
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);
}
+3 -8
View File
@@ -363,12 +363,7 @@ impl TryFrom<Session> 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<Session> 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<Session> 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
+32 -48
View File
@@ -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 `<replayed>/<to replay>` 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
/// `<replayed>/<to replay>` 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<ListSessionsRequest>,
) -> Result<Response<Self::ListSessionsStream>, 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<SearchSessionsRequest>,
) -> Result<Response<Self::SearchSessionsStream>, 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()
);
}
@@ -31,24 +31,12 @@ pub(crate) trait HarnessFilterRequest {
}
}
impl HarnessFilterRequest for ListSessionsRequest {
fn harness_filter(&self) -> Option<i32> {
self.harness
}
}
impl HarnessFilterRequest for TailSessionsRequest {
fn harness_filter(&self) -> Option<i32> {
self.harness
}
}
impl HarnessFilterRequest for SearchSessionsRequest {
fn harness_filter(&self) -> Option<i32> {
self.harness
}
}
impl From<&DomainSessionFilter> for SessionFilter {
fn from(value: &DomainSessionFilter) -> Self {
let path = |p: &Option<PathBuf>| p.as_ref().map(|p| p.to_string_lossy().into_owned());
@@ -99,58 +87,6 @@ impl TryFrom<SessionFilter> 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<SessionFilter>;
/// The bare `updated_since`, for a request that has one.
fn bare_updated_since(&self) -> Option<prost_types::Timestamp> {
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<DomainSessionFilter, ParseError> {
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<SessionFilter> {
self.filter.clone()
}
fn bare_updated_since(&self) -> Option<prost_types::Timestamp> {
self.updated_since
}
}
impl SessionFilterRequest for SearchSessionsRequest {
fn session_filter(&self) -> Option<SessionFilter> {
self.filter.clone()
}
fn bare_cwd(&self) -> Option<&str> {
self.cwd.as_deref()
}
}
impl From<SessionMatch> 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<HarnessKind>,
#[case] filtered: Option<HarnessKind>,
#[case] expected: Option<HarnessKind>,
) {
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<i64>,
#[case] filtered: Option<i64>,
#[case] expected: Option<i64>,
) {
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 {
@@ -434,11 +434,11 @@ struct UnsafeCalls {
cwds: HashMap<HarnessSession, PathBuf>,
}
/// 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));
}