350 lines
12 KiB
Rust
350 lines
12 KiB
Rust
//! Round-trip and recovery tests for the on-disk sidecar session store.
|
|
|
|
mod common;
|
|
|
|
use std::collections::HashMap;
|
|
use std::path::PathBuf;
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use maestro_daemon::sidecar::SidecarStore;
|
|
use maestro_protocol::{Session, SessionStatus, SessionView};
|
|
use serde_json::Value;
|
|
use tempfile::TempDir;
|
|
use uuid::Uuid;
|
|
|
|
fn sample(id: Uuid) -> Session {
|
|
Session {
|
|
id,
|
|
cwd: PathBuf::from("/tmp/work"),
|
|
command: "bash".into(),
|
|
env: HashMap::from([("FOO".into(), "bar".into())]),
|
|
metadata: HashMap::from([("name".into(), Value::String("maestro-impl".into()))]),
|
|
worktree: None,
|
|
status: SessionStatus::Running,
|
|
created_at: DateTime::parse_from_rfc3339("2026-05-24T12:00:00Z")
|
|
.unwrap()
|
|
.with_timezone(&Utc),
|
|
foreground: None,
|
|
activity: Default::default(),
|
|
activity_detail: None,
|
|
last_activity_at: chrono::Utc::now(),
|
|
activity_since: chrono::Utc::now(),
|
|
current_cwd: Some(PathBuf::from("/tmp/work/crates")),
|
|
blocked_work: None,
|
|
last_focused_at: None,
|
|
acp_session_id: None,
|
|
view: SessionView::Terminal,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn write_then_read_round_trip() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let s = sample(id);
|
|
|
|
store.write(&s).expect("write");
|
|
let loaded = store.read(id).expect("read");
|
|
assert_eq!(loaded.as_ref(), Some(&s));
|
|
|
|
let path = store.path_for(id);
|
|
assert!(path.exists());
|
|
assert!(path.ends_with(format!("{id}.json")));
|
|
}
|
|
|
|
#[test]
|
|
fn read_missing_returns_none() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
assert!(store.read(Uuid::new_v4()).unwrap().is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn delete_removes_file_and_is_idempotent() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let s = sample(id);
|
|
|
|
store.write(&s).unwrap();
|
|
store.mark_for_recovery(id).unwrap();
|
|
assert!(store.path_for(id).exists());
|
|
assert!(store.recovery_path_for(id).exists());
|
|
|
|
store.delete(id).unwrap();
|
|
assert!(!store.path_for(id).exists());
|
|
assert!(!store.recovery_path_for(id).exists());
|
|
|
|
// second delete is a no-op
|
|
store.delete(id).unwrap();
|
|
}
|
|
|
|
#[test]
|
|
fn recovery_marker_is_consumed_once() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
|
|
assert!(!store.take_recovery_marker(id).unwrap());
|
|
store.mark_for_recovery(id).unwrap();
|
|
assert!(store.take_recovery_marker(id).unwrap());
|
|
assert!(!store.take_recovery_marker(id).unwrap());
|
|
}
|
|
|
|
#[test]
|
|
fn load_all_recovers_every_sidecar_and_skips_garbage() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
|
|
let s1 = sample(Uuid::new_v4());
|
|
let s2 = sample(Uuid::new_v4());
|
|
store.write(&s1).unwrap();
|
|
store.write(&s2).unwrap();
|
|
|
|
// garbage files in the directory should be ignored
|
|
std::fs::write(dir.path().join("not-json.txt"), b"hello").unwrap();
|
|
std::fs::write(dir.path().join("broken.json"), b"{not json}").unwrap();
|
|
|
|
let mut loaded = store.load_all().expect("load_all");
|
|
loaded.sort_by_key(|s| s.id);
|
|
let mut expected = vec![s1, s2];
|
|
expected.sort_by_key(|s| s.id);
|
|
assert_eq!(loaded, expected);
|
|
}
|
|
|
|
#[test]
|
|
fn load_all_returns_empty_when_directory_missing() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path().join("does-not-exist"));
|
|
assert!(store.load_all().unwrap().is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn sidecar_missing_view_key_reads_as_terminal() {
|
|
// A pre-#133 sidecar on disk has no `view` field at all.
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
|
|
let mut value = serde_json::to_value(sample(id)).unwrap();
|
|
value.as_object_mut().unwrap().remove("view");
|
|
std::fs::write(store.path_for(id), serde_json::to_vec(&value).unwrap()).unwrap();
|
|
|
|
let loaded = store.read(id).expect("read").expect("present");
|
|
assert_eq!(loaded.view, SessionView::Terminal);
|
|
}
|
|
|
|
#[test]
|
|
fn sidecar_structured_view_round_trips() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let mut s = sample(id);
|
|
s.view = SessionView::Structured;
|
|
|
|
store.write(&s).expect("write");
|
|
let loaded = store.read(id).expect("read");
|
|
assert_eq!(
|
|
loaded.as_ref().map(|s| s.view),
|
|
Some(SessionView::Structured)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn write_creates_parent_directory() {
|
|
let dir = TempDir::new().unwrap();
|
|
let nested = dir.path().join("a/b/c");
|
|
let store = SidecarStore::new(&nested);
|
|
let s = sample(Uuid::new_v4());
|
|
store.write(&s).expect("write");
|
|
assert!(nested.exists());
|
|
assert!(store.path_for(s.id).exists());
|
|
}
|
|
|
|
// ---- per-session event log (#135) ----
|
|
|
|
use maestro_acp::schema::{ContentBlock, SessionNotification, SessionUpdate, TextContent};
|
|
use maestro_daemon::sidecar::EventCursor;
|
|
|
|
fn note(text: &str) -> SessionNotification {
|
|
SessionNotification::new(
|
|
"acp-1",
|
|
SessionUpdate::AgentMessageChunk(maestro_acp::schema::ContentChunk::new(
|
|
ContentBlock::Text(TextContent::new(text)),
|
|
)),
|
|
)
|
|
}
|
|
|
|
fn text_of(n: &SessionNotification) -> &str {
|
|
match &n.update {
|
|
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
|
ContentBlock::Text(t) => &t.text,
|
|
other => panic!("unexpected content {other:?}"),
|
|
},
|
|
other => panic!("unexpected update {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn append_assigns_line_position_as_seq_and_reads_back() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
|
|
let mut log = store.open_events(id).expect("open");
|
|
assert_eq!(log.last_seq(), 0);
|
|
assert_eq!(log.append(¬e("a")).unwrap(), 1);
|
|
assert_eq!(log.append(¬e("b")).unwrap(), 2);
|
|
assert_eq!(log.append(¬e("c")).unwrap(), 3);
|
|
assert_eq!(log.last_seq(), 3);
|
|
|
|
let path = store.events_path_for(id);
|
|
assert!(path.ends_with(format!("{id}.events.jsonl")), "{path:?}");
|
|
let raw = std::fs::read_to_string(&path).unwrap();
|
|
assert_eq!(raw.lines().count(), 3, "one object per line:\n{raw}");
|
|
for line in raw.lines() {
|
|
let v: Value = serde_json::from_str(line).expect("each line is one JSON object");
|
|
assert_eq!(v["sessionId"], "acp-1");
|
|
}
|
|
|
|
let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
|
|
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
|
|
assert_eq!(
|
|
all.iter()
|
|
.map(|e| text_of(&e.notification))
|
|
.collect::<Vec<_>>(),
|
|
vec!["a", "b", "c"]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn since_reads_forward_and_before_reads_backward_with_limit() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let mut log = store.open_events(id).unwrap();
|
|
for i in 1..=6 {
|
|
log.append(¬e(&format!("n{i}"))).unwrap();
|
|
}
|
|
|
|
// `since` is exclusive: what a client that already holds seq 2 needs.
|
|
let fwd = store.read_events(id, EventCursor::Since(2), 3).unwrap();
|
|
assert_eq!(fwd.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4, 5]);
|
|
// Past the end: nothing, not an error.
|
|
assert!(store
|
|
.read_events(id, EventCursor::Since(6), 3)
|
|
.unwrap()
|
|
.is_empty());
|
|
|
|
// `before` is exclusive and pages backward: the `limit` events closest
|
|
// to the cursor, still in ascending seq order.
|
|
let back = store
|
|
.read_events(id, EventCursor::Before(Some(5)), 2)
|
|
.unwrap();
|
|
assert_eq!(back.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4]);
|
|
// No cursor: the tail of the log, for a first paint.
|
|
let tail = store.read_events(id, EventCursor::Before(None), 2).unwrap();
|
|
assert_eq!(tail.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![5, 6]);
|
|
// Paging past the start runs out cleanly.
|
|
let start = store
|
|
.read_events(id, EventCursor::Before(Some(2)), 10)
|
|
.unwrap();
|
|
assert_eq!(start.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
|
|
assert!(store
|
|
.read_events(id, EventCursor::Before(Some(1)), 10)
|
|
.unwrap()
|
|
.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn missing_log_reads_as_empty() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
assert!(store
|
|
.read_events(id, EventCursor::Since(0), 10)
|
|
.unwrap()
|
|
.is_empty());
|
|
assert!(store
|
|
.read_events(id, EventCursor::Before(None), 10)
|
|
.unwrap()
|
|
.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn malformed_lines_are_skipped_but_keep_their_seq() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let mut log = store.open_events(id).unwrap();
|
|
log.append(¬e("first")).unwrap();
|
|
drop(log);
|
|
|
|
// Garbage in the middle and a line that is JSON but not a notification.
|
|
let path = store.events_path_for(id);
|
|
let mut raw = std::fs::read_to_string(&path).unwrap();
|
|
raw.push_str("{not json\n");
|
|
raw.push_str("{\"unrelated\": true}\n");
|
|
std::fs::write(&path, raw).unwrap();
|
|
let mut log = store.open_events(id).unwrap();
|
|
assert_eq!(log.last_seq(), 3);
|
|
assert_eq!(log.append(¬e("fourth")).unwrap(), 4);
|
|
|
|
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
|
|
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
|
|
assert_eq!(text_of(&all[1].notification), "fourth");
|
|
}
|
|
|
|
#[test]
|
|
fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
|
|
// A hard kill mid-write leaves a partial line with no newline.
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
let mut log = store.open_events(id).unwrap();
|
|
log.append(¬e("whole")).unwrap();
|
|
drop(log);
|
|
|
|
let path = store.events_path_for(id);
|
|
let mut raw = std::fs::read(&path).unwrap();
|
|
raw.extend_from_slice(br#"{"sessionId":"acp-1","update":{"sessionUpd"#);
|
|
std::fs::write(&path, &raw).unwrap();
|
|
|
|
// Reads: the partial line is skipped, nothing else is lost.
|
|
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
|
|
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
|
|
|
|
// Reopening terminates the partial line so the next append does not
|
|
// glue onto it; the partial keeps a seq of its own.
|
|
let mut log = store.open_events(id).unwrap();
|
|
assert_eq!(log.last_seq(), 2);
|
|
assert_eq!(log.append(¬e("after")).unwrap(), 3);
|
|
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
|
|
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
|
|
assert_eq!(text_of(&all[1].notification), "after");
|
|
let raw = std::fs::read_to_string(&path).unwrap();
|
|
assert_eq!(raw.lines().count(), 3, "{raw}");
|
|
}
|
|
|
|
#[test]
|
|
fn delete_removes_the_event_log_and_load_all_ignores_it() {
|
|
let dir = TempDir::new().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
store.write(&sample(id)).unwrap();
|
|
let mut log = store.open_events(id).unwrap();
|
|
log.append(¬e("x")).unwrap();
|
|
drop(log);
|
|
assert!(store.events_path_for(id).exists());
|
|
|
|
let loaded = store.load_all().unwrap();
|
|
assert_eq!(loaded.len(), 1, "the .jsonl must not be read as a sidecar");
|
|
|
|
store.delete(id).unwrap();
|
|
assert!(!store.path_for(id).exists());
|
|
assert!(!store.events_path_for(id).exists());
|
|
// Idempotent with the log already gone.
|
|
store.delete(id).unwrap();
|
|
}
|