use std::collections::{BTreeMap, HashSet};
use std::io::Write;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use axum::{
extract::Path as AxumPath,
http::StatusCode,
routing::{get, post},
Extension, Json, Router,
};
use serde::{Deserialize, Serialize};
use serde_json::json;
use crate::ApiError;
pub(crate) const CONVERSATIONS: &str = "/v1/conversations";
pub(crate) const CONVERSATION: &str = "/v1/conversations/{conversation_id}";
pub(crate) const CONVERSATION_DELETE: &str = "/v1/conversations/{conversation_id}/delete";
const MAX_CONVERSATIONS: usize = 500;
const MAX_MESSAGES: usize = 2_000;
const MAX_CONTENT_BYTES: usize = 1 << 20;
const MAX_CONVERSATION_BYTES: usize = 8 << 20;
const MAX_MESSAGE_ID_LEN: usize = 128;
const MAX_TITLE_CHARS: usize = 200;
const MAX_APPEND: usize = 256;
const DERIVED_TITLE_CHARS: usize = 60;
const ROLES: [&str; 3] = ["user", "assistant", "system"];
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct MessageNode {
pub(crate) id: String,
#[serde(default)]
pub(crate) parent_id: Option<String>,
pub(crate) role: String,
#[serde(default)]
pub(crate) content: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) reasoning_content: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) reasoning_ms: Option<u64>,
pub(crate) created_at: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) metadata: Option<serde_json::Value>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Conversation {
pub(crate) id: String,
#[serde(default)]
pub(crate) title: Option<String>,
#[serde(default)]
pub(crate) model: Option<String>,
pub(crate) created_at: u64,
pub(crate) updated_at: u64,
#[serde(default)]
pub(crate) head_id: Option<String>,
#[serde(default)]
pub(crate) messages: Vec<MessageNode>,
}
impl Conversation {
fn summary(&self) -> Summary {
Summary {
object: "conversation.summary",
id: self.id.clone(),
title: self.title.clone(),
model: self.model.clone(),
created_at: self.created_at,
updated_at: self.updated_at,
head_id: self.head_id.clone(),
message_count: self.messages.len(),
}
}
fn has(&self, id: &str) -> bool {
self.messages.iter().any(|m| m.id == id)
}
}
#[derive(Debug, Serialize)]
struct ConversationBody {
object: &'static str,
#[serde(flatten)]
conversation: Conversation,
}
impl From<Conversation> for ConversationBody {
fn from(conversation: Conversation) -> Self {
Self {
object: "conversation",
conversation,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct Summary {
object: &'static str,
id: String,
title: Option<String>,
model: Option<String>,
created_at: u64,
updated_at: u64,
head_id: Option<String>,
message_count: usize,
}
#[derive(Debug, Serialize)]
struct ListBody {
object: &'static str,
data: Vec<Summary>,
}
#[derive(Debug, Serialize)]
struct DeletedBody {
object: &'static str,
id: String,
deleted: bool,
}
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct NewMessage {
pub(crate) id: String,
#[serde(default)]
pub(crate) parent_id: Option<String>,
pub(crate) role: String,
#[serde(default)]
pub(crate) content: String,
#[serde(default)]
pub(crate) reasoning_content: Option<String>,
#[serde(default)]
pub(crate) reasoning_ms: Option<u64>,
#[serde(default)]
pub(crate) metadata: Option<serde_json::Value>,
}
#[derive(Debug, Default, Deserialize)]
pub(crate) struct CreateRequest {
#[serde(default)]
title: Option<String>,
#[serde(default)]
model: Option<String>,
#[serde(default)]
head_id: Option<String>,
#[serde(default)]
messages: Vec<NewMessage>,
}
#[derive(Debug, Default, Deserialize)]
pub(crate) struct UpdateRequest {
#[serde(default)]
title: Option<String>,
#[serde(default)]
model: Option<String>,
#[serde(default)]
head_id: Option<String>,
#[serde(default)]
append: Vec<NewMessage>,
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum StoreError {
NotFound,
Invalid { code: &'static str, message: String },
TooLarge { code: &'static str, message: String },
Full(String),
Io(String),
}
impl StoreError {
fn invalid(code: &'static str, message: impl Into<String>) -> Self {
Self::Invalid {
code,
message: message.into(),
}
}
fn too_large(code: &'static str, message: impl Into<String>) -> Self {
Self::TooLarge {
code,
message: message.into(),
}
}
fn into_api_error(self) -> ApiError {
let (status, code, message) = match self {
Self::NotFound => (
StatusCode::NOT_FOUND,
"conversation_not_found",
"no conversation with that id is stored on this server".to_string(),
),
Self::Invalid { code, message } => (StatusCode::BAD_REQUEST, code, message),
Self::TooLarge { code, message } => (StatusCode::PAYLOAD_TOO_LARGE, code, message),
Self::Full(message) => (StatusCode::INSUFFICIENT_STORAGE, "store_full", message),
Self::Io(message) => (
StatusCode::SERVICE_UNAVAILABLE,
"conversation_store_unwritable",
message,
),
};
(
status,
Json(json!({
"error": {
"message": message,
"type": "invalid_request_error",
"code": code,
}
})),
)
}
}
type StoreResult<T> = Result<T, StoreError>;
const ID_PREFIX: &str = "conv_";
static COUNTER: AtomicU64 = AtomicU64::new(0);
fn process_stamp() -> u64 {
use std::sync::OnceLock;
static STAMP: OnceLock<u64> = OnceLock::new();
*STAMP.get_or_init(|| {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0)
})
}
fn next_conversation_id() -> String {
let n = COUNTER.fetch_add(1, Ordering::Relaxed);
format!(
"{ID_PREFIX}{:012x}{:06x}",
process_stamp() & 0xffff_ffff_ffff,
n
)
}
pub(crate) fn is_conversation_id(id: &str) -> bool {
let Some(rest) = id.strip_prefix(ID_PREFIX) else {
return false;
};
!rest.is_empty() && rest.len() <= 32 && rest.bytes().all(|b| b.is_ascii_hexdigit())
}
pub(crate) struct ConversationStore {
dir: PathBuf,
inner: Mutex<BTreeMap<String, Conversation>>,
}
impl ConversationStore {
pub(crate) fn from_env() -> Self {
Self::open(default_dir())
}
pub(crate) fn open(dir: PathBuf) -> Self {
let inner = Mutex::new(load_dir(&dir));
Self { dir, inner }
}
fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Conversation>> {
self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
pub(crate) fn list(&self) -> Vec<Summary> {
let map = self.lock();
let mut out: Vec<Summary> = map.values().map(Conversation::summary).collect();
out.sort_by(|a, b| {
b.updated_at
.cmp(&a.updated_at)
.then_with(|| b.id.cmp(&a.id))
});
out
}
pub(crate) fn get(&self, id: &str) -> StoreResult<Conversation> {
self.lock().get(id).cloned().ok_or(StoreError::NotFound)
}
pub(crate) fn create(&self, request: CreateRequest) -> StoreResult<Conversation> {
let mut map = self.lock();
if map.len() >= MAX_CONVERSATIONS {
return Err(StoreError::Full(format!(
"this server stores at most {MAX_CONVERSATIONS} conversations and is holding \
{}; delete one before creating another (nothing is evicted automatically)",
map.len()
)));
}
let now = crate::unix_now();
let mut conversation = Conversation {
id: next_conversation_id(),
title: validated_title(request.title)?,
model: request.model,
created_at: now,
updated_at: now,
head_id: None,
messages: Vec::new(),
};
append_messages(&mut conversation, request.messages, now)?;
set_head(&mut conversation, request.head_id)?;
derive_title(&mut conversation);
check_size(&conversation)?;
self.persist(&conversation)?;
map.insert(conversation.id.clone(), conversation.clone());
Ok(conversation)
}
pub(crate) fn update(&self, id: &str, request: UpdateRequest) -> StoreResult<Conversation> {
let mut map = self.lock();
let current = map.get(id).ok_or(StoreError::NotFound)?;
let mut next = current.clone();
let mut changed = false;
if let Some(title) = validated_title(request.title)? {
if next.title.as_deref() != Some(title.as_str()) {
next.title = Some(title);
changed = true;
}
}
if let Some(model) = request.model {
if next.model.as_deref() != Some(model.as_str()) {
next.model = Some(model);
changed = true;
}
}
if !request.append.is_empty() {
append_messages(&mut next, request.append, crate::unix_now())?;
changed = true;
}
if request.head_id.is_some() {
let before = next.head_id.clone();
set_head(&mut next, request.head_id)?;
changed |= before != next.head_id;
}
if derive_title(&mut next) {
changed = true;
}
if !changed {
return Ok(next);
}
next.updated_at = crate::unix_now();
check_size(&next)?;
self.persist(&next)?;
map.insert(next.id.clone(), next.clone());
Ok(next)
}
pub(crate) fn delete(&self, id: &str) -> StoreResult<()> {
let mut map = self.lock();
let conversation = map.get(id).ok_or(StoreError::NotFound)?;
let path = self.file_for(&conversation.id);
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(StoreError::Io(format!(
"could not delete {}: {e}",
path.display()
)))
}
}
map.remove(id);
Ok(())
}
fn file_for(&self, id: &str) -> PathBuf {
self.dir.join(format!("{id}.json"))
}
fn persist(&self, conversation: &Conversation) -> StoreResult<()> {
let io = |what: &str, e: std::io::Error| {
StoreError::Io(format!(
"conversation store at {} is not writable ({what}: {e})",
self.dir.display()
))
};
std::fs::create_dir_all(&self.dir).map_err(|e| io("create directory", e))?;
let bytes = serde_json::to_vec_pretty(conversation)
.map_err(|e| StoreError::Io(format!("could not serialize conversation: {e}")))?;
let final_path = self.file_for(&conversation.id);
let tmp_path = self.dir.join(format!("{}.json.tmp", conversation.id));
{
let mut file = std::fs::File::create(&tmp_path).map_err(|e| io("create", e))?;
file.write_all(&bytes).map_err(|e| io("write", e))?;
file.sync_all().map_err(|e| io("sync", e))?;
}
std::fs::rename(&tmp_path, &final_path).map_err(|e| {
std::fs::remove_file(&tmp_path).ok();
io("rename", e)
})
}
}
fn default_dir() -> PathBuf {
std::env::var("FERROX_CONVERSATIONS_DIR")
.map(PathBuf::from)
.unwrap_or_else(|_| PathBuf::from("./ferrox-conversations"))
}
fn load_dir(dir: &PathBuf) -> BTreeMap<String, Conversation> {
let mut map = BTreeMap::new();
let Ok(entries) = std::fs::read_dir(dir) else {
return map;
};
for entry in entries.flatten() {
let path = entry.path();
let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
continue;
};
if path.extension().and_then(|e| e.to_str()) != Some("json") || !is_conversation_id(stem) {
continue;
}
let text = match std::fs::read_to_string(&path) {
Ok(text) => text,
Err(e) => {
tracing::warn!("conversation {} could not be read: {e}", path.display());
continue;
}
};
match serde_json::from_str::<Conversation>(&text) {
Ok(conversation) if conversation.id == stem => {
map.insert(conversation.id.clone(), conversation);
}
Ok(conversation) => tracing::warn!(
"conversation file {} holds id {:?}; skipped rather than guessing which is right",
path.display(),
conversation.id
),
Err(e) => tracing::warn!(
"conversation file {} is not readable as a conversation ({e}); left in place, \
not loaded",
path.display()
),
}
if map.len() >= MAX_CONVERSATIONS {
tracing::warn!(
"conversation store at {} holds more than {MAX_CONVERSATIONS} files; the rest \
are on disk but not loaded",
dir.display()
);
break;
}
}
map
}
fn validated_title(title: Option<String>) -> StoreResult<Option<String>> {
let Some(title) = title else { return Ok(None) };
if title.chars().count() > MAX_TITLE_CHARS {
return Err(StoreError::too_large(
"title_too_long",
format!("a title is at most {MAX_TITLE_CHARS} characters"),
));
}
Ok(Some(title))
}
fn append_messages(
conversation: &mut Conversation,
messages: Vec<NewMessage>,
now: u64,
) -> StoreResult<()> {
if messages.len() > MAX_APPEND {
return Err(StoreError::too_large(
"too_many_messages",
format!("at most {MAX_APPEND} messages may be appended in one request"),
));
}
if conversation.messages.len() + messages.len() > MAX_MESSAGES {
return Err(StoreError::too_large(
"conversation_full",
format!(
"a conversation holds at most {MAX_MESSAGES} messages (branches included); \
start a new conversation rather than losing the old one"
),
));
}
let existing: HashSet<String> = conversation.messages.iter().map(|m| m.id.clone()).collect();
let mut staged: Vec<MessageNode> = Vec::with_capacity(messages.len());
let mut staged_ids: HashSet<String> = HashSet::new();
for message in messages {
if message.id.is_empty() || message.id.len() > MAX_MESSAGE_ID_LEN {
return Err(StoreError::invalid(
"invalid_message_id",
format!("a message id must be 1..={MAX_MESSAGE_ID_LEN} bytes"),
));
}
if existing.contains(&message.id) || staged_ids.contains(&message.id) {
return Err(StoreError::invalid(
"duplicate_message_id",
format!(
"message {:?} is already in this conversation; appending is not an update",
message.id
),
));
}
if !ROLES.contains(&message.role.as_str()) {
return Err(StoreError::invalid(
"unsupported_role",
format!(
"role {:?} is not one of {}; a message this server cannot replay to \
/v1/chat/completions is refused rather than stored",
message.role,
ROLES.join(", ")
),
));
}
if message.content.len() > MAX_CONTENT_BYTES {
return Err(StoreError::too_large(
"message_too_large",
format!("a message holds at most {MAX_CONTENT_BYTES} bytes of content"),
));
}
let reasoning_content = message.reasoning_content.filter(|r| !r.is_empty());
let reasoning_ms = message.reasoning_ms.filter(|_| reasoning_content.is_some());
if reasoning_content
.as_ref()
.is_some_and(|r| r.len() > MAX_CONTENT_BYTES)
{
return Err(StoreError::too_large(
"message_too_large",
format!("a message holds at most {MAX_CONTENT_BYTES} bytes of reasoning_content"),
));
}
if let Some(parent) = message.parent_id.as_deref() {
if !existing.contains(parent) && !staged_ids.contains(parent) {
return Err(StoreError::invalid(
"unknown_parent",
format!(
"message {:?} names parent {parent:?}, which is not in this \
conversation; a transcript that cannot be walked back to a root \
cannot be replayed",
message.id
),
));
}
}
staged_ids.insert(message.id.clone());
staged.push(MessageNode {
id: message.id,
parent_id: message.parent_id,
role: message.role,
content: message.content,
reasoning_content,
reasoning_ms,
created_at: now,
metadata: message.metadata,
});
}
conversation.messages.extend(staged);
Ok(())
}
fn set_head(conversation: &mut Conversation, head_id: Option<String>) -> StoreResult<()> {
let Some(head) = head_id else { return Ok(()) };
if head.is_empty() {
conversation.head_id = None;
return Ok(());
}
if !conversation.has(&head) {
return Err(StoreError::invalid(
"unknown_head",
format!("head_id {head:?} names no message in this conversation"),
));
}
conversation.head_id = Some(head);
Ok(())
}
fn derive_title(conversation: &mut Conversation) -> bool {
if conversation.title.is_some() {
return false;
}
let Some(first) = conversation
.messages
.iter()
.find(|m| m.role == "user" && !m.content.trim().is_empty())
else {
return false;
};
let line: String = first
.content
.split_whitespace()
.collect::<Vec<_>>()
.join(" ");
conversation.title = Some(truncate_chars(&line, DERIVED_TITLE_CHARS));
true
}
fn truncate_chars(text: &str, max: usize) -> String {
if text.chars().count() <= max {
return text.to_string();
}
let mut out: String = text.chars().take(max).collect();
out.push('…');
out
}
fn check_size(conversation: &Conversation) -> StoreResult<()> {
let size = serde_json::to_vec(conversation)
.map(|b| b.len())
.unwrap_or(0);
if size > MAX_CONVERSATION_BYTES {
return Err(StoreError::too_large(
"conversation_too_large",
format!(
"this conversation would be {size} bytes and the ceiling is \
{MAX_CONVERSATION_BYTES}; nothing was stored and nothing was dropped"
),
));
}
Ok(())
}
type Store = Extension<Arc<ConversationStore>>;
async fn list(Extension(store): Store) -> Json<ListBody> {
Json(ListBody {
object: "list",
data: store.list(),
})
}
async fn create(
Extension(store): Store,
Json(request): Json<CreateRequest>,
) -> Result<(StatusCode, Json<ConversationBody>), ApiError> {
let conversation = store.create(request).map_err(StoreError::into_api_error)?;
Ok((StatusCode::CREATED, Json(conversation.into())))
}
async fn fetch(
Extension(store): Store,
AxumPath(id): AxumPath<String>,
) -> Result<Json<ConversationBody>, ApiError> {
let conversation = store.get(&id).map_err(StoreError::into_api_error)?;
Ok(Json(conversation.into()))
}
async fn update(
Extension(store): Store,
AxumPath(id): AxumPath<String>,
Json(request): Json<UpdateRequest>,
) -> Result<Json<ConversationBody>, ApiError> {
let conversation = store
.update(&id, request)
.map_err(StoreError::into_api_error)?;
Ok(Json(conversation.into()))
}
async fn delete(
Extension(store): Store,
AxumPath(id): AxumPath<String>,
) -> Result<Json<DeletedBody>, ApiError> {
store.delete(&id).map_err(StoreError::into_api_error)?;
Ok(Json(DeletedBody {
object: "conversation.deleted",
id,
deleted: true,
}))
}
pub(crate) fn router<S>() -> Router<S>
where
S: Clone + Send + Sync + 'static,
{
router_with(Arc::new(ConversationStore::from_env()))
}
pub(crate) fn router_with<S>(store: Arc<ConversationStore>) -> Router<S>
where
S: Clone + Send + Sync + 'static,
{
Router::new()
.route(CONVERSATIONS, get(list).post(create))
.route(&crate::axum_path(CONVERSATION), get(fetch).post(update))
.route(&crate::axum_path(CONVERSATION_DELETE), post(delete))
.layer(Extension(store))
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use axum::http::Request;
use http_body_util::BodyExt;
use serde_json::Value;
use tower::ServiceExt;
struct TempDir(PathBuf);
impl TempDir {
fn new(tag: &str) -> Self {
let dir = std::env::temp_dir().join(format!(
"ferrox_conversations_{tag}_{}_{:?}",
std::process::id(),
std::thread::current().id()
));
std::fs::remove_dir_all(&dir).ok();
std::fs::remove_file(&dir).ok();
TempDir(dir)
}
}
impl Drop for TempDir {
fn drop(&mut self) {
std::fs::remove_dir_all(&self.0).ok();
std::fs::remove_file(&self.0).ok();
}
}
fn store(dir: &TempDir) -> Arc<ConversationStore> {
Arc::new(ConversationStore::open(dir.0.clone()))
}
fn conversation_path(id: &str) -> String {
CONVERSATION.replace("{conversation_id}", id)
}
fn conversation_delete_path(id: &str) -> String {
CONVERSATION_DELETE.replace("{conversation_id}", id)
}
fn msg(id: &str, parent: Option<&str>, role: &str, content: &str) -> NewMessage {
NewMessage {
id: id.to_string(),
parent_id: parent.map(str::to_string),
role: role.to_string(),
content: content.to_string(),
reasoning_content: None,
reasoning_ms: None,
metadata: None,
}
}
fn created(messages: Vec<NewMessage>) -> CreateRequest {
CreateRequest {
messages,
..Default::default()
}
}
#[test]
fn a_created_conversation_is_readable_and_has_a_server_minted_id() {
let dir = TempDir::new("create");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "hello")]))
.unwrap();
assert!(is_conversation_id(&conversation.id), "{}", conversation.id);
assert_eq!(store.get(&conversation.id).unwrap(), conversation);
assert_eq!(conversation.messages[0].role, "user");
}
#[test]
fn the_store_survives_a_restart() {
let dir = TempDir::new("restart");
let id = {
let store = store(&dir);
store
.create(created(vec![msg("m1", None, "user", "remember me")]))
.unwrap()
.id
};
let reopened = store(&dir);
let conversation = reopened.get(&id).unwrap();
assert_eq!(conversation.messages[0].content, "remember me");
assert_eq!(reopened.list().len(), 1);
}
#[test]
fn a_branch_keeps_both_answers() {
let dir = TempDir::new("branch");
let store = store(&dir);
let conversation = store
.create(created(vec![
msg("u1", None, "user", "hi"),
msg("a1", Some("u1"), "assistant", "first answer"),
]))
.unwrap();
let updated = store
.update(
&conversation.id,
UpdateRequest {
append: vec![msg("a2", Some("u1"), "assistant", "second answer")],
head_id: Some("a2".to_string()),
..Default::default()
},
)
.unwrap();
assert_eq!(updated.messages.len(), 3);
assert_eq!(updated.head_id.as_deref(), Some("a2"));
assert!(
updated.messages.iter().any(|m| m.content == "first answer"),
"regenerating must branch, not overwrite"
);
}
#[test]
fn a_dangling_parent_is_refused_rather_than_stored() {
let dir = TempDir::new("dangling");
let store = store(&dir);
let err = store
.create(created(vec![msg("a1", Some("nobody"), "assistant", "x")]))
.unwrap_err();
assert!(matches!(
err,
StoreError::Invalid {
code: "unknown_parent",
..
}
));
}
#[test]
fn a_batch_can_carry_a_parent_and_its_child() {
let dir = TempDir::new("batch");
let store = store(&dir);
let conversation = store
.create(created(vec![
msg("u1", None, "user", "hi"),
msg("a1", Some("u1"), "assistant", "hello"),
]))
.unwrap();
assert_eq!(conversation.messages.len(), 2);
}
#[test]
fn a_bad_message_leaves_the_whole_batch_unstored() {
let dir = TempDir::new("atomic");
let store = store(&dir);
let conversation = store.create(CreateRequest::default()).unwrap();
let err = store
.update(
&conversation.id,
UpdateRequest {
append: vec![
msg("u1", None, "user", "kept?"),
msg("a1", Some("ghost"), "assistant", "no"),
],
..Default::default()
},
)
.unwrap_err();
assert!(matches!(err, StoreError::Invalid { .. }));
assert!(
store.get(&conversation.id).unwrap().messages.is_empty(),
"half a batch would make the client diff to find out what landed"
);
}
#[test]
fn a_duplicate_message_id_is_refused_because_appending_is_not_updating() {
let dir = TempDir::new("dupe");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "one")]))
.unwrap();
let err = store
.update(
&conversation.id,
UpdateRequest {
append: vec![msg("m1", None, "user", "two")],
..Default::default()
},
)
.unwrap_err();
assert!(matches!(
err,
StoreError::Invalid {
code: "duplicate_message_id",
..
}
));
assert_eq!(
store.get(&conversation.id).unwrap().messages[0].content,
"one"
);
}
#[test]
fn an_unknown_head_is_refused() {
let dir = TempDir::new("head");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "one")]))
.unwrap();
assert!(store
.update(
&conversation.id,
UpdateRequest {
head_id: Some("elsewhere".into()),
..Default::default()
},
)
.is_err());
}
#[test]
fn a_role_this_server_cannot_replay_is_refused() {
let dir = TempDir::new("role");
let store = store(&dir);
let err = store
.create(created(vec![msg("m1", None, "tool", "{}")]))
.unwrap_err();
assert!(matches!(
err,
StoreError::Invalid {
code: "unsupported_role",
..
}
));
}
#[test]
fn the_server_stamps_the_time_and_the_client_cannot() {
let dir = TempDir::new("time");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "hi")]))
.unwrap();
assert!(conversation.messages[0].created_at > 1_700_000_000);
}
#[test]
fn an_untitled_conversation_takes_its_first_user_line() {
let dir = TempDir::new("title");
let store = store(&dir);
let conversation = store
.create(created(vec![msg(
"m1",
None,
"user",
" what is a \n gguf file? ",
)]))
.unwrap();
assert_eq!(conversation.title.as_deref(), Some("what is a gguf file?"));
}
#[test]
fn a_long_title_is_cut_on_a_character_boundary() {
let long = "é".repeat(DERIVED_TITLE_CHARS + 10);
let cut = truncate_chars(&long, DERIVED_TITLE_CHARS);
assert_eq!(cut.chars().count(), DERIVED_TITLE_CHARS + 1);
assert!(cut.ends_with('…'));
}
#[test]
fn an_explicit_title_wins_over_the_derived_one() {
let dir = TempDir::new("explicit");
let store = store(&dir);
let conversation = store
.create(CreateRequest {
title: Some("Kernels".into()),
messages: vec![msg("m1", None, "user", "unrelated question")],
..Default::default()
})
.unwrap();
assert_eq!(conversation.title.as_deref(), Some("Kernels"));
}
#[test]
fn an_empty_update_does_not_rewrite_the_conversation() {
let dir = TempDir::new("noop");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "hi")]))
.unwrap();
let again = store
.update(&conversation.id, UpdateRequest::default())
.unwrap();
assert_eq!(again.updated_at, conversation.updated_at);
}
#[test]
fn delete_removes_the_file_as_well_as_the_entry() {
let dir = TempDir::new("delete");
let store = store(&dir);
let conversation = store
.create(created(vec![msg("m1", None, "user", "hi")]))
.unwrap();
let path = dir.0.join(format!("{}.json", conversation.id));
assert!(path.exists());
store.delete(&conversation.id).unwrap();
assert!(!path.exists());
assert!(matches!(
store.get(&conversation.id),
Err(StoreError::NotFound)
));
assert!(store.list().is_empty());
}
#[test]
fn nothing_is_evicted_when_the_store_is_full() {
let dir = TempDir::new("full");
let store = store(&dir);
for _ in 0..MAX_CONVERSATIONS {
store.create(CreateRequest::default()).unwrap();
}
let err = store.create(CreateRequest::default()).unwrap_err();
assert!(matches!(err, StoreError::Full(_)));
assert_eq!(store.list().len(), MAX_CONVERSATIONS);
}
#[test]
fn an_oversized_message_is_refused() {
let dir = TempDir::new("big");
let store = store(&dir);
let err = store
.create(created(vec![msg(
"m1",
None,
"user",
&"x".repeat(MAX_CONTENT_BYTES + 1),
)]))
.unwrap_err();
assert!(matches!(
err,
StoreError::TooLarge {
code: "message_too_large",
..
}
));
}
#[test]
fn a_failed_write_is_not_reported_as_a_stored_conversation() {
let dir = TempDir::new("unwritable");
std::fs::write(&dir.0, b"not a directory").unwrap();
let store = store(&dir);
let err = store
.create(created(vec![msg("m1", None, "user", "hi")]))
.unwrap_err();
assert!(matches!(err, StoreError::Io(_)), "{err:?}");
assert!(
store.list().is_empty(),
"memory must not hold a conversation that never reached disk"
);
}
#[test]
fn a_chain_of_thought_survives_a_restart_beside_its_answer() {
let dir = TempDir::new("reasoning");
let id = {
let store = store(&dir);
let mut thought = msg("a1", Some("u1"), "assistant", "");
thought.reasoning_content = Some("Let me think about".to_string());
let mut blank = msg("a2", Some("a1"), "assistant", "answer");
blank.reasoning_content = Some(String::new());
store
.create(created(vec![
msg("u1", None, "user", "why"),
thought,
blank,
]))
.unwrap()
.id
};
let conversation = store(&dir).get(&id).unwrap();
assert_eq!(
conversation.messages[1].reasoning_content.as_deref(),
Some("Let me think about")
);
assert_eq!(conversation.messages[1].content, "");
assert_eq!(
conversation.messages[2].reasoning_content, None,
"an empty thought is no thought"
);
let on_disk = std::fs::read_to_string(dir.0.join(format!("{id}.json"))).unwrap();
assert_eq!(
on_disk.matches("reasoning_content").count(),
1,
"the key is written only where there is a thought: {on_disk}"
);
}
#[test]
fn a_thoughts_duration_survives_a_restart_and_never_without_a_thought() {
let dir = TempDir::new("reasoning-ms");
let id = {
let store = store(&dir);
let mut timed = msg("a1", Some("u1"), "assistant", "answer");
timed.reasoning_content = Some("Let me think".to_string());
timed.reasoning_ms = Some(15_250);
let mut untimed = msg("a2", Some("a1"), "assistant", "again");
untimed.reasoning_content = Some("Hmm".to_string());
let mut thoughtless = msg("a3", Some("a2"), "assistant", "plain");
thoughtless.reasoning_ms = Some(3_000);
let mut blank = msg("a4", Some("a3"), "assistant", "blank");
blank.reasoning_content = Some(String::new());
blank.reasoning_ms = Some(4_000);
store
.create(created(vec![
msg("u1", None, "user", "why"),
timed,
untimed,
thoughtless,
blank,
]))
.unwrap()
.id
};
let conversation = store(&dir).get(&id).unwrap();
assert_eq!(conversation.messages[1].reasoning_ms, Some(15_250));
assert_eq!(conversation.messages[2].reasoning_ms, None);
assert_eq!(
conversation.messages[3].reasoning_ms, None,
"a duration with no thought beside it is a number about nothing"
);
assert_eq!(
conversation.messages[4].reasoning_ms, None,
"an empty thought is no thought, so it has no duration either"
);
let on_disk = std::fs::read_to_string(dir.0.join(format!("{id}.json"))).unwrap();
assert_eq!(
on_disk.matches("reasoning_ms").count(),
1,
"the key is written only where there is a duration: {on_disk}"
);
}
#[test]
fn a_record_from_before_reasoning_was_stored_loads_unchanged() {
let dir = TempDir::new("pre-reasoning");
std::fs::create_dir_all(&dir.0).unwrap();
std::fs::write(
dir.0.join("conv_0000000000000000.json"),
br#"{"id":"conv_0000000000000000","created_at":1,"updated_at":1,"head_id":"a1",
"messages":[{"id":"u1","parent_id":null,"role":"user","content":"hi","created_at":1},
{"id":"a1","parent_id":"u1","role":"assistant","content":"hello","created_at":1}]}"#,
)
.unwrap();
let conversation = store(&dir).get("conv_0000000000000000").unwrap();
assert_eq!(conversation.messages.len(), 2);
assert_eq!(conversation.messages[1].content, "hello");
assert_eq!(conversation.messages[1].reasoning_content, None);
assert_eq!(conversation.messages[1].reasoning_ms, None);
}
#[test]
fn an_oversized_thought_is_refused_like_an_oversized_answer() {
let dir = TempDir::new("big-thought");
let store = store(&dir);
let mut thought = msg("m1", None, "assistant", "short");
thought.reasoning_content = Some("x".repeat(MAX_CONTENT_BYTES + 1));
let err = store.create(created(vec![thought])).unwrap_err();
assert!(matches!(
err,
StoreError::TooLarge {
code: "message_too_large",
..
}
));
}
#[test]
fn an_unreadable_file_is_skipped_and_left_alone() {
let dir = TempDir::new("corrupt");
std::fs::create_dir_all(&dir.0).unwrap();
let path = dir.0.join("conv_deadbeef.json");
std::fs::write(&path, b"{ not json").unwrap();
let store = store(&dir);
assert!(store.list().is_empty());
assert!(
path.exists(),
"the bytes stay on disk for a human; nothing here deletes a transcript it could \
not parse"
);
}
#[test]
fn a_file_whose_inner_id_disagrees_with_its_name_is_skipped() {
let dir = TempDir::new("mismatch");
std::fs::create_dir_all(&dir.0).unwrap();
std::fs::write(
dir.0.join("conv_00000000.json"),
serde_json::to_vec(&Conversation {
id: "conv_11111111".into(),
title: None,
model: None,
created_at: 1,
updated_at: 1,
head_id: None,
messages: vec![],
})
.unwrap(),
)
.unwrap();
assert!(store(&dir).list().is_empty());
}
#[test]
fn only_server_minted_ids_are_recognised() {
assert!(is_conversation_id(&next_conversation_id()));
assert!(!is_conversation_id("conv_../../etc/passwd"));
assert!(!is_conversation_id("../conv_dead"));
assert!(!is_conversation_id("conv_"));
assert!(!is_conversation_id("chatcmpl-1"));
}
#[test]
fn the_list_is_newest_first() {
let dir = TempDir::new("order");
let store = store(&dir);
let a = store.create(CreateRequest::default()).unwrap();
let b = store.create(CreateRequest::default()).unwrap();
let ids: Vec<String> = store.list().into_iter().map(|s| s.id).collect();
assert_eq!(ids, vec![b.id, a.id]);
}
#[test]
fn a_summary_is_not_mistakable_for_a_conversation() {
let dir = TempDir::new("summary");
let store = store(&dir);
store
.create(created(vec![msg("m1", None, "user", "hi")]))
.unwrap();
let summary = &store.list()[0];
assert_eq!(summary.object, "conversation.summary");
assert_eq!(summary.message_count, 1);
let json = serde_json::to_value(summary).unwrap();
assert!(
json.get("messages").is_none(),
"a summary carries no transcript, so it must not be tagged as one"
);
}
#[test]
fn client_metadata_comes_back_byte_identical() {
let dir = TempDir::new("metadata");
let store = store(&dir);
let custom = json!({ "custom": { "stats": { "line": "TTFT 40 ms" } } });
let conversation = store
.create(created(vec![NewMessage {
metadata: Some(custom.clone()),
..msg("m1", None, "assistant", "hi")
}]))
.unwrap();
assert_eq!(conversation.messages[0].metadata, Some(custom.clone()));
let reopened = ConversationStore::open(dir.0.clone());
assert_eq!(
reopened.get(&conversation.id).unwrap().messages[0].metadata,
Some(custom)
);
}
#[test]
fn no_template_reaches_the_router_with_its_braces() {
for template in [CONVERSATION, CONVERSATION_DELETE] {
assert!(template.contains('{'), "{template} has no placeholder");
let mounted = crate::axum_path(template);
assert!(
!mounted.contains('{') && !mounted.contains('}'),
"{template} would be mounted as {mounted}, whose braces axum reads as a \
literal segment"
);
assert!(mounted.contains(':'), "{template} lost its placeholder");
}
assert_eq!(
crate::axum_path(CONVERSATION),
"/v1/conversations/:conversation_id"
);
assert_eq!(
conversation_path("conv_1"),
crate::axum_path(CONVERSATION).replace(":conversation_id", "conv_1")
);
assert_eq!(
conversation_delete_path("conv_1"),
"/v1/conversations/conv_1/delete"
);
}
#[test]
fn the_item_routes_sit_under_the_collection() {
assert!(CONVERSATION.starts_with(CONVERSATIONS));
assert!(CONVERSATION_DELETE.starts_with(CONVERSATION));
assert!(CONVERSATIONS.starts_with("/v1/"));
}
async fn call(app: &Router, request: Request<Body>) -> (StatusCode, Value) {
let response = app.clone().oneshot(request).await.unwrap();
let status = response.status();
let bytes = response.into_body().collect().await.unwrap().to_bytes();
let body = if bytes.is_empty() {
Value::Null
} else {
serde_json::from_slice(&bytes).unwrap_or(Value::Null)
};
(status, body)
}
async fn get_path(app: &Router, path: &str) -> (StatusCode, Value) {
call(
app,
Request::builder().uri(path).body(Body::empty()).unwrap(),
)
.await
}
async fn post_path(app: &Router, path: &str, body: Value) -> (StatusCode, Value) {
call(
app,
Request::builder()
.method("POST")
.uri(path)
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.unwrap(),
)
.await
}
fn test_app(dir: &TempDir) -> Router {
router_with::<()>(store(dir)).with_state(())
}
#[tokio::test]
async fn the_http_surface_round_trips_a_conversation() {
let dir = TempDir::new("http");
let app = test_app(&dir);
let (status, created) = post_path(
&app,
CONVERSATIONS,
json!({ "messages": [{ "id": "u1", "role": "user", "content": "hi" }] }),
)
.await;
assert_eq!(status, StatusCode::CREATED);
assert_eq!(created["object"], "conversation");
let id = created["id"].as_str().unwrap().to_string();
let (status, listed) = get_path(&app, CONVERSATIONS).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(listed["object"], "list");
assert_eq!(listed["data"][0]["id"], id.as_str());
let (status, fetched) = get_path(&app, &conversation_path(&id)).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(fetched["messages"][0]["content"], "hi");
let (status, updated) = post_path(
&app,
&conversation_path(&id),
json!({
"append": [
{ "id": "a1", "parent_id": "u1", "role": "assistant", "content": "hello" }
],
"head_id": "a1",
}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(updated["head_id"], "a1");
assert_eq!(updated["messages"].as_array().unwrap().len(), 2);
let (status, deleted) = post_path(&app, &conversation_delete_path(&id), json!({})).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(deleted["deleted"], true);
let (status, _) = get_path(&app, &conversation_path(&id)).await;
assert_eq!(status, StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn an_unknown_id_gets_the_handler_not_a_bare_404() {
let dir = TempDir::new("bare404");
let app = test_app(&dir);
let (status, body) = get_path(&app, &conversation_path("conv_nothing")).await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(body["error"]["code"], "conversation_not_found");
let (status, body) =
post_path(&app, &conversation_delete_path("conv_nothing"), json!({})).await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(body["error"]["code"], "conversation_not_found");
}
#[tokio::test]
async fn a_traversal_shaped_id_is_just_an_unknown_id() {
let dir = TempDir::new("traversal");
let app = test_app(&dir);
let (status, body) = get_path(&app, "/v1/conversations/..%2F..%2Fetc%2Fpasswd").await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(body["error"]["code"], "conversation_not_found");
}
#[tokio::test]
async fn a_refused_append_says_which_rule_it_broke() {
let dir = TempDir::new("refusal");
let app = test_app(&dir);
let (_, created) = post_path(&app, CONVERSATIONS, json!({})).await;
let id = created["id"].as_str().unwrap().to_string();
let (status, body) = post_path(
&app,
&conversation_path(&id),
json!({ "append": [{ "id": "a1", "parent_id": "ghost", "role": "assistant" }] }),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["error"]["code"], "unknown_parent");
assert!(body["error"]["message"].as_str().unwrap().contains("ghost"));
}
}