maestro/crates/maestro-daemon/tests/sidecar.rs

//! 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::{PersistedEvent, SidecarStore};
use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
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,
        auto_approve: false,
        queued_prompts: Vec::new(),
    }
}

#[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:?}"),
    }
}

fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
        panic!("expected ACP notification")
    };
    text_of(notification)
}

#[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(note("a")).unwrap(), 1);
    assert_eq!(log.append(note("b")).unwrap(), 2);
    assert_eq!(log.append(note("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(event_text).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(note(&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(note("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(note("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!(event_text(&all[1]), "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(note("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(note("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!(event_text(&all[1]), "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(note("x")).unwrap();
    drop(log);
    let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl"));
    std::fs::write(&legacy_approvals, b"obsolete").unwrap();
    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());
    assert!(!legacy_approvals.exists());
    // Idempotent with the log already gone.
    store.delete(id).unwrap();
}

#[test]
fn approval_rows_share_event_cursors_with_legacy_acp_lines() {
    let dir = TempDir::new().unwrap();
    let store = SidecarStore::new(dir.path());
    let id = Uuid::new_v4();
    let legacy = serde_json::to_vec(&note("before")).unwrap();
    std::fs::create_dir_all(dir.path()).unwrap();
    std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap();

    let mut log = store.open_events(id).unwrap();
    assert_eq!(
        log.append(PersistedEvent::Approval {
            row: TranscriptRow {
                id: "approval:secret".into(),
                kind: TranscriptRowKind::Approval,
                data: serde_json::json!({"tool": "bash"}),
            },
        })
        .unwrap(),
        2
    );
    assert_eq!(log.append(note("after")).unwrap(), 3);

    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
    assert_eq!(
        all.iter().map(|event| event.seq).collect::<Vec<_>>(),
        [1, 2, 3]
    );
    assert!(matches!(all[1].event, PersistedEvent::Approval { .. }));
    assert_eq!(event_text(&all[0]), "before");
    assert_eq!(event_text(&all[2]), "after");

    let since = store.read_events(id, EventCursor::Since(1), 1).unwrap();
    assert_eq!(since[0].seq, 2);
    let before = store
        .read_events(id, EventCursor::Before(Some(3)), 1)
        .unwrap();
    assert_eq!(before[0].seq, 2);
}