use super::{Conversation, Record};
use crate::assistant::{
config::Seat,
door_link::Link,
handshake::{Session, Version},
tools::Identity,
};
use serde_json::{Value, json};
use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader};
async fn fixture() -> anyhow::Result<(
Record,
Link,
BufReader<tokio::net::UnixStream>,
tempfile::TempDir,
)> {
use std::os::unix::fs::PermissionsExt as _;
let directory = tempfile::tempdir()?;
std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o700))?;
let path = directory.path().join("door.sock");
let listener = tokio::net::UnixListener::bind(&path)?;
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?;
let accepting = tokio::spawn(async move {
let mut peer = BufReader::new(listener.accept().await?.0);
let mut hello = String::new();
peer.read_line(&mut hello).await?;
peer.get_mut()
.write_all(b"{\"welcome\":{\"person\":\"person\"}}\n")
.await?;
Ok::<_, anyhow::Error>(peer)
});
let config = directory.path().join("serve.json");
std::fs::write(
&config,
serde_json::to_vec(
&json!({"data_dir":directory.path(),"socket":directory.path().join("unused.sock"),"name":"fixture","branch_cache_bytes_per_shard":1_048_576,"max_frame":1_048_576,"max_clients":4,"max_page":1000,"timeout_ms":15_000}),
)?,
)?;
let seat: Seat = serde_json::from_value(
json!({"harness":{"id":"codex","home":directory.path(),"program":"/bin/true","args":[],"write_queue":128},"door_socket":path,"serve_config":config,"issuer":"http://127.0.0.1:1","store":"fixture"}),
)?;
let identity = Identity {
agent: "agent".into(),
launch: "proof".into(),
pass: "pass".into(),
};
let session = Session {
id: "a".into(),
version: Version::One,
resumable: true,
capability: "agentCapabilities.loadSession",
updates: Vec::new(),
config: json!([]),
info: json!({}),
};
let link = Link::open(&seat, &identity, &session).await?;
let peer = accepting.await??;
let mut record = Record::new(
seat,
&identity,
"person".into(),
json!({}),
link.output.clone(),
)?;
for id in ["a", "b"] {
record.conversations.insert(
id.into(),
Conversation {
seq: 0,
live_seq: 0,
meta: json!({"title":"title","turns":0,"last_seq":0,"state":"live"}),
},
);
}
Ok((record, link, peer, directory))
}
async fn close(mut link: Link) -> anyhow::Result<()> {
crate::assistant::closing::stop(&mut link.reader, "reader").await?;
crate::assistant::closing::stop(&mut link.writer, "writer")
.await?
.unwrap_or(Ok(()))
}
fn text(id: &str, words: &str, replayed: bool) -> Value {
json!({"item":{"conversation":id,"kind":"text","role":"agent","text":words,"message_id":null,"replayed":replayed}})
}
#[tokio::test]
async fn round4_4_completed_message_and_interrupted_tool_advance_past_live_sequences()
-> anyhow::Result<()> {
let (mut record, link, _peer, _directory) = fixture().await?;
record.accept(text("a", "first ", false))?;
record.accept(json!({"item":{"conversation":"a","kind":"tool_call","tool_call_id":"tool","status":"in_progress"}}))?;
record.accept(text("a", "second", false))?;
let live = record.conversations["a"].seq;
record.finish_messages("a")?;
assert_eq!(record.completed.len(), 2);
assert!(
record
.completed
.iter()
.all(|item| item["seq"].as_u64().is_some_and(|seq| seq > live)),
"durable completion is hidden by the panel's live cursor"
);
assert_ne!(record.completed[0]["seq"], record.completed[1]["seq"]);
assert!(
record
.completed
.iter()
.any(|item| item["text"] == "first second")
);
assert!(
record
.completed
.iter()
.any(|item| item["status"] == "interrupted")
);
close(link).await
}
#[tokio::test]
async fn round4_5_replay_never_enters_the_record_or_a_live_message() -> anyhow::Result<()> {
let (mut record, link, _peer, _directory) = fixture().await?;
record.accept(text("a", "live", false))?;
record.accept(text("a", "history", true))?;
assert_eq!(record.messages.len(), 1);
let message = record
.messages
.values()
.next()
.ok_or_else(|| anyhow::anyhow!("live message absent"))?;
assert_eq!(
message["text"], "live",
"history was appended to a live answer"
);
assert!(record.completed.iter().all(|item| item["replayed"] != true));
record.accept(text("b", "earlier", true))?;
assert!(
record
.messages
.keys()
.all(|(conversation, _)| conversation != "b"),
"replay was retained for persistence"
);
close(link).await
}
#[tokio::test]
async fn round4_7_record_refusal_names_the_affected_conversation() -> anyhow::Result<()> {
let (mut record, link, mut peer, _directory) = fixture().await?;
record
.completed
.push(json!({"kind":"request","conversation":"b","seq":1}));
record.failed("permission_refused")?;
let mut frame = String::new();
peer.read_line(&mut frame).await?;
let failure: Value = serde_json::from_str(&frame)?;
assert_eq!(
failure["item"]["conversation"], "b",
"record failure was assigned to the first cached conversation"
);
close(link).await
}
#[tokio::test]
async fn round4_7_retryable_write_failure_is_live_only() -> anyhow::Result<()> {
let (mut record, link, _peer, _directory) = fixture().await?;
record
.completed
.push(json!({"kind":"request","conversation":"b","seq":1}));
record.failed("permission_refused")?;
assert!(
record
.completed
.iter()
.all(|item| item["source"] != "record"),
"a temporary refusal would be persisted after successful retry"
);
close(link).await
}
#[tokio::test]
async fn round4_buffer_sizing_does_linear_work_and_preserves_the_bound() -> anyhow::Result<()> {
let (mut record, link, peer, directory) = fixture().await?;
for index in 0..100 {
record.accept(json!({"item":{"conversation":"a","kind":"request","request_id":index,"text":"request"}}))?;
}
let final_bytes = serde_json::to_vec(&record.completed)?.len();
assert!(
record.measured_bytes.get() <= final_bytes * 3,
"record sizing repeatedly serialized the retained buffer: {} bytes visited for {} bytes",
record.measured_bytes.get(),
final_bytes
);
record.seat.harness.content_bytes = 1;
record.seat.harness.write_queue = 1;
assert!(
record.check_bound().is_err(),
"cached accounting lost the actual byte bound"
);
close(link).await?;
drop(peer);
drop(directory);
Ok(())
}
#[tokio::test]
async fn round4_10_a_content_only_tool_update_after_replay_stays_pending() -> anyhow::Result<()> {
let (mut record, link, _peer, _directory) = fixture().await?;
record.accept(json!({"item":{"conversation":"a","kind":"tool_call","tool_call_id":"tool","status":"pending","replayed":true}}))?;
record.accept(json!({"item":{"conversation":"a","kind":"tool_call","tool_call_id":"tool","title":"Reading"}}))?;
assert_eq!(record.tools.len(), 1);
assert!(record.completed.is_empty());
record.finish_messages("a")?;
assert_eq!(record.completed.len(), 1);
assert_eq!(record.completed[0]["status"], "interrupted");
close(link).await
}