use super::assistant_test_support::{records, *};
use serde_json::json;
#[path = "assistant_replay_tests.rs"]
mod revision;
#[test]
fn disconnect_pushes_presence_false_and_releases_seat() -> Result {
let f = Fixture::new(json!({}))?;
let seat = f.seat("agent-pass", "wards")?;
let mut events = Stream::open(f.answer("GET", "/events", &json!({})))?;
assert!(events.event()?.contains("true"));
drop(seat);
let gone = events.event()?;
assert!(
gone.contains("presence") && gone.contains("false"),
"{gone}"
);
let replacement = f.seat("agent-pass", "wards")?;
drop(replacement);
events.close()
}
#[test]
fn two_tabs_receive_same_live_item() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let mut a = Stream::open(f.answer("GET", "/events", &json!({})))?;
let mut b = Stream::open(f.answer("GET", "/events", &json!({})))?;
a.event()?;
b.event()?;
seat.item(1)?;
let item = a.event()?;
assert_eq!(item, b.event()?);
assert!(item.contains("id: conversation:1"));
a.close()?;
b.close()
}
#[test]
fn client_hangup_frees_stream_slot_without_another_item() -> Result {
let f = Fixture::new(json!({"max_per_person":1}))?;
let seat = f.seat("agent-pass", "wards")?;
let mut stream = Stream::open(f.answer("GET", "/events", &json!({})))?;
stream.event()?;
check(
f.answer("GET", "/events", &json!({})),
429,
"max_per_person",
)?;
stream.close()?;
let mut replacement = Stream::open(f.answer("GET", "/events", &json!({})))?;
assert!(replacement.event()?.contains("presence"));
replacement.close()?;
drop(seat);
Ok(())
}
#[test]
fn replay_holds_live_items_and_deduplicates_seam() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let store = records::start(&f, records::rows(vec![item(6), item(7)]), move |rows| {
rows.lock()
.map_err(|_| anyhow::anyhow!("rows"))?
.insert(format!("{PERSON}/conversation/{:020}", 8), item(8));
seat.item(7)?;
seat.item(8)
})?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=5",
&json!({}),
))?;
assert!(stream.event()?.contains("presence"));
assert!(stream.event()?.contains("id: conversation:6"));
assert!(stream.event()?.contains("id: conversation:7"));
assert!(stream.event()?.contains("id: conversation:8"));
stream.close()?;
let calls = store.finish()?;
assert!(
calls
.iter()
.any(|v| v["operation"]["method"] == "documents.list")
);
Ok(())
}
#[test]
fn last_event_id_and_conversations_use_person_record_prefix() -> Result {
let f = Fixture::new(json!({}))?;
let store = super::assistant_test_support::record_store(
&f,
vec![
json!({"title":"Ward","harness":"codex","created_at_ms":1,"turns":3,"updated_at_ms":9,"state":"dormant"}),
],
|| Ok(()),
)?;
let out = written(f.answer("GET", "/conversations", &json!({})))?;
assert!(
out.contains("Ward") && out.contains("dormant") && out.contains("turns"),
"{out}"
);
let calls = store.join().map_err(|_| anyhow::anyhow!("store panic"))??;
assert!(
calls
.iter()
.any(|v| v["operation"]["params"]["prefix"] == format!("{PERSON}/"))
);
Ok(())
}
#[test]
fn pass_refresh_refusal_ends_stream_without_stopping_seat() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let mut stream = Stream::open(f.answer("GET", "/events", &json!({})))?;
stream.event()?;
let expired = f
.door
.pass(&f.cookie, crate::serve::pass::now() + 3601)
.err()
.ok_or_else(|| anyhow::anyhow!("expired pass admitted"))?;
assert_eq!(expired.code, "pass_expired");
seat.item(1)?;
let end = stream.event()?;
assert!(end.contains("pass_expired"), "{end}");
stream.close()?;
assert!(f.seat("agent-pass", "wards").is_err());
Ok(())
}
#[test]
fn last_event_id_replays_after_its_conversation_cursor() -> Result {
let f = Fixture::new(json!({}))?;
let store = record_store(&f, vec![item(6)], || Ok(()))?;
let mut request = f.request("GET", "/events?conversation=conversation", &json!({}));
request
.headers
.push(("last-event-id".into(), "conversation:5".into()));
let mut stream = Stream::open(super::route::answer(
&f.door,
&f.site,
&request,
crate::serve::pass::now(),
))?;
stream.event()?;
assert!(stream.event()?.contains("id: conversation:6"));
stream.close()?;
let calls = store.join().map_err(|_| anyhow::anyhow!("store panic"))??;
assert!(
calls.iter().any(|call| call["operation"]["params"]["after"]
== format!("{PERSON}/conversation/{:020}", 5))
);
Ok(())
}
#[test]
fn stream_store_bound_names_bound_and_count() -> Result {
let f = Fixture::new(json!({"max_per_store":1}))?;
let mut stream = Stream::open(f.answer("GET", "/events", &json!({})))?;
stream.event()?;
let refusal = written(f.answer("GET", "/events", &json!({})))?;
assert!(refusal.contains("max_per_store=1; count=1"), "{refusal}");
stream.close()
}
#[test]
fn live_gap_cannot_advance_the_durable_reconnect_cursor() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let store = records::start(&f, records::rows(vec![item(6), item(8)]), move |rows| {
for seq in [7, 9] {
rows.lock()
.map_err(|_| anyhow::anyhow!("rows"))?
.insert(format!("{PERSON}/conversation/{seq:020}"), item(seq));
}
seat.item(7)?;
seat.item(8)?;
seat.item(9)
})?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=5",
&json!({}),
))?;
stream.event()?;
for (seq, cursor) in [(6, Some(6)), (8, None), (7, Some(8)), (9, Some(9))] {
let event = stream.event()?;
let event = if event.contains("event: replayed") {
stream.event()?
} else {
event
};
assert!(event.contains(&format!("item {seq}")), "{event}");
if let Some(cursor) = cursor {
assert!(
event.contains(&format!("id: conversation:{cursor}\n")),
"{event}"
);
} else {
assert!(
!event.contains("id:"),
"gap advanced the reconnect cursor: {event}"
);
}
}
stream.close()?;
store.finish()?;
Ok(())
}
#[test]
fn a_lagged_stream_keeps_its_slot_until_it_closes() -> Result {
let f = Fixture::new(json!({"max_per_store":1}))?;
let hub = f.door.assistant()?;
let pair = (PERSON.to_owned(), "wards".to_owned());
let subscription = hub.subscribe(pair.clone())?;
hub.item(&pair, &item(1))?;
hub.item(&pair, &item(2))?;
let refused = hub
.subscribe(pair.clone())
.err()
.ok_or_else(|| anyhow::anyhow!("lagged stream released a live slot"))?;
assert!(refused.message.contains("max_per_store=1; count=1"));
drop(subscription);
drop(hub.subscribe(pair)?);
Ok(())
}