diff --git a/Cargo.lock b/Cargo.lock index 9455f5c..1b3a415 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2051,7 +2051,7 @@ dependencies = [ [[package]] name = "image-api" -version = "1.4.0" +version = "1.5.0" dependencies = [ "actix", "actix-cors", diff --git a/Cargo.toml b/Cargo.toml index 860e6ae..270e9e4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "image-api" -version = "1.4.0" +version = "1.5.0" authors = ["Cameron Cordes "] edition = "2024" diff --git a/migrations/2026-08-25-000000_persona_chat_multi/down.sql b/migrations/2026-08-25-000000_persona_chat_multi/down.sql new file mode 100644 index 0000000..bd0fcd7 --- /dev/null +++ b/migrations/2026-08-25-000000_persona_chat_multi/down.sql @@ -0,0 +1,32 @@ +-- Collapse back to one conversation per (user, persona). Where a persona has +-- several, the most recently updated one wins and the rest are dropped — +-- the v1 schema has nowhere to put them. + +CREATE TABLE persona_chat_conversations_old ( + user_id INTEGER NOT NULL, + persona_id TEXT NOT NULL, + messages_json TEXT NOT NULL DEFAULT '[]', + turn_count INTEGER NOT NULL DEFAULT 0, + updated_at BIGINT NOT NULL, + PRIMARY KEY (user_id, persona_id) +); + +INSERT INTO persona_chat_conversations_old ( + user_id, persona_id, messages_json, turn_count, updated_at +) +SELECT user_id, persona_id, messages_json, turn_count, updated_at +FROM persona_chat_conversations c +WHERE c.updated_at = ( + SELECT MAX(c2.updated_at) + FROM persona_chat_conversations c2 + WHERE c2.user_id = c.user_id AND c2.persona_id = c.persona_id +) +GROUP BY user_id, persona_id; + +DROP INDEX IF EXISTS idx_persona_chat_updated; +DROP INDEX IF EXISTS idx_persona_chat_persona; +DROP TABLE persona_chat_conversations; +ALTER TABLE persona_chat_conversations_old RENAME TO persona_chat_conversations; + +CREATE INDEX idx_persona_chat_updated + ON persona_chat_conversations (user_id, updated_at DESC); diff --git a/migrations/2026-08-25-000000_persona_chat_multi/up.sql b/migrations/2026-08-25-000000_persona_chat_multi/up.sql new file mode 100644 index 0000000..64fcfd1 --- /dev/null +++ b/migrations/2026-08-25-000000_persona_chat_multi/up.sql @@ -0,0 +1,50 @@ +-- Multiple conversations per persona. +-- +-- v1 keyed a transcript on (user_id, persona_id), so a persona had exactly +-- one rolling conversation and there was no way to start a fresh topic +-- without discarding the old one. The key is now an opaque `conversation_id`, +-- with (user_id, persona_id) demoted to an index. +-- +-- `title` is a short generated summary of the opening exchange, used as the +-- conversation's name in the list. Empty until the first turn completes; the +-- client falls back to the persona name while it is blank. +-- +-- SQLite cannot redefine a primary key in place, so this is the standard +-- create-copy-drop-rename dance. Existing transcripts carry over with a +-- generated id and an empty title. + +CREATE TABLE persona_chat_conversations_new ( + conversation_id TEXT NOT NULL PRIMARY KEY, + user_id INTEGER NOT NULL, + persona_id TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + messages_json TEXT NOT NULL DEFAULT '[]', + turn_count INTEGER NOT NULL DEFAULT 0, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL +); + +INSERT INTO persona_chat_conversations_new ( + conversation_id, user_id, persona_id, title, + messages_json, turn_count, created_at, updated_at +) +SELECT + lower(hex(randomblob(16))), + user_id, + persona_id, + '', + messages_json, + turn_count, + updated_at, + updated_at +FROM persona_chat_conversations; + +DROP INDEX IF EXISTS idx_persona_chat_updated; +DROP TABLE persona_chat_conversations; +ALTER TABLE persona_chat_conversations_new RENAME TO persona_chat_conversations; + +CREATE INDEX idx_persona_chat_updated + ON persona_chat_conversations (user_id, updated_at DESC); + +CREATE INDEX idx_persona_chat_persona + ON persona_chat_conversations (user_id, persona_id); diff --git a/src/ai/handlers.rs b/src/ai/handlers.rs index f4c0012..1a98d35 100644 --- a/src/ai/handlers.rs +++ b/src/ai/handlers.rs @@ -2051,7 +2051,7 @@ pub(crate) async fn cancel_turn_impl( entry.set_terminal_status(crate::ai::turn_registry::TurnStatus::Cancelled); span.set_status(Status::Ok); -HttpResponse::Ok().json(serde_json::json!({ + HttpResponse::Ok().json(serde_json::json!({ "cancelled": true })) } diff --git a/src/ai/insight_chat.rs b/src/ai/insight_chat.rs index 3bcd9fd..bc070f2 100644 --- a/src/ai/insight_chat.rs +++ b/src/ai/insight_chat.rs @@ -30,11 +30,11 @@ pub const DEFAULT_MAX_ITERATIONS: usize = 6; const DEFAULT_NUM_CTX: i32 = 32768; /// Headroom reserved for the model's response, deducted from the context /// budget when deciding whether to truncate the replayed history. -const RESPONSE_HEADROOM_TOKENS: usize = 2048; +pub(crate) const RESPONSE_HEADROOM_TOKENS: usize = 2048; /// Cheap byte-to-token approximation used by the truncation pass. The real /// tokenization is model-specific; this avoids carrying tiktoken just for a /// soft bound. -const BYTES_PER_TOKEN: usize = 4; +pub(crate) const BYTES_PER_TOKEN: usize = 4; /// Flat token cost charged per inlined image in the truncation budget. A /// 1024px-longest-edge JPEG (see `load_image_as_base64`) costs vision models on /// the order of ~1.3K tokens. Crucially, the raw base64 (hundreds of KB of @@ -335,6 +335,13 @@ impl InsightChatService { if truncated { span.set_attribute(KeyValue::new("history_truncated", true)); } + // Everything in `messages` at this point is replayed history that is + // ALREADY in the tree; only what the turn appends past this mark + // becomes new nodes. Captured after the budget pass, not from + // `path.len()`: truncation drains from the middle, so `path.len()` + // over-counts and the slice below would either swallow the new user + // turn or panic on an out-of-range start index. + let history_len = messages.len(); // 7. Append the new user turn. messages.push(ChatMessage::user(req.user_message.clone())); @@ -461,8 +468,7 @@ impl InsightChatService { // relying on store_insight to flip prior rows' is_current=false. // Append new messages (this turn's user + assistant exchanges) as // tree nodes chained from the previous active_leaf_id. - let path_len = path.len(); - let new_messages = messages[path_len..].to_vec(); + let new_messages = messages[history_len..].to_vec(); let mut parent_id = Some(store.active_leaf_id); for msg in &new_messages { let new_id = store.append_node(parent_id, msg.clone()); @@ -942,7 +948,6 @@ impl InsightChatService { let path = store .path_to_leaf(store.active_leaf_id) .ok_or_else(|| anyhow!("active_leaf_id {} not found in tree", store.active_leaf_id))?; - let path_len = path.len(); let mut messages: Vec = path.iter().map(|n| n.message.clone()).collect(); let stored_backend = insight.backend.clone(); @@ -1003,6 +1008,10 @@ impl InsightChatService { if truncated { let _ = entry.push_event(ChatStreamEvent::Truncated).await; } + // See the note in `chat_turn`: the new-node boundary must be read + // after the budget pass, because truncation drains replayed history + // out of the middle of `messages`. + let history_len = messages.len(); messages.push(ChatMessage::user(req.user_message.clone())); @@ -1046,7 +1055,7 @@ impl InsightChatService { } // Append new messages as tree nodes. - let new_messages = messages[path_len..].to_vec(); + let new_messages = messages[history_len..].to_vec(); let mut parent_id = Some(store.active_leaf_id); for msg in &new_messages { let new_id = store.append_node(parent_id, msg.clone()); @@ -2219,7 +2228,7 @@ pub fn env_max_iterations() -> usize { /// Read AGENTIC_CHAT_DEFAULT_NUM_CTX once per call — the assumed context /// window for the truncation budget when the request omits `num_ctx`. Same /// no-static-global rationale as `env_max_iterations` above. -fn env_default_num_ctx() -> i32 { +pub(crate) fn env_default_num_ctx() -> i32 { std::env::var("AGENTIC_CHAT_DEFAULT_NUM_CTX") .ok() .and_then(|s| s.parse::().ok()) @@ -2534,7 +2543,7 @@ pub struct ForkInfo { /// tool-dispatch assistant (empty content + tool_calls) is still attributed to /// the next rendered message. Detecting forks only on rendered nodes would miss /// regenerations where the model replied with a tool call. -fn render_tree_path( +pub(crate) fn render_tree_path( store: &ChatHistoryStore, path: &[&StoredChatNode], ) -> (Vec, usize, Vec, Vec>) { diff --git a/src/ai/llm_client.rs b/src/ai/llm_client.rs index ca4ee58..ff33b54 100644 --- a/src/ai/llm_client.rs +++ b/src/ai/llm_client.rs @@ -407,8 +407,8 @@ impl ChatHistoryStore { .collect() } - /// The parent of the given node, if any. Test-only helper. - #[cfg(test)] + /// The parent of the given node, if any. Used to re-anchor a rewind + /// that discards every rendered message onto the seed node above them. pub fn parent_of(&self, node_id: u64) -> Option<&StoredChatNode> { let node = self.nodes.iter().find(|n| n.id == node_id)?; let parent_id = node.parent_id?; diff --git a/src/ai/mod.rs b/src/ai/mod.rs index d22697a..d39645a 100644 --- a/src/ai/mod.rs +++ b/src/ai/mod.rs @@ -7,13 +7,13 @@ pub mod gpu; pub mod handlers; pub mod insight_chat; pub mod insight_generator; -pub mod persona_chat; pub mod llamacpp; pub mod llm_client; pub mod local_llm; pub mod nl_query; pub mod ollama; pub mod openrouter; +pub mod persona_chat; pub mod pronunciation; pub mod sms_client; pub mod tts; @@ -33,16 +33,18 @@ pub use handlers::{ get_available_models_handler, get_insight_handler, get_insight_history_handler, get_openrouter_models_handler, rate_insight_handler, turn_async_handler, turn_replay_handler, }; -pub use persona_chat::{ - persona_chat_history_handler, persona_chat_reset_handler, persona_chat_turn_handler, - persona_turn_cancel_handler, persona_turn_replay_handler, -}; pub use insight_generator::InsightGenerator; pub use llamacpp::LlamaCppClient; #[allow(unused_imports)] pub use llm_client::{ ChatMessage, LlmClient, ModelCapabilities, Tool, ToolCall, ToolCallFunction, ToolFunction, }; +pub use persona_chat::{ + persona_chat_branches_handler, persona_chat_conversations_handler, + persona_chat_create_conversation_handler, persona_chat_delete_conversation_handler, + persona_chat_history_handler, persona_chat_rewind_handler, persona_chat_switch_branch_handler, + persona_chat_turn_handler, persona_turn_cancel_handler, persona_turn_replay_handler, +}; // LocalLlm is constructed by binaries (reembed_embeddings, importers), not the server #[allow(unused_imports)] pub use local_llm::LocalLlm; diff --git a/src/ai/persona_chat.rs b/src/ai/persona_chat.rs index dee054d..596a809 100644 --- a/src/ai/persona_chat.rs +++ b/src/ai/persona_chat.rs @@ -1,8 +1,9 @@ //! Open chat with persona — file-anchored insight chat's sibling. //! -//! The persona chat is the same agentic loop, but anchored on `(user_id, -//! persona_id)` instead of `(library_id, file_path)`. A single rolling -//! transcript per persona; no amend, no rewind, no branches in v1. +//! The persona chat is the same agentic loop, but anchored on a +//! `conversation_id` instead of `(library_id, file_path)`. A persona can +//! hold any number of separate conversations; each is a branching +//! transcript tree supporting rewind and fork, exactly like the file chat. //! //! The session reuses three building blocks from the file chat: //! 1. `InsightGenerator::build_tool_definitions` for the tool catalog @@ -21,8 +22,8 @@ //! registry. //! - The HTTP handlers under ` /persona_chat/*`. -use actix_web::{delete, get, post, web, HttpRequest, HttpResponse, Responder}; -use anyhow::{Context, Result, anyhow}; +use actix_web::{HttpRequest, HttpResponse, Responder, delete, get, post, web}; +use anyhow::{Context, Result, anyhow, bail}; use chrono::Utc; use opentelemetry::KeyValue; use opentelemetry::trace::{Span, Status, Tracer}; @@ -34,9 +35,12 @@ use uuid::Uuid; use crate::ai::backend::{BackendKind, SamplingOverrides}; use crate::ai::handlers::ReplayQuery; -use crate::ai::insight_chat::{ChatStreamEvent, DEFAULT_MAX_ITERATIONS, env_max_iterations}; +use crate::ai::insight_chat::{ + BYTES_PER_TOKEN, ChatStreamEvent, DEFAULT_MAX_ITERATIONS, ForkInfo, RESPONSE_HEADROOM_TOKENS, + apply_context_budget, env_default_num_ctx, env_max_iterations, render_tree_path, +}; use crate::ai::insight_generator::InsightGenerator; -use crate::ai::llm_client::ChatMessage; +use crate::ai::llm_client::{ChatHistoryStore, ChatMessage, StoredChatNode}; use crate::ai::turn_registry::{TurnEntry, TurnRegistry}; use crate::data::Claims; use crate::database::PersonaDao; @@ -51,6 +55,7 @@ use crate::state::AppState; #[derive(Debug)] pub enum PersonaChatError { UnknownPersona(String), + UnknownConversation(String), EmptyMessage, MessageTooLong, ConcurrentTurn, @@ -61,9 +66,14 @@ impl std::fmt::Display for PersonaChatError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { PersonaChatError::UnknownPersona(p) => write!(f, "persona '{p}' not found"), + PersonaChatError::UnknownConversation(c) => { + write!(f, "conversation '{c}' not found") + } PersonaChatError::EmptyMessage => write!(f, "user_message must not be empty"), PersonaChatError::MessageTooLong => write!(f, "user_message exceeds 8192 chars"), - PersonaChatError::ConcurrentTurn => write!(f, "another turn for this persona is in flight"), + PersonaChatError::ConcurrentTurn => { + write!(f, "another turn for this persona is in flight") + } PersonaChatError::Db(e) => write!(f, "database error: {e}"), } } @@ -94,63 +104,54 @@ pub fn resolve_persona_system_prompt( .map(|p| p.system_prompt.clone()) } -/// Build the initial messages vector for a fresh persona conversation. +/// Build the seed messages for a fresh persona conversation. /// -/// Unlike the file chat (which seeds with a photo context), the open chat -/// seeds with a single system message + an opening assistant turn that -/// explains the surface ("you can ask anything your tools can reach"). -/// This keeps the agent loop's `messages[0]` = system invariant and gives -/// the LLM a stable on-screen greeting to reproduce across persona picks. -pub fn seed_messages_for_persona( - system_prompt: &str, - persona_name: &str, -) -> Vec { - let mut messages = Vec::new(); - messages.push(ChatMessage { -role: "system".to_string(), - content: system_prompt.to_string(), - tool_calls: None, - images: None, - }); - messages.push(ChatMessage { - role: "assistant".to_string(), - content: format!( - "Hi — I'm {} ready to help. Ask anything your tools can reach \ - (memories, files, SMS, calendar, places).", - persona_name - ), - tool_calls: None, - images: None, - }); - messages +/// Just the system prompt. The file chat seeds a synthetic "describe this +/// photo" user turn because its transcript is anchored on an image; an open +/// chat has no anchor, so the user's own first question is message one. +/// +/// v1 also seeded an assistant greeting here. It never reached the screen — +/// the flat renderer stripped it — and the tree renderer has no equivalent +/// stripping pass, so it is gone. `drop_seed_greeting` handles rows that +/// still carry one. +pub fn seed_messages_for_persona(system_prompt: &str) -> Vec { + vec![ChatMessage::system(system_prompt.to_string())] } -/// Round-trip the persisted transcript through `serde_json`. Cheap because -/// the schema is already a flat `Vec` (same shape as -/// `training_messages` on the file chat). -pub fn decode_history(raw: &str) -> Result> { +/// Parse a persisted transcript into the conversation tree. +/// +/// Two on-disk formats are accepted, mirroring the file chat's reader: +/// 1. `ChatHistoryStore` JSON — the tree, which is what every write produces +/// now. +/// 2. A flat `Vec` — rows written before the tree migration, +/// upgraded on read into a single linear branch. +pub fn decode_store(raw: &str) -> Result { + if let Ok(flat) = serde_json::from_str::>(raw) { + return Ok(ChatHistoryStore::from_flat_array(drop_seed_greeting(flat))); + } serde_json::from_str(raw).with_context(|| "failed to deserialize persona chat history") } -pub fn encode_history(messages: &[ChatMessage]) -> Result { - serde_json::to_string(messages) - .with_context(|| "failed to serialize persona chat history") +pub fn encode_store(store: &ChatHistoryStore) -> Result { + serde_json::to_string(store).with_context(|| "failed to serialize persona chat history") } -/// Strip the seed system message + greeting assistant message before -/// persisting, so a fresh conversation never re-greets when reopened -/// (the client renders empty-state instead). -pub fn strip_seed(messages: Vec) -> Vec { - let mut out = messages; - // Drop leading system + opening assistant in order. - while out - .first() - .map(|m| m.role == "system" || (m.role == "assistant" && is_seed_greeting(&m.content))) - .unwrap_or(false) +/// Drop the v1 seed greeting from a pre-migration transcript. +/// +/// The greeting was invisible under the flat renderer, which stripped every +/// leading system/greeting message. The tree renderer skips `system` nodes +/// but renders every assistant node, so without this the upgrade would +/// surface a bubble the user has never seen. Scoped to an assistant message +/// preceded only by system messages, so a genuine reply that happens to open +/// with the same words is left alone. +fn drop_seed_greeting(mut messages: Vec) -> Vec { + if let Some(idx) = messages.iter().position(|m| m.role != "system") + && messages[idx].role == "assistant" + && is_seed_greeting(&messages[idx].content) { - out.remove(0); + messages.remove(idx); } - out + messages } fn is_seed_greeting(content: &str) -> bool { @@ -163,9 +164,10 @@ fn is_seed_greeting(content: &str) -> bool { #[derive(Debug, Deserialize)] pub struct PersonaChatTurnRequest { - /// Active persona id. Must match a row in the user's persona store - /// (built-ins seeded by migration + customs via /personas). - pub persona_id: String, + /// Conversation to append to. The persona is read from the stored + /// conversation, not the request — a transcript's voice is fixed when it + /// is created, so a stale client can't quietly swap it mid-thread. + pub conversation_id: String, /// Free-text user message. Trimmed before validation; empty input is a 400. pub user_message: String, #[serde(default)] @@ -195,15 +197,29 @@ pub struct PersonaChatTurnRequest { #[derive(Debug, Serialize)] pub struct PersonaChatHistoryView { + /// Conversation this transcript belongs to. Empty for the empty + /// envelope returned when the conversation has no messages yet. + pub conversation_id: String, + /// Persona whose voice this conversation uses. + pub persona_id: String, + /// Generated name for the conversation; empty until the first turn has + /// produced enough of an exchange to summarize. + pub title: String, /// Rendered transcript — same message shape the file chat's history /// endpoint ships, so the client's renderer is shared. pub messages: Vec, pub turn_count: u32, pub model_version: String, pub backend: String, - pub active_leaf_id: i64, - pub viewing_branch_id: i64, - pub fork_info: Vec, + /// Leaf the conversation is anchored to. A new turn extends this leaf. + pub active_leaf_id: u64, + /// Leaf actually being rendered — equals `active_leaf_id` unless the + /// client asked for an alternate branch via `branch_id`. + pub viewing_branch_id: u64, + /// Per-message divergence markers, `None` where the path doesn't fork. + /// Empty for a conversation that has never been rewound. + #[serde(skip_serializing_if = "Vec::is_empty")] + pub fork_info: Vec>, } /// One rendered line of the persona transcript. Mirrors the file chat's @@ -218,90 +234,200 @@ pub struct RenderedPersonaMessage { pub tools: Vec, } -/// Flatten the raw persisted transcript into rendered lines for the -/// history endpoint. Mirrors `render_tree_path` for a linear store: -/// `system` lines are dropped, `tool` results are folded into the -/// assistant message that follows them, and an assistant line whose only -/// payload is `tool_calls` is scaffolding (never rendered on its own). -pub fn render_flat_transcript(messages: Vec) -> Vec { - use crate::ai::insight_chat::{truncate_tool_result, ToolInvocation}; +/// Render a branch path into the persona wire shape. +/// +/// A thin wrapper over the file chat's `render_tree_path` so both surfaces +/// fold tool scaffolding, detect divergences and rank branches identically. +/// The one persona-specific adjustment is `is_initial`: the file chat's +/// first rendered message is the synthetic "describe this photo" prompt, +/// which must never be rewound or regenerated, whereas a persona chat's +/// first message is the user's own question and both actions are legitimate +/// on it. +fn render_persona_path( + store: &ChatHistoryStore, + path: &[&StoredChatNode], +) -> ( + Vec, + usize, + Vec, + Vec>, +) { + let (rendered, turn_count, node_ids, fork_info) = render_tree_path(store, path); + let rendered = rendered + .into_iter() + .map(|m| RenderedPersonaMessage { + role: m.role, + content: m.content, + is_initial: false, + tools: m + .tools + .into_iter() + .map(|t| crate::ai::handlers::HistoryToolInvocation { + name: t.name, + arguments: t.arguments, + result: t.result, + result_truncated: t.result_truncated, + }) + .collect(), + }) + .collect(); + (rendered, turn_count, node_ids, fork_info) +} - let mut rendered = Vec::new(); - let mut user_turns_seen = 0usize; - let mut pending_tools: Vec = Vec::new(); - let mut pending_calls: std::collections::VecDeque<(String, serde_json::Value)> = - std::collections::VecDeque::new(); +/// The node a rewind should re-anchor `active_leaf_id` on. +/// +/// `node_ids[i]` is the tree node behind rendered message `i`, so keeping +/// everything before index `i` means anchoring on `node_ids[i - 1]`. Index 0 +/// discards the whole transcript, which anchors on the parent of the first +/// rendered node — the seed system node — so that a resend forks there and +/// the discarded path stays reachable via the chip on the new first bubble. +fn rewind_target_leaf( + store: &ChatHistoryStore, + node_ids: &[u64], + discard_from_rendered_index: usize, +) -> Result { + if discard_from_rendered_index == 0 { + let first = *node_ids + .first() + .ok_or_else(|| anyhow!("discard_from_rendered_index out of range"))?; + return store + .parent_of(first) + .map(|n| n.id) + .ok_or_else(|| anyhow!("cannot rewind past the start of the conversation")); + } + node_ids + .get(discard_from_rendered_index - 1) + .copied() + .ok_or_else(|| anyhow!("discard_from_rendered_index out of range")) +} - for msg in messages { - match msg.role.as_str() { - "system" => continue, - "tool" => { - if let Some((name, arguments)) = pending_calls.pop_front() { - let (result, result_truncated) = truncate_tool_result(&msg.content); - pending_tools.push(ToolInvocation { - name, - arguments, - result, - result_truncated, - }); - } - } - "assistant" => { - let has_tool_calls = msg - .tool_calls - .as_ref() - .map(|c| !c.is_empty()) - .unwrap_or(false); - if has_tool_calls && msg.content.trim().is_empty() { - if let Some(ref tcs) = msg.tool_calls { - for tc in tcs { - pending_calls.push_back(( - tc.function.name.clone(), - tc.function.arguments.clone(), - )); - } - } - continue; - } - let tools = std::mem::take(&mut pending_tools); - pending_calls.clear(); - rendered.push(RenderedPersonaMessage { - role: "assistant".to_string(), - content: msg.content, - is_initial: false, - tools: tools - .into_iter() - .map(|t| crate::ai::handlers::HistoryToolInvocation { - name: t.name, - arguments: t.arguments, - result: t.result, - result_truncated: t.result_truncated, - }) - .collect(), - }); - } - "user" => { - let is_initial = user_turns_seen == 0; - user_turns_seen += 1; - pending_tools.clear(); - pending_calls.clear(); - rendered.push(RenderedPersonaMessage { - role: "user".to_string(), - content: msg.content, - is_initial, - tools: Vec::new(), - }); - } - _ => continue, +/// Hard cap on a generated conversation title. Long enough to be +/// descriptive, short enough to fit a nav bar and a list row heading. +const TITLE_MAX_CHARS: usize = 48; + +/// Clean up whatever the model returned into a usable title. +/// +/// Small models are chatty about this task: they wrap titles in quotes, +/// prefix them with "Title:", tack on a trailing period, or return a whole +/// sentence. Everything here is defensive against that, and anything still +/// too long is truncated on a character boundary. +pub fn sanitize_title(raw: &str) -> String { + let mut text = raw.trim(); + // Some models answer with a preamble line then the title; take the first + // non-empty line and drop the rest. + if let Some(line) = text.lines().map(str::trim).find(|l| !l.is_empty()) { + text = line; + } + let lowered = text.to_lowercase(); + for prefix in ["title:", "conversation title:", "chat title:"] { + if lowered.starts_with(prefix) { + text = text[prefix.len()..].trim(); + break; } } + let text = text + .trim_matches(|c: char| c == '"' || c == '\'' || c == '“' || c == '”' || c == '*') + .trim() + .trim_end_matches('.') + .trim(); + let flattened = text.split_whitespace().collect::>().join(" "); + if flattened.chars().count() <= TITLE_MAX_CHARS { + return flattened; + } + let truncated: String = flattened.chars().take(TITLE_MAX_CHARS).collect(); + format!("{}…", truncated.trim_end()) +} - rendered +/// Title to use when the model can't be reached or returns nothing usable. +/// +/// The user's opening question is a better name than "Untitled": it is +/// exactly the thing they would scan the list for. +pub fn fallback_title(messages: &[ChatMessage]) -> String { + messages + .iter() + .find(|m| m.role == "user") + .map(|m| sanitize_title(&m.content)) + .unwrap_or_default() +} + +/// The prompt used to name a conversation from its opening exchange. +pub fn title_prompt(user_message: &str, assistant_reply: &str) -> String { + format!( + "Summarize this conversation opening as a short title of at most six \ + words. Reply with the title alone — no quotes, no punctuation at the \ + end, no preamble.\n\nUser: {}\n\nAssistant: {}", + conversation_snippet(user_message), + conversation_snippet(assistant_reply), + ) +} + +/// Soft cap for the preview line on the chat list. Long enough to tell two +/// conversations apart, short enough that a row stays two lines on a phone. +const SNIPPET_MAX_CHARS: usize = 140; + +/// Collapse a message into a single-line preview for the chat list. +/// +/// Newlines become spaces so a markdown reply doesn't blow the row height, +/// and the cut is taken on a character boundary — slicing a UTF-8 string by +/// byte index panics the moment a reply contains an emoji or an accent. +pub fn conversation_snippet(content: &str) -> String { + let flattened = content.split_whitespace().collect::>().join(" "); + if flattened.chars().count() <= SNIPPET_MAX_CHARS { + return flattened; + } + let truncated: String = flattened.chars().take(SNIPPET_MAX_CHARS).collect(); + format!("{}…", truncated.trim_end()) +} + +/// Map the internal fork markers onto the serialized wire type shared with +/// the file chat's history response. +fn wire_fork_info( + fork_info: Vec>, +) -> Vec> { + fork_info + .into_iter() + .map(|f| { + f.map(|fi| crate::ai::handlers::ChatForkInfo { + position: fi.position, + total: fi.total, + node_id: fi.node_id, + }) + }) + .collect() +} + +/// One row of the persona chat list: enough to render a conversation card +/// without shipping the whole transcript. +#[derive(Debug, Serialize)] +pub struct PersonaConversationSummary { + pub conversation_id: String, + pub persona_id: String, + pub persona_name: String, + /// Generated name for the conversation. Empty until the first turn + /// completes; the client falls back to the persona name. + pub title: String, + /// Last rendered message on the active branch, flattened to one line. + pub snippet: String, + /// Role behind `snippet`, so the list can prefix the user's own last + /// word with "You:". + pub snippet_role: String, + /// Assistant turns on the active branch. + pub turn_count: u32, + /// Rendered messages on the active branch (user + assistant bubbles). + pub message_count: usize, + pub created_at: i64, + pub updated_at: i64, + /// Whether the conversation has been forked — the list marks these so a + /// user can tell which transcripts have alternates parked in them. + pub has_branches: bool, } impl PersonaChatHistoryView { pub fn empty() -> Self { Self { + conversation_id: String::new(), + persona_id: String::new(), + title: String::new(), messages: Vec::new(), turn_count: 0, model_version: String::new(), @@ -317,20 +443,18 @@ impl PersonaChatHistoryView { // Per-persona in-flight mutex // ───────────────────────────────────────────────────────────────────── -/// One async mutex per (user_id, persona_id) — serializes concurrent -/// turns for the same persona so the JSON blob doesn't race. Held for -/// the lifetime of the spawned turn task. -pub type PersonaLockMap = TokioMutex>>>; +/// One async mutex per conversation — serializes concurrent writers to a +/// single transcript (a turn appending, a rewind re-pointing the leaf) so +/// the stored tree never interleaves two mutations. +pub type PersonaLockMap = TokioMutex>>>; -/// Tracks in-flight persona turns so a second dispatch for the same -/// `(user_id, persona_id)` pair is rejected (HTTP 409) instead of -/// silently queueing behind the per-persona lock, where it would sit -/// "running" (and the client pending) until the first turn finishes. -/// The lock in `run_streaming_turn` stays as the last-line race guard -/// for the persisted JSON blob. +/// Tracks the turn in flight for each conversation so a second dispatch +/// against the same conversation is rejected (HTTP 409) instead of racing +/// the first. Keyed on `conversation_id` alone — ids are uuids, so they +/// don't collide across users. #[derive(Default)] pub struct InFlightPersonaTurns { - map: StdMutex>, + turns: StdMutex>, } impl InFlightPersonaTurns { @@ -338,46 +462,38 @@ impl InFlightPersonaTurns { Self::default() } - /// Claim `(user_id, persona_id)` for `turn_id`. Fails with the id of - /// the turn already in flight when the slot is taken. - pub fn claim(&self, user_id: i32, persona_id: &str, turn_id: &str) -> Result<(), String> { - let mut map = self.map.lock().expect("in-flight map poisoned"); - let key = (user_id, persona_id.to_string()); - if let Some(existing) = map.get(&key) { + /// Claim `conversation_id` for `turn_id`. Fails with the id of the turn + /// already holding the slot, which the handler reports so the client can + /// re-attach to it rather than just seeing a bare conflict. + pub fn claim(&self, conversation_id: &str, turn_id: &str) -> Result<(), String> { + let mut map = self.turns.lock().expect("in-flight turns poisoned"); + if let Some(existing) = map.get(conversation_id) { return Err(existing.clone()); } - map.insert(key, turn_id.to_string()); + map.insert(conversation_id.to_string(), turn_id.to_string()); Ok(()) } - /// Release the slot. Idempotent — removing an absent key is a no-op. - pub fn release(&self, user_id: i32, persona_id: &str) { - if let Ok(mut map) = self.map.lock() { - map.remove(&(user_id, persona_id.to_string())); + pub fn release(&self, conversation_id: &str) { + if let Ok(mut map) = self.turns.lock() { + map.remove(conversation_id); } } } -/// RAII release of an in-flight slot. Moved into the spawned turn task so -/// the slot is freed when the task ends — whether by natural completion, -/// an error, or an abort from the cancel endpoint. +/// Frees the in-flight slot when the spawned turn task ends — on completion, +/// error, or abort. struct InFlightGuard { in_flight: Arc, - user_id: i32, - persona_id: String, + conversation_id: String, } impl Drop for InFlightGuard { fn drop(&mut self) { - self.in_flight.release(self.user_id, &self.persona_id); + self.in_flight.release(&self.conversation_id); } } -// ───────────────────────────────────────────────────────────────────── -// Session — owns the agent generator, the persona DAO, and the turn -// registry, and exposes dispatch + streaming. -// ───────────────────────────────────────────────────────────────────── - #[derive(Clone)] pub struct PersonaChatSession { generator: Arc, @@ -400,29 +516,93 @@ impl PersonaChatSession { } } - /// Load the rolling transcript for `(user_id, persona_id)`. Returns an - /// empty envelope when no conversation has been started yet. - pub fn load_history( + /// Start a new conversation with a persona. + /// + /// Always a fresh row: a persona is expected to hold several separate + /// conversations, so this never resumes an existing one. + pub fn create_conversation( &self, user_id: i32, persona_id: &str, - ) -> Result { + ) -> Result { + let persona_id = persona_id.trim(); + if persona_id.is_empty() { + return Err(PersonaChatError::UnknownPersona(String::new())); + } let cx = opentelemetry::Context::current(); let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); - let row = dao - .get_persona_chat(&cx, user_id, persona_id) - .map_err(|e| anyhow!("failed to load persona chat: {e:?}"))?; - let Some(row) = row else { - return Ok(PersonaChatHistoryView::empty()); + // Validate up front rather than letting the first turn 404: a + // conversation row pointing at a persona that doesn't exist would be + // a permanently dead entry in the list. + let persona = dao + .get_persona(&cx, user_id, persona_id) + .map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))?; + if persona.is_none() { + return Err(PersonaChatError::UnknownPersona(persona_id.to_string())); + } + dao.create_persona_chat(&cx, user_id, persona_id, Utc::now().timestamp_millis()) + .map_err(|e| PersonaChatError::Db(anyhow!("{e:?}"))) + } + + /// Delete a conversation and its transcript outright. + pub fn delete_conversation(&self, user_id: i32, conversation_id: &str) -> Result<()> { + let cx = opentelemetry::Context::current(); + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + dao.delete_persona_chat(&cx, user_id, conversation_id) + .map_err(|e| anyhow!("failed to delete persona chat: {e:?}"))?; + Ok(()) + } + + /// Load a transcript branch for one conversation. + /// + /// `branch_id` renders an alternate leaf without making it active, so + /// the client can preview a fork before committing to it; `None` renders + /// the active branch. Returns an empty envelope for a conversation that + /// has been created but not yet spoken in. + pub fn load_history( + &self, + user_id: i32, + conversation_id: &str, + branch_id: Option, + ) -> Result { + let cx = opentelemetry::Context::current(); + let row = { + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + dao.get_persona_chat(&cx, user_id, conversation_id) + .map_err(|e| anyhow!("failed to load persona chat: {e:?}"))? }; - let messages = decode_history(&row.messages_json)?; - // Hide the seed system/greeting, then fold tool scaffolding into - // rendered lines (the client shares the file chat's renderer). - let messages = render_flat_transcript(strip_seed(messages)); - Ok(PersonaChatHistoryView { - turn_count: row.turn_count as u32, - messages, + let Some(row) = row else { + bail!("conversation not found"); + }; + let store = decode_store(&row.messages_json)?; + + let identity = PersonaChatHistoryView { + conversation_id: row.conversation_id.clone(), + persona_id: row.persona_id.clone(), + title: row.title.clone(), ..PersonaChatHistoryView::empty() + }; + // A conversation holding only its seed system node renders as the + // empty state, not as a zero-message transcript with a leaf id. + if store.nodes.is_empty() { + return Ok(identity); + } + + let target_leaf = branch_id.unwrap_or(store.active_leaf_id); + let path = store + .path_to_leaf(target_leaf) + .ok_or_else(|| anyhow!("branch_id {target_leaf} not found in tree"))?; + let (messages, turn_count, _node_ids, fork_info) = render_persona_path(&store, &path); + + Ok(PersonaChatHistoryView { + messages, + // Assistant turns actually on this path — not the stored count, + // which tracks the active branch. + turn_count: turn_count as u32, + active_leaf_id: store.active_leaf_id, + viewing_branch_id: target_leaf, + fork_info: wire_fork_info(fork_info), + ..identity }) } @@ -443,22 +623,31 @@ impl PersonaChatSession { if trimmed.len() > 8192 { return Err(PersonaChatError::MessageTooLong); } - let persona_id = req.persona_id.trim().to_string(); - if persona_id.is_empty() { - return Err(PersonaChatError::UnknownPersona(persona_id)); + let conversation_id = req.conversation_id.trim().to_string(); + if conversation_id.is_empty() { + return Err(PersonaChatError::UnknownConversation(conversation_id)); } let cx = opentelemetry::Context::current(); - let persona = { + // The persona comes from the stored conversation, not the request: + // a transcript's voice is fixed when it is created, so a stale + // client can't quietly swap it half-way through a thread. + let (persona, persona_id) = { let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); - dao.get_persona(&cx, user_id, &persona_id) + let row = dao + .get_persona_chat(&cx, user_id, &conversation_id) .map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))? + .ok_or_else(|| PersonaChatError::UnknownConversation(conversation_id.clone()))?; + let persona = dao + .get_persona(&cx, user_id, &row.persona_id) + .map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))? + .ok_or_else(|| PersonaChatError::UnknownPersona(row.persona_id.clone()))?; + (persona, row.persona_id) }; - let persona = persona.ok_or_else(|| PersonaChatError::UnknownPersona(persona_id.clone()))?; let turn_id = Uuid::new_v4().to_string(); self.in_flight - .claim(user_id, &persona_id, &turn_id) + .claim(&conversation_id, &turn_id) .map_err(|_| PersonaChatError::ConcurrentTurn)?; let entry = Arc::new(TurnEntry::new_persona( @@ -478,14 +667,15 @@ impl PersonaChatSession { }; let in_flight_guard = InFlightGuard { in_flight: self.in_flight.clone(), - user_id, - persona_id: persona_id.clone(), + conversation_id: conversation_id.clone(), }; + let conversation_for_span = conversation_id.clone(); let handle = tokio::spawn(async move { let tracer = global_tracer(); let mut span = tracer.start("ai.persona_chat.turn.execute"); span.set_attribute(KeyValue::new("turn_id", turn_id_for_task.clone())); span.set_attribute(KeyValue::new("persona_id", persona_id.clone())); + span.set_attribute(KeyValue::new("conversation_id", conversation_for_span)); span.set_attribute(KeyValue::new("library_id", library_id as i64)); let result = svc @@ -532,33 +722,63 @@ impl PersonaChatSession { user_id: i32, _library_id: i32, ) -> Result<()> { - // Per-persona mutex — serializes concurrent turns so the JSON - // blob doesn't race. - let lock_key = (user_id, persona.persona_id.clone()); + // Per-conversation mutex — serializes this turn against a + // concurrent rewind or branch switch on the same transcript. + let conversation_id = req.conversation_id.trim().to_string(); let lock = { let mut locks = self.chat_locks.lock().await; locks - .entry(lock_key.clone()) + .entry(conversation_id.clone()) .or_insert_with(|| Arc::new(TokioMutex::new(()))) .clone() }; let _guard = lock.lock().await; - // Build the messages vector from persisted history. + // Replay the active branch of the conversation tree. let cx = opentelemetry::Context::current(); let row = { let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); - dao.get_persona_chat(&cx, user_id, &persona.persona_id) + dao.get_persona_chat(&cx, user_id, &conversation_id) .map_err(|e| anyhow!("failed to load persona chat: {e:?}"))? }; + let row = row.ok_or_else(|| anyhow!("conversation not found"))?; + let needs_title = row.title.trim().is_empty(); - let mut messages = match row { - Some(r) => decode_history(&r.messages_json)?, - None => { - let prompt = resolve_persona_system_prompt(Some(&persona)) - .unwrap_or_else(|| "You are a helpful assistant.".to_string()); - seed_messages_for_persona(&prompt, &persona.name) + let system_prompt = resolve_persona_system_prompt(Some(&persona)) + .unwrap_or_else(|| "You are a helpful assistant.".to_string()); + + let stored = Some(decode_store(&row.messages_json)?); + // `seeded` marks a conversation whose messages are NOT yet backed by + // tree nodes, so the persistence step below knows to write the seed + // out along with the turn. + let (mut store, mut messages, seeded) = match stored { + Some(store) if !store.nodes.is_empty() => { + let mut messages: Vec = { + let path = store.path_to_leaf(store.active_leaf_id).ok_or_else(|| { + anyhow!("active_leaf_id {} not found in tree", store.active_leaf_id) + })?; + path.iter().map(|n| n.message.clone()).collect() + }; + // Re-resolve the system prompt from the persona on every turn. + // The stored node stays as the historical record of what the + // model saw; editing a persona has to take effect on + // conversations already in flight, which it would not if we + // replayed the copy seeded at conversation start forever. + if let Some(sys) = messages.first_mut() + && sys.role == "system" + { + sys.content = system_prompt.clone(); + } + (store, messages, false) } + _ => ( + ChatHistoryStore { + nodes: Vec::new(), + active_leaf_id: 0, + }, + seed_messages_for_persona(&system_prompt), + true, + ), }; // Backend selection. Defaults to local; explicit `backend` wins. @@ -583,15 +803,19 @@ impl PersonaChatSession { min_p: req.min_p, enable_thinking: req.enable_thinking, }; - let backend = self.generator.resolve_backend(backend_kind, &overrides).await?; + let backend = self + .generator + .resolve_backend(backend_kind, &overrides) + .await?; let model_used = backend.model().to_string(); // Build a no-photo tool catalog. `current_gate_opts_for_persona` // already probes each tool's backing table for presence and // honours the persona's `allow_agent_corrections` toggle. - let gate_opts = self - .generator - .current_gate_opts_for_persona(backend.images_inline, Some((user_id, &persona.persona_id))); + let gate_opts = self.generator.current_gate_opts_for_persona( + backend.images_inline, + Some((user_id, &persona.persona_id)), + ); let tools = InsightGenerator::build_tool_definitions(gate_opts); // Append the new user turn + apply the per-turn system-prompt @@ -600,12 +824,32 @@ impl PersonaChatSession { let override_text = req.system_prompt.as_deref(); let prompt_override = if let Some(s) = override_text { let t = s.trim(); - if t.is_empty() { None } else { Some(t.to_string()) } + if t.is_empty() { + None + } else { + Some(t.to_string()) + } } else { None }; - let base_count = messages.len(); + // Trim to the model's context window before the turn. Without this a + // rolling persona transcript grows unbounded until the backend + // silently drops the front of it mid-conversation. + let budget_tokens = (req.num_ctx.unwrap_or_else(env_default_num_ctx) as usize) + .saturating_sub(RESPONSE_HEADROOM_TOKENS); + let budget_bytes = budget_tokens.saturating_mul(BYTES_PER_TOKEN); + let truncated = apply_context_budget(&mut messages, budget_bytes); + if truncated { + let _ = entry.push_event(ChatStreamEvent::Truncated).await; + } + + // How much of `messages` is already backed by tree nodes. Read after + // the budget pass, because truncation drains replayed history out of + // the middle — taking the path length instead would misalign the + // slice below and re-append old turns as new nodes. + let in_tree_len = if seeded { 0 } else { messages.len() }; + messages.push(ChatMessage::user(req.user_message.clone())); if let Some(ref note) = prompt_override { // Suffix the system prompt with the override for this turn only. @@ -634,6 +878,13 @@ impl PersonaChatSession { .await .map_err(|e| anyhow!("persona chat loop failed: {e:?}"))?; + // Cancelled mid-flight: the DELETE handler has already pushed the + // terminal frame and flipped the registry status. Persisting here + // would commit a half-finished turn and emit a second terminal frame. + if outcome.cancelled { + return Ok(()); + } + // Restore the persisted system prompt before serializing (the // override was for this turn only). if let Some(ref note) = prompt_override @@ -645,21 +896,68 @@ impl PersonaChatSession { } } - // Persist the updated transcript. `messages[0..base_count]` is the - // persisted slice; everything past that is the new turn. - let json = encode_history(&messages)?; - let turn_count = messages.len().saturating_sub(base_count) as i32; + // Append this turn as a chain of nodes hanging off the branch we + // replayed. Rewind and fork work by re-pointing `active_leaf_id` at + // an earlier node, so a discarded path stays reachable instead of + // being overwritten. + let mut parent_id = if seeded { + None + } else { + Some(store.active_leaf_id) + }; + for msg in &messages[in_tree_len..] { + parent_id = Some(store.append_node(parent_id, msg.clone())); + } + if let Some(last_id) = parent_id { + store.active_leaf_id = last_id; + } + + let json = encode_store(&store)?; + // Assistant turns on the new active branch. Cumulative, unlike the + // per-write delta this column used to hold, so the list screen can + // show how long a conversation actually is. + let turn_count = store + .path_to_leaf(store.active_leaf_id) + .map(|path| render_persona_path(&store, &path).1) + .unwrap_or(0) as i32; { let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); - dao.upsert_persona_chat(&cx, user_id, &persona.persona_id, &json, turn_count, Utc::now().timestamp_millis()) + let rows = dao + .update_persona_chat( + &cx, + user_id, + &conversation_id, + &json, + turn_count, + Utc::now().timestamp_millis(), + ) .map_err(|e| anyhow!("failed to persist persona chat: {e:?}"))?; + if rows == 0 { + bail!("conversation not found"); + } + } + + // Name the conversation off its opening exchange. Best-effort and + // after persistence: a failed title must not lose the turn, and the + // list falls back to the persona name while the title is empty. + if needs_title { + let title = self + .generate_title(&backend, &messages) + .await + .unwrap_or_else(|| fallback_title(&messages)); + if !title.is_empty() { + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + if let Err(e) = dao.set_persona_chat_title(&cx, user_id, &conversation_id, &title) { + log::warn!("failed to store persona chat title: {e:?}"); + } + } } let _ = entry .push_event(ChatStreamEvent::Done { tool_calls_made: outcome.tool_calls_made, iterations_used: outcome.iterations_used, - truncated: false, + truncated, prompt_tokens: outcome.last_prompt_eval_count, eval_tokens: outcome.last_eval_count, num_ctx: req.num_ctx, @@ -672,16 +970,243 @@ impl PersonaChatSession { Ok(()) } - /// Wipe the rolling transcript for `(user_id, persona_id)`. Reserved - /// for a future "New conversation" affordance; the v1 UI doesn't call - /// this but the endpoint is shipped so the contract is stable. - pub fn reset(&self, user_id: i32, persona_id: &str) -> Result<()> { + /// Every conversation this user has going, newest first. + /// + /// Conversations whose persona has since been deleted are omitted: they + /// cannot be resumed (dispatching a turn against an unknown persona is a + /// 404), so listing them would only offer a dead-end tap target. + /// Conversations that hold nothing but the seed are omitted too — the + /// user has never actually said anything in them. + pub fn list_conversations(&self, user_id: i32) -> Result> { let cx = opentelemetry::Context::current(); + let (rows, personas) = { + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + let rows = dao + .list_persona_chats(&cx, user_id) + .map_err(|e| anyhow!("failed to list persona chats: {e:?}"))?; + let personas = dao + .list_personas(&cx, user_id) + .map_err(|e| anyhow!("failed to list personas: {e:?}"))?; + (rows, personas) + }; + let names: HashMap = personas + .into_iter() + .map(|p| (p.persona_id, p.name)) + .collect(); + + let mut out = Vec::with_capacity(rows.len()); + for entry in rows { + let Some(persona_name) = names.get(&entry.persona_id).cloned() else { + continue; + }; + // A single unreadable transcript shouldn't take the whole list + // down with it — skip the row and keep going. + let Ok(store) = decode_store(&entry.messages_json) else { + log::warn!( + "skipping unreadable persona chat transcript {}", + entry.conversation_id + ); + continue; + }; + // An empty conversation still gets a row: the user created it and + // needs somewhere to tap back into. It just has no preview yet. + let rendered = store + .path_to_leaf(store.active_leaf_id) + .map(|path| render_persona_path(&store, &path)); + let (messages, turn_count) = match rendered { + Some((messages, turn_count, _ids, _forks)) => (messages, turn_count), + None => (Vec::new(), 0), + }; + let last = messages.last(); + + out.push(PersonaConversationSummary { + conversation_id: entry.conversation_id, + persona_id: entry.persona_id, + persona_name, + title: entry.title, + snippet: last + .map(|m| conversation_snippet(&m.content)) + .unwrap_or_default(), + snippet_role: last.map(|m| m.role.clone()).unwrap_or_default(), + turn_count: turn_count as u32, + message_count: messages.len(), + created_at: entry.created_at, + updated_at: entry.updated_at, + has_branches: store.leaves().len() > 1, + }); + } + Ok(out) + } + + /// Ask the model to name a conversation from its opening exchange. + /// + /// Returns `None` on any failure — a title is a nicety, and the caller + /// falls back to the user's first message. Uses the same backend the + /// turn ran on so a local-only setup never reaches the network. + async fn generate_title( + &self, + backend: &crate::ai::backend::ResolvedBackend, + messages: &[ChatMessage], + ) -> Option { + let user_message = messages.iter().find(|m| m.role == "user")?; + // The last assistant message with actual prose — earlier ones may be + // empty tool-dispatch scaffolding. + let reply = messages + .iter() + .rev() + .find(|m| m.role == "assistant" && !m.content.trim().is_empty())?; + + let prompt = title_prompt(&user_message.content, &reply.content); + match backend.chat().generate(&prompt, None, None).await { + Ok(raw) => { + let title = sanitize_title(&crate::ai::llm_client::strip_think_blocks(&raw)); + if title.is_empty() { None } else { Some(title) } + } + Err(e) => { + log::warn!("persona chat title generation failed: {e:?}"); + None + } + } + } + + /// Load the tree, mutate it under the per-conversation lock, and persist. + /// + /// Every branch operation is the same three steps around a different + /// mutation, and all of them have to serialize against an in-flight turn + /// (which appends to `active_leaf_id`) or the two writers clobber each + /// other's copy of the transcript. + async fn mutate_store(&self, user_id: i32, conversation_id: &str, mutate: F) -> Result<()> + where + F: FnOnce(&mut ChatHistoryStore) -> Result<()>, + { + let lock = { + let mut locks = self.chat_locks.lock().await; + locks + .entry(conversation_id.to_string()) + .or_insert_with(|| Arc::new(TokioMutex::new(()))) + .clone() + }; + let _guard = lock.lock().await; + + let cx = opentelemetry::Context::current(); + let row = { + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + dao.get_persona_chat(&cx, user_id, conversation_id) + .map_err(|e| anyhow!("failed to load persona chat: {e:?}"))? + }; + let row = row.ok_or_else(|| anyhow!("conversation not found"))?; + let mut store = decode_store(&row.messages_json)?; + if store.nodes.is_empty() { + bail!("no chat history for this conversation"); + } + + mutate(&mut store)?; + + let json = encode_store(&store)?; + let turn_count = store + .path_to_leaf(store.active_leaf_id) + .map(|path| render_persona_path(&store, &path).1) + .unwrap_or(0) as i32; let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); - dao.clear_persona_chat(&cx, user_id, persona_id) - .map_err(|e| anyhow!("failed to reset persona chat: {e:?}"))?; + let rows = dao + .update_persona_chat( + &cx, + user_id, + conversation_id, + &json, + turn_count, + Utc::now().timestamp_millis(), + ) + .map_err(|e| anyhow!("failed to persist persona chat: {e:?}"))?; + if rows == 0 { + bail!("conversation not found"); + } Ok(()) } + + /// Rewind so the rendered message at `discard_from_rendered_index` — and + /// everything after it — leaves the active path. + /// + /// Nothing is deleted: the active leaf is re-pointed at the last kept + /// node, and the discarded path survives as a fork the user can switch + /// back to. Unlike the file chat, index 0 is rewindable: its index 0 is a + /// synthetic "describe this photo" prompt, whereas here it is the user's + /// own first question and editing it is a reasonable thing to want. + pub async fn rewind( + &self, + user_id: i32, + conversation_id: &str, + discard_from_rendered_index: usize, + ) -> Result<()> { + self.mutate_store(user_id, conversation_id, |store| { + let path = store + .path_to_leaf(store.active_leaf_id) + .ok_or_else(|| anyhow!("active_leaf_id not found in tree"))?; + let (_rendered, _turns, node_ids, _forks) = render_persona_path(store, &path); + + store.active_leaf_id = + rewind_target_leaf(store, &node_ids, discard_from_rendered_index)?; + Ok(()) + }) + .await + } + + /// Make `leaf_id` the active branch. The path being left behind becomes + /// a regular fork rather than being discarded. + pub async fn switch_branch( + &self, + user_id: i32, + conversation_id: &str, + leaf_id: u64, + ) -> Result<()> { + self.mutate_store(user_id, conversation_id, |store| { + if !store.nodes.iter().any(|n| n.id == leaf_id) { + bail!("branch_id {leaf_id} not found in tree"); + } + if !store.children_of(leaf_id).is_empty() { + bail!("branch_id {leaf_id} is not a leaf node"); + } + store.active_leaf_id = leaf_id; + Ok(()) + }) + .await + } + + /// List the branches of a persona's conversation tree. + /// + /// With `node_id` set (from a rendered message's `fork_info.node_id`) + /// the list is the position-ranked siblings at that one divergence; + /// without it, every leaf in the tree. + pub fn get_branches( + &self, + user_id: i32, + conversation_id: &str, + node_id: Option, + viewing_leaf: Option, + ) -> Result<(Vec, u64)> { + let cx = opentelemetry::Context::current(); + let row = { + let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); + dao.get_persona_chat(&cx, user_id, conversation_id) + .map_err(|e| anyhow!("failed to load persona chat: {e:?}"))? + }; + let row = row.ok_or_else(|| anyhow!("conversation not found"))?; + let store = decode_store(&row.messages_json)?; + if store.nodes.is_empty() { + bail!("no chat history for this conversation"); + } + + let list = match node_id { + Some(fork_node) => { + if !store.nodes.iter().any(|n| n.id == fork_node) { + bail!("node_id {fork_node} not found in tree"); + } + store.branch_options_at(fork_node, viewing_leaf.unwrap_or(store.active_leaf_id)) + } + None => store.leaves_with_info(), + }; + Ok((list, store.active_leaf_id)) + } } // ───────────────────────────────────────────────────────────────────── @@ -693,16 +1218,19 @@ impl PersonaChatSession { /// persona transcript is keyed on `(user_id, persona_id)` only. #[derive(Debug, Deserialize)] pub struct PersonaChatHistoryQuery { - pub persona_id: String, + pub conversation_id: String, #[serde(default)] #[allow(dead_code)] pub library: Option, + /// Render the branch anchored at this leaf instead of the active one, + /// so the client can preview a fork without committing to it. + #[serde(default)] + pub branch_id: Option, } -/// Body for POST /persona_chat/reset — wipes the rolling transcript for -/// one persona. +/// Body for POST /persona_chat/conversations — start a new conversation. #[derive(Debug, Deserialize)] -pub struct PersonaChatResetRequest { +pub struct PersonaChatCreateRequest { pub persona_id: String, } @@ -716,16 +1244,20 @@ pub async fn persona_chat_history_handler( app_state: web::Data, ) -> impl Responder { let user_id = _claims.sub.parse::().unwrap_or(1); - match app_state - .persona_chat_session - .load_history(user_id, &query.persona_id) - { + match app_state.persona_chat_session.load_history( + user_id, + &query.conversation_id, + query.branch_id, + ) { Ok(view) => HttpResponse::Ok().json(view), Err(e) => { - log::error!("persona chat history load failed: {e}"); - HttpResponse::InternalServerError().json(serde_json::json!({ - "error": format!("{e}") - })) + let msg = format!("{e}"); + if msg.contains("not found") { + HttpResponse::NotFound().json(serde_json::json!({ "error": msg })) + } else { + log::error!("persona chat history load failed: {msg}"); + HttpResponse::InternalServerError().json(serde_json::json!({ "error": msg })) + } } } } @@ -743,7 +1275,10 @@ pub async fn persona_chat_turn_handler( let parent_context = extract_context_from_request(&http_request); let tracer = global_tracer(); let mut span = tracer.start_with_context("http.persona_chat.turn", &parent_context); - span.set_attribute(KeyValue::new("persona_id", request.persona_id.clone())); + span.set_attribute(KeyValue::new( + "conversation_id", + request.conversation_id.clone(), + )); // The transcript is not library-scoped, but the turn's tool catalog // (memories, SMS, places) resolves against a library — mirror the @@ -778,7 +1313,9 @@ pub async fn persona_chat_turn_handler( Err(e) => { span.set_status(Status::error(format!("{e}"))); match &e { - PersonaChatError::UnknownPersona(_) => HttpResponse::NotFound(), + PersonaChatError::UnknownPersona(_) | PersonaChatError::UnknownConversation(_) => { + HttpResponse::NotFound() + } PersonaChatError::EmptyMessage | PersonaChatError::MessageTooLong => { HttpResponse::BadRequest() } @@ -793,23 +1330,52 @@ pub async fn persona_chat_turn_handler( } } -/// POST /persona_chat/reset — wipe the rolling transcript for one -/// persona. Shipped so the contract is stable; the v1 UI has no -/// "New conversation" button yet. -#[post("/persona_chat/reset")] -pub async fn persona_chat_reset_handler( - _claims: Claims, - request: web::Json, +/// POST /persona_chat/conversations — start a new conversation with a +/// persona. Always creates a fresh transcript, so a persona can hold several +/// separate threads. +#[post("/persona_chat/conversations")] +pub async fn persona_chat_create_conversation_handler( + claims: Claims, + request: web::Json, app_state: web::Data, ) -> impl Responder { - let user_id = _claims.sub.parse::().unwrap_or(1); + let user_id = claims.sub.parse::().unwrap_or(1); match app_state .persona_chat_session - .reset(user_id, &request.persona_id) + .create_conversation(user_id, &request.persona_id) { - Ok(()) => HttpResponse::Ok().json(serde_json::json!({ "reset": true })), + Ok(conversation_id) => HttpResponse::Created().json(serde_json::json!({ + "conversation_id": conversation_id, + })), + Err(e) => match &e { + PersonaChatError::UnknownPersona(_) => { + HttpResponse::NotFound().json(serde_json::json!({ "error": format!("{e}") })) + } + _ => { + log::error!("persona chat create failed: {e}"); + HttpResponse::InternalServerError() + .json(serde_json::json!({ "error": format!("{e}") })) + } + }, + } +} + +/// DELETE /persona_chat/conversations/{conversation_id} — remove a +/// conversation and its transcript. +#[delete("/persona_chat/conversations/{conversation_id}")] +pub async fn persona_chat_delete_conversation_handler( + claims: Claims, + path: web::Path, + app_state: web::Data, +) -> impl Responder { + let user_id = claims.sub.parse::().unwrap_or(1); + match app_state + .persona_chat_session + .delete_conversation(user_id, &path.into_inner()) + { + Ok(()) => HttpResponse::Ok().json(serde_json::json!({ "deleted": true })), Err(e) => { - log::error!("persona chat reset failed: {e}"); + log::error!("persona chat delete failed: {e}"); HttpResponse::InternalServerError().json(serde_json::json!({ "error": format!("{e}") })) @@ -817,6 +1383,146 @@ pub async fn persona_chat_reset_handler( } } +/// GET /persona_chat/conversations — every conversation this user has +/// going, newest first. Backs the chat list screen; the detail screen loads +/// a transcript from `/persona_chat/history`. +#[get("/persona_chat/conversations")] +pub async fn persona_chat_conversations_handler( + claims: Claims, + app_state: web::Data, +) -> impl Responder { + let user_id = claims.sub.parse::().unwrap_or(1); + match app_state.persona_chat_session.list_conversations(user_id) { + Ok(conversations) => HttpResponse::Ok().json(serde_json::json!({ + "conversations": conversations, + })), + Err(e) => { + log::error!("persona chat conversation list failed: {e}"); + HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("{e}") + })) + } + } +} + +/// Body for POST /persona_chat/rewind. +#[derive(Debug, Deserialize)] +pub struct PersonaChatRewindRequest { + pub conversation_id: String, + /// 0-based index into the rendered transcript. This message and + /// everything after it leaves the active path (and stays reachable as a + /// fork). Unlike the file chat, 0 is a legal index here. + pub discard_from_rendered_index: usize, +} + +/// Body for POST /persona_chat/switch-branch. +#[derive(Debug, Deserialize)] +pub struct PersonaChatSwitchBranchRequest { + pub conversation_id: String, + /// Leaf node to make active. Must be an existing leaf in the tree. + pub branch_id: u64, +} + +/// Query params for GET /persona_chat/branches. +#[derive(Debug, Deserialize)] +pub struct PersonaChatBranchesQuery { + pub conversation_id: String, + /// From a rendered message's `fork_info.node_id` — scopes the list to + /// the siblings at that one divergence instead of every leaf. + #[serde(default)] + pub node_id: Option, + /// The leaf the client is showing, so scoped options in the same subtree + /// anchor to it and the client can spot "the branch I am on". + #[serde(default)] + pub viewing_branch_id: Option, +} + +/// Map a branch-operation error onto the same status codes the file chat's +/// equivalents return, so the client can share its error handling. +fn branch_error_response(context: &str, e: &anyhow::Error) -> HttpResponse { + let msg = format!("{e}"); + log::error!("{context}: {msg}"); + if msg.contains("conversation not found") { + HttpResponse::NotFound().json(serde_json::json!({ "error": msg })) + } else if msg.contains("no chat history") { + HttpResponse::Conflict().json(serde_json::json!({ "error": msg })) + } else if msg.contains("not found") + || msg.contains("not a leaf") + || msg.contains("out of range") + || msg.contains("cannot rewind past") + { + HttpResponse::BadRequest().json(serde_json::json!({ "error": msg })) + } else { + HttpResponse::InternalServerError().json(serde_json::json!({ "error": msg })) + } +} + +/// POST /persona_chat/rewind — move the tail of the conversation off the +/// active path. Nothing is deleted; the discarded path stays reachable as a +/// fork, which is what makes "edit & resend" non-destructive. +#[post("/persona_chat/rewind")] +pub async fn persona_chat_rewind_handler( + claims: Claims, + request: web::Json, + app_state: web::Data, +) -> impl Responder { + let user_id = claims.sub.parse::().unwrap_or(1); + match app_state + .persona_chat_session + .rewind( + user_id, + &request.conversation_id, + request.discard_from_rendered_index, + ) + .await + { + Ok(()) => HttpResponse::Ok().json(serde_json::json!({ "success": true })), + Err(e) => branch_error_response("persona chat rewind failed", &e), + } +} + +/// POST /persona_chat/switch-branch — make an alternate path the active one. +#[post("/persona_chat/switch-branch")] +pub async fn persona_chat_switch_branch_handler( + claims: Claims, + request: web::Json, + app_state: web::Data, +) -> impl Responder { + let user_id = claims.sub.parse::().unwrap_or(1); + match app_state + .persona_chat_session + .switch_branch(user_id, &request.conversation_id, request.branch_id) + .await + { + Ok(()) => HttpResponse::Ok().json(serde_json::json!({ "success": true })), + Err(e) => branch_error_response("persona chat branch switch failed", &e), + } +} + +/// GET /persona_chat/branches — list the alternate paths of a persona's +/// conversation. Pair an entry's `id` with the history endpoint's +/// `branch_id` to preview it, or with switch-branch to adopt it. +#[get("/persona_chat/branches")] +pub async fn persona_chat_branches_handler( + claims: Claims, + query: web::Query, + app_state: web::Data, +) -> impl Responder { + let user_id = claims.sub.parse::().unwrap_or(1); + match app_state.persona_chat_session.get_branches( + user_id, + &query.conversation_id, + query.node_id, + query.viewing_branch_id, + ) { + Ok((branches, active_leaf_id)) => HttpResponse::Ok().json(serde_json::json!({ + "branches": branches, + "active_leaf_id": active_leaf_id, + })), + Err(e) => branch_error_response("persona chat branch list failed", &e), + } +} + /// GET /persona_chat/turn/{turn_id} — SSE replay for a persona turn. /// Delegates to the file chat's replay handler: the registry is keyed /// on `turn_id` only, and the `turn_info` frame already carries @@ -897,63 +1603,159 @@ mod tests { } #[test] - fn seed_messages_for_persona_includes_system_and_greeting() { - let msgs = seed_messages_for_persona("Be terse.", "Journal"); - assert_eq!(msgs.len(), 2); + fn seed_messages_for_persona_is_system_only() { + // No synthetic first user turn (there is no photo to anchor on) and + // no greeting (v1 shipped one, but it never rendered). + let msgs = seed_messages_for_persona("Be terse."); + assert_eq!(msgs.len(), 1); assert_eq!(msgs[0].role, "system"); assert_eq!(msgs[0].content, "Be terse."); - assert_eq!(msgs[1].role, "assistant"); - assert!(msgs[1].content.starts_with("Hi — I'm")); - assert!(msgs[1].content.contains("Journal")); + } + + fn greeting(name: &str) -> ChatMessage { + assistant_msg(&format!( + "Hi — I'm {name} ready to help. Ask anything your tools can reach \ + (memories, files, SMS, calendar, places)." + )) } #[test] - fn strip_seed_drops_leading_system_and_greeting() { - let mut msgs = seed_messages_for_persona("sys", "Default"); - msgs.push(ChatMessage::user("hi".to_string())); - msgs.push(assistant_msg("hello")); - msgs.push(ChatMessage::user("how are you?".to_string())); - let stripped = strip_seed(msgs); - assert_eq!(stripped.len(), 3); - assert_eq!(stripped[0].role, "user"); - assert_eq!(stripped[0].content, "hi"); - assert_eq!(stripped[2].role, "user"); - assert_eq!(stripped[2].content, "how are you?"); - } - - #[test] - fn strip_seed_preserves_an_unrelated_leading_assistant() { - // If the first message isn't the seed greeting, leave it alone — - // strip_seed is conservative. - let msgs = vec![ - assistant_msg("carry-over from last open"), - ChatMessage::user("hi".to_string()), - ]; - let stripped = strip_seed(msgs); - assert_eq!(stripped.len(), 2); - assert_eq!(stripped[0].content, "carry-over from last open"); - } - - #[test] - fn encode_decode_round_trips_through_json() { - let msgs = vec![ + fn decode_store_upgrades_a_flat_transcript_into_a_linear_tree() { + let flat = vec![ ChatMessage::system("sys".to_string()), ChatMessage::user("hi".to_string()), assistant_msg("hello"), ]; - let json = encode_history(&msgs).unwrap(); - let back = decode_history(&json).unwrap(); - assert_eq!(back.len(), msgs.len()); - assert_eq!(back[1].role, "user"); - assert_eq!(back[2].content, "hello"); + let store = decode_store(&serde_json::to_string(&flat).unwrap()).unwrap(); + assert_eq!(store.nodes.len(), 3); + assert_eq!(store.nodes[0].parent_id, None); + assert_eq!(store.nodes[1].parent_id, Some(store.nodes[0].id)); + assert_eq!(store.nodes[2].parent_id, Some(store.nodes[1].id)); + assert_eq!(store.active_leaf_id, store.nodes[2].id); } #[test] - fn decode_history_rejects_garbage() { - let err = decode_history("not-json").unwrap_err(); + fn decode_store_drops_the_v1_seed_greeting_on_upgrade() { + // Pre-migration rows carry an assistant greeting the flat renderer + // hid. The tree renderer renders every assistant node, so the + // upgrade has to drop it or the user gains a bubble they never saw. + let flat = vec![ + ChatMessage::system("sys".to_string()), + greeting("Journal"), + ChatMessage::user("hi".to_string()), + ]; + let store = decode_store(&serde_json::to_string(&flat).unwrap()).unwrap(); + assert_eq!(store.nodes.len(), 2); + assert!( + !store.nodes.iter().any(|n| n.message.role == "assistant"), + "greeting node survived the upgrade" + ); + } + + #[test] + fn decode_store_keeps_a_real_reply_that_reads_like_the_greeting() { + // Same words, but after a user turn — a genuine reply, not the seed. + let flat = vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("who are you?".to_string()), + greeting("Journal"), + ]; + let store = decode_store(&serde_json::to_string(&flat).unwrap()).unwrap(); + assert_eq!(store.nodes.len(), 3); + assert_eq!(store.nodes[2].message.role, "assistant"); + } + + #[test] + fn encode_decode_store_round_trips_a_forked_tree() { + let mut store = ChatHistoryStore::from_flat_array(vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("q".to_string()), + ]); + let user_id = store.active_leaf_id; + let a1 = store.append_node(Some(user_id), assistant_msg("first answer")); + let a2 = store.append_node(Some(user_id), assistant_msg("second answer")); + store.active_leaf_id = a2; + + let back = decode_store(&encode_store(&store).unwrap()).unwrap(); + assert_eq!(back.active_leaf_id, a2); + assert_eq!(back.nodes.len(), 4); + assert_eq!(back.children_of(user_id).len(), 2); + assert_eq!(back.fork_at_node(a1), Some((1, 2))); + assert_eq!(back.fork_at_node(a2), Some((2, 2))); + } + + #[test] + fn decode_store_rejects_garbage() { + let err = decode_store("not-json").unwrap_err(); assert!(err.to_string().contains("failed to deserialize")); } + #[test] + fn render_persona_path_never_marks_a_message_initial() { + // The file chat reserves is_initial for its synthetic "describe this + // photo" prompt, which the UI refuses to rewind or regenerate. A + // persona chat's first message is the user's own question, so both + // actions have to stay available on it. + let store = ChatHistoryStore::from_flat_array(vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("first question".to_string()), + assistant_msg("reply"), + ]); + let path = store.path_to_leaf(store.active_leaf_id).unwrap(); + let (rendered, turn_count, node_ids, fork_info) = render_persona_path(&store, &path); + + assert_eq!(rendered.len(), 2, "the system node is not rendered"); + assert!(rendered.iter().all(|m| !m.is_initial)); + assert_eq!(rendered[0].role, "user"); + assert_eq!(rendered[1].role, "assistant"); + assert_eq!(turn_count, 1, "one assistant turn on this path"); + assert_eq!(node_ids.len(), 2); + assert!(fork_info.iter().all(|f| f.is_none()), "no divergence yet"); + } + + #[test] + fn render_persona_path_marks_the_divergent_reply_on_a_forked_tree() { + let mut store = ChatHistoryStore::from_flat_array(vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("q".to_string()), + ]); + let user_node = store.active_leaf_id; + let _a1 = store.append_node(Some(user_node), assistant_msg("first answer")); + let a2 = store.append_node(Some(user_node), assistant_msg("second answer")); + store.active_leaf_id = a2; + + let path = store.path_to_leaf(a2).unwrap(); + let (rendered, _turns, _ids, fork_info) = render_persona_path(&store, &path); + assert_eq!(rendered.len(), 2); + assert!( + fork_info[0].is_none(), + "the user turn itself has not forked" + ); + let f = fork_info[1] + .as_ref() + .expect("divergence on the second reply"); + assert_eq!((f.position, f.total), (2, 2)); + assert_eq!( + f.node_id, user_node, + "divergence is attributed to the user turn" + ); + } + + #[test] + fn wire_fork_info_preserves_position_total_and_node() { + let wired = wire_fork_info(vec![ + None, + Some(ForkInfo { + position: 2, + total: 3, + node_id: 7, + }), + ]); + assert!(wired[0].is_none()); + let f = wired[1].as_ref().unwrap(); + assert_eq!((f.position, f.total, f.node_id), (2, 3, 7)); + } + #[test] fn persona_chat_history_view_empty_has_zero_counts() { let v = PersonaChatHistoryView::empty(); @@ -967,7 +1769,7 @@ mod tests { #[test] fn persona_chat_validation_flags_empty_user_message() { let _req = PersonaChatTurnRequest { - persona_id: "default".to_string(), + conversation_id: "conv-1".to_string(), user_message: " ".to_string(), model: None, backend: None, @@ -989,7 +1791,7 @@ mod tests { #[test] fn persona_chat_validation_flags_oversized_user_message() { let _req = PersonaChatTurnRequest { - persona_id: "default".to_string(), + conversation_id: "conv-1".to_string(), user_message: "x".repeat(8193), model: None, backend: None, @@ -1025,21 +1827,21 @@ mod tests { #[test] fn persona_chat_turn_returns_409_when_concurrent_turn_in_flight() { let in_flight = InFlightPersonaTurns::new(); - assert!(in_flight.claim(1, "default", "t1").is_ok()); + assert!(in_flight.claim("conv-a", "t1").is_ok()); - // A second dispatch for the same (user, persona) is rejected with + // A second dispatch against the same conversation is rejected with // the in-flight turn's id (the handler maps this to HTTP 409), // and the rejection does not clobber the in-flight claim. - assert_eq!(in_flight.claim(1, "default", "t2").unwrap_err(), "t1"); - assert_eq!(in_flight.claim(1, "default", "t3").unwrap_err(), "t1"); + assert_eq!(in_flight.claim("conv-a", "t2").unwrap_err(), "t1"); + assert_eq!(in_flight.claim("conv-a", "t3").unwrap_err(), "t1"); - // Other personas and users are unaffected. - assert!(in_flight.claim(1, "journal", "t4").is_ok()); - assert!(in_flight.claim(2, "default", "t5").is_ok()); + // A second conversation with the same persona runs independently — + // that is the whole point of separate conversations. + assert!(in_flight.claim("conv-b", "t4").is_ok()); // Releasing frees the slot for a fresh dispatch. - in_flight.release(1, "default"); - assert!(in_flight.claim(1, "default", "t6").is_ok()); + in_flight.release("conv-a"); + assert!(in_flight.claim("conv-a", "t6").is_ok()); } #[test] @@ -1047,17 +1849,16 @@ mod tests { let in_flight = Arc::new(InFlightPersonaTurns::new()); let tracker = in_flight.clone(); // Mirrors dispatch_turn: claim first, guard holds the slot. - assert!(tracker.claim(1, "default", "t-guard").is_ok()); + assert!(tracker.claim("conv-a", "t-guard").is_ok()); { let _guard = InFlightGuard { in_flight, - user_id: 1, - persona_id: "default".to_string(), + conversation_id: "conv-a".to_string(), }; - assert_eq!(tracker.claim(1, "default", "other").unwrap_err(), "t-guard"); + assert_eq!(tracker.claim("conv-a", "other").unwrap_err(), "t-guard"); } // Guard dropped → slot free again. - assert!(tracker.claim(1, "default", "t7").is_ok()); + assert!(tracker.claim("conv-a", "t7").is_ok()); } #[test] @@ -1066,7 +1867,7 @@ mod tests { // file-chat request shapes. A camelCase regression here 400s the // client's dispatch with "missing field persona_id". let body = r#"{ - "persona_id": "journal", + "conversation_id": "conv-1", "user_message": "hello", "num_ctx": 4096, "top_p": 0.9, @@ -1077,7 +1878,7 @@ mod tests { "library": "main" }"#; let req: PersonaChatTurnRequest = serde_json::from_str(body).unwrap(); - assert_eq!(req.persona_id, "journal"); + assert_eq!(req.conversation_id, "conv-1"); assert_eq!(req.user_message, "hello"); assert_eq!(req.num_ctx, Some(4096)); assert_eq!(req.top_p, Some(0.9)); @@ -1091,6 +1892,9 @@ mod tests { #[test] fn persona_chat_history_view_serializes_snake_case_wire_shape() { let view = PersonaChatHistoryView { + conversation_id: "conv-1".to_string(), + persona_id: "journal".to_string(), + title: "June recap".to_string(), messages: vec![RenderedPersonaMessage { role: "user".to_string(), content: "hi".to_string(), @@ -1100,23 +1904,284 @@ mod tests { turn_count: 1, model_version: "x".to_string(), backend: "local".to_string(), - active_leaf_id: 0, - viewing_branch_id: 0, - fork_info: Vec::new(), + active_leaf_id: 4, + viewing_branch_id: 2, + fork_info: vec![Some(crate::ai::handlers::ChatForkInfo { + position: 1, + total: 2, + node_id: 3, + })], }; let json = serde_json::to_value(&view).unwrap(); // The client's shared ChatHistoryView reads snake_case keys. + assert_eq!(json["conversation_id"], "conv-1"); + assert_eq!(json["persona_id"], "journal"); + assert_eq!(json["title"], "June recap"); assert_eq!(json["turn_count"], 1); - assert_eq!(json["active_leaf_id"], 0); - assert_eq!(json["viewing_branch_id"], 0); + assert_eq!(json["active_leaf_id"], 4); + assert_eq!(json["viewing_branch_id"], 2); assert_eq!(json["messages"][0]["is_initial"], true); assert!(json["messages"][0].get("tools").is_none()); + // fork_info matches the file chat's shape so the client's shared + // branch-picker code can read either surface. + assert_eq!(json["fork_info"][0]["position"], 1); + assert_eq!(json["fork_info"][0]["total"], 2); + assert_eq!(json["fork_info"][0]["node_id"], 3); } #[test] - fn persona_chat_reset_request_deserializes_snake_case_wire_body() { - let req: PersonaChatResetRequest = + fn persona_chat_create_request_deserializes_snake_case_wire_body() { + let req: PersonaChatCreateRequest = serde_json::from_str(r#"{"persona_id": "journal"}"#).unwrap(); assert_eq!(req.persona_id, "journal"); } -} \ No newline at end of file + + /// system → user → assistant, plus a second assistant forking off the + /// user turn. Returns (store, rendered node ids on the active path). + fn forked_store() -> (ChatHistoryStore, Vec) { + let mut store = ChatHistoryStore::from_flat_array(vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("q".to_string()), + ]); + let user_node = store.active_leaf_id; + let _a1 = store.append_node(Some(user_node), assistant_msg("first answer")); + let a2 = store.append_node(Some(user_node), assistant_msg("second answer")); + store.active_leaf_id = a2; + let path = store.path_to_leaf(a2).unwrap(); + let (_r, _t, node_ids, _f) = render_persona_path(&store, &path); + (store, node_ids) + } + + #[test] + fn rewind_target_leaf_keeps_everything_before_the_discarded_index() { + let (store, node_ids) = forked_store(); + // Discard the reply (index 1) — the user turn at index 0 stays, and + // the reply survives as a fork rather than being deleted. + let leaf = rewind_target_leaf(&store, &node_ids, 1).unwrap(); + assert_eq!(leaf, node_ids[0]); + } + + #[test] + fn rewind_target_leaf_at_zero_reanchors_on_the_seed_node() { + // The file chat refuses index 0 because its index 0 is a synthetic + // prompt. Here it is the user's own question, so editing it has to + // work — the anchor becomes the system node above it. + let (store, node_ids) = forked_store(); + let leaf = rewind_target_leaf(&store, &node_ids, 0).unwrap(); + assert_eq!(leaf, store.nodes[0].id, "anchors on the seed system node"); + assert_eq!(store.nodes[0].message.role, "system"); + } + + #[test] + fn rewind_target_leaf_rejects_an_index_past_the_transcript() { + let (store, node_ids) = forked_store(); + let err = rewind_target_leaf(&store, &node_ids, 99).unwrap_err(); + assert!(err.to_string().contains("out of range")); + } + + #[test] + fn rewind_target_leaf_errors_when_nothing_sits_above_index_zero() { + // A transcript with no seed system node has nothing to re-anchor on. + let store = ChatHistoryStore::from_flat_array(vec![ChatMessage::user("q".to_string())]); + let path = store.path_to_leaf(store.active_leaf_id).unwrap(); + let (_r, _t, node_ids, _f) = render_persona_path(&store, &path); + let err = rewind_target_leaf(&store, &node_ids, 0).unwrap_err(); + assert!(err.to_string().contains("cannot rewind past the start")); + } + + #[test] + fn rewind_target_leaf_rejects_an_empty_transcript() { + let store = ChatHistoryStore::from_flat_array(Vec::new()); + let err = rewind_target_leaf(&store, &[], 0).unwrap_err(); + assert!(err.to_string().contains("out of range")); + } + + #[test] + fn sanitize_title_strips_the_quotes_small_models_add() { + assert_eq!(sanitize_title("\"June recap\""), "June recap"); + assert_eq!(sanitize_title("'June recap'"), "June recap"); + assert_eq!(sanitize_title("“June recap”"), "June recap"); + assert_eq!(sanitize_title("**June recap**"), "June recap"); + } + + #[test] + fn sanitize_title_strips_a_label_prefix_and_trailing_period() { + assert_eq!(sanitize_title("Title: June recap."), "June recap"); + assert_eq!(sanitize_title("Chat title: June recap"), "June recap"); + assert_eq!( + sanitize_title("CONVERSATION TITLE: June recap"), + "June recap" + ); + } + + #[test] + fn sanitize_title_takes_the_first_line_of_a_chatty_reply() { + // Small models like to explain themselves after answering. + assert_eq!( + sanitize_title("June recap\n\nI chose this because it summarizes…"), + "June recap" + ); + } + + #[test] + fn sanitize_title_collapses_whitespace() { + assert_eq!(sanitize_title(" June recap \t "), "June recap"); + } + + #[test] + fn sanitize_title_truncates_on_a_character_boundary() { + // Byte-index slicing here would panic on the first accented word. + let long = "é".repeat(TITLE_MAX_CHARS + 20); + let title = sanitize_title(&long); + assert_eq!(title.chars().count(), TITLE_MAX_CHARS + 1); + assert!(title.ends_with('…')); + } + + #[test] + fn sanitize_title_returns_empty_for_junk() { + assert_eq!(sanitize_title(" "), ""); + assert_eq!(sanitize_title("\"\""), ""); + } + + #[test] + fn fallback_title_names_a_conversation_after_the_opening_question() { + // Better than "Untitled": it is the thing the user scans for. + let messages = vec![ + ChatMessage::system("sys".to_string()), + ChatMessage::user("What happened in June?".to_string()), + assistant_msg("Quite a lot."), + ]; + assert_eq!(fallback_title(&messages), "What happened in June?"); + } + + #[test] + fn fallback_title_is_empty_without_a_user_message() { + let messages = vec![ChatMessage::system("sys".to_string())]; + assert_eq!(fallback_title(&messages), ""); + } + + #[test] + fn title_prompt_carries_both_sides_of_the_opening_exchange() { + let prompt = title_prompt("What happened in June?", "Quite a lot."); + assert!(prompt.contains("What happened in June?")); + assert!(prompt.contains("Quite a lot.")); + // The instruction the sanitizer is the backstop for. + assert!(prompt.contains("at most six")); + } + + #[test] + fn conversation_snippet_flattens_whitespace_to_one_line() { + // A markdown reply must not blow up the row height on the list. + let snippet = conversation_snippet("first line\n\n- bullet\n- another"); + assert_eq!(snippet, "first line - bullet - another"); + } + + #[test] + fn conversation_snippet_leaves_a_short_message_untouched() { + assert_eq!(conversation_snippet("hello there"), "hello there"); + } + + #[test] + fn conversation_snippet_truncates_on_a_character_boundary() { + // Multi-byte characters: a byte-index slice here would panic. + let content = "é".repeat(SNIPPET_MAX_CHARS + 50); + let snippet = conversation_snippet(&content); + assert_eq!(snippet.chars().count(), SNIPPET_MAX_CHARS + 1); + assert!(snippet.ends_with('…')); + } + + #[test] + fn conversation_snippet_handles_an_empty_message() { + assert_eq!(conversation_snippet(" "), ""); + } + + #[test] + fn persona_conversation_summary_serializes_snake_case_wire_shape() { + let summary = PersonaConversationSummary { + conversation_id: "conv-1".to_string(), + persona_id: "journal".to_string(), + persona_name: "Journal".to_string(), + title: "June recap".to_string(), + snippet: "last thing said".to_string(), + snippet_role: "assistant".to_string(), + turn_count: 3, + message_count: 6, + created_at: 1_600_000_000_000, + updated_at: 1_700_000_000_000, + has_branches: true, + }; + let json = serde_json::to_value(&summary).unwrap(); + assert_eq!(json["conversation_id"], "conv-1"); + assert_eq!(json["title"], "June recap"); + assert_eq!(json["persona_id"], "journal"); + assert_eq!(json["persona_name"], "Journal"); + assert_eq!(json["snippet_role"], "assistant"); + assert_eq!(json["turn_count"], 3); + assert_eq!(json["message_count"], 6); + assert_eq!(json["updated_at"], 1_700_000_000_000i64); + assert_eq!(json["has_branches"], true); + } + + #[test] + fn persona_chat_rewind_request_deserializes_snake_case_wire_body() { + let req: PersonaChatRewindRequest = serde_json::from_str( + r#"{"conversation_id": "conv-1", "discard_from_rendered_index": 3}"#, + ) + .unwrap(); + assert_eq!(req.conversation_id, "conv-1"); + assert_eq!(req.discard_from_rendered_index, 3); + } + + #[test] + fn persona_chat_switch_branch_request_deserializes_snake_case_wire_body() { + let req: PersonaChatSwitchBranchRequest = + serde_json::from_str(r#"{"conversation_id": "conv-1", "branch_id": 9}"#).unwrap(); + assert_eq!(req.conversation_id, "conv-1"); + assert_eq!(req.branch_id, 9); + } + + #[test] + fn persona_chat_branches_query_defaults_the_optional_scoping_params() { + let q: PersonaChatBranchesQuery = + serde_json::from_str(r#"{"conversation_id": "conv-1"}"#).unwrap(); + assert_eq!(q.node_id, None, "unscoped listing walks every leaf"); + assert_eq!(q.viewing_branch_id, None); + + let q: PersonaChatBranchesQuery = serde_json::from_str( + r#"{"conversation_id": "conv-1", "node_id": 4, "viewing_branch_id": 6}"#, + ) + .unwrap(); + assert_eq!(q.node_id, Some(4)); + assert_eq!(q.viewing_branch_id, Some(6)); + } + + #[test] + fn branch_options_at_ranks_the_siblings_of_a_divergence() { + // The list the branch picker renders: both replies to the same user + // turn, position-ranked, with the viewed one anchored. + let (store, _node_ids) = forked_store(); + let user_node = store.nodes[1].id; + let options = store.branch_options_at(user_node, store.active_leaf_id); + assert_eq!(options.len(), 2); + assert_eq!(options[0].position, Some(1)); + assert_eq!(options[1].position, Some(2)); + assert!( + options.iter().any(|o| o.id == store.active_leaf_id), + "the branch being viewed is one of the options" + ); + } + + #[test] + fn persona_chat_history_query_accepts_an_optional_branch_id() { + let q: PersonaChatHistoryQuery = + serde_json::from_str(r#"{"conversation_id": "conv-1"}"#).unwrap(); + assert_eq!( + q.branch_id, None, + "omitting branch_id renders the active branch" + ); + + let q: PersonaChatHistoryQuery = + serde_json::from_str(r#"{"conversation_id": "conv-1", "branch_id": 12}"#).unwrap(); + assert_eq!(q.branch_id, Some(12)); + } +} diff --git a/src/database/persona_dao.rs b/src/database/persona_dao.rs index af751a6..f4a2939 100644 --- a/src/database/persona_dao.rs +++ b/src/database/persona_dao.rs @@ -13,16 +13,48 @@ use crate::otel::trace_db_call; /// One row of the persona-chat transcript. Lives in /// `persona_chat_conversations` (one row per `(user_id, persona_id)`). /// -/// `turn_count` is the number of new turn-rows since the last persisted -/// slice — used by the SSE `done` event for stats dashboards. The actual -/// transcript is the `messages_json` blob. +/// One persona conversation. +/// +/// `turn_count` is the cumulative number of assistant turns on the active +/// branch — what the list screen shows as the length of a conversation. The +/// transcript itself is the `messages_json` blob, holding a serialized +/// `ChatHistoryStore` tree. `title` is a generated summary of the opening +/// exchange, empty until the first turn completes. #[derive(Clone, Debug)] pub struct PersonaChatRow { + pub conversation_id: String, + pub persona_id: String, + pub title: String, pub messages_json: String, pub turn_count: i32, + pub created_at: i64, pub updated_at: i64, } +/// Column tuple → `PersonaChatRow`. Shared by the single-row and list +/// queries so their `select(...)` orders can never drift apart. +fn persona_chat_row( + (conversation_id, persona_id, title, messages_json, turn_count, created_at, updated_at): ( + String, + String, + String, + String, + i32, + i64, + i64, + ), +) -> PersonaChatRow { + PersonaChatRow { + conversation_id, + persona_id, + title, + messages_json, + turn_count, + created_at, + updated_at, + } +} + /// Patch shape for update_persona. None = leave field alone. Built-ins are /// allowed to flip `include_all_memories` but should reject name/prompt /// edits at the handler layer (built-in copy lives in the migration). @@ -96,39 +128,72 @@ pub trait PersonaDao: Sync + Send { // ── Persona-chat (open chat with persona) persistence ─────────── // - // Keyed by `(user_id, persona_id)` with a single rolling transcript - // per pair. No tree branching in v1 — matches the locked-in scope. + // Keyed on an opaque `conversation_id` so one persona can hold several + // separate conversations. `user_id` is checked on every read and write: + // the id is a bearer token for a transcript, so ownership can never be + // assumed from the id alone. - /// Fetch the rolling transcript for one `(user, persona)`. None when - /// the user has never started a conversation with this persona. + /// Fetch one conversation. None when it doesn't exist or belongs to + /// another user — the two are deliberately indistinguishable to callers. fn get_persona_chat( &mut self, cx: &opentelemetry::Context, user_id: i32, - persona_id: &str, + conversation_id: &str, ) -> Result, DbError>; - /// Upsert (create-or-replace) the rolling transcript. Called once per - /// completed turn with the full new `messages_json`. The `turn_count` - /// is the number of user/assistant pairs added in this write. - fn upsert_persona_chat( + /// Every conversation this user has going, newest first. Backs the chat + /// list screen, so it returns the transcript blob too — the snippet is + /// derived from the tree rather than denormalized into its own column. + fn list_persona_chats( + &mut self, + cx: &opentelemetry::Context, + user_id: i32, + ) -> Result, DbError>; + + /// Start a new conversation with a persona, returning its id. Several + /// conversations with the same persona are expected, so this never + /// reuses an existing row. + fn create_persona_chat( &mut self, cx: &opentelemetry::Context, user_id: i32, persona_id: &str, + created_at: i64, + ) -> Result; + + /// Replace a conversation's transcript. Called once per completed turn + /// with the full new `messages_json`; `turn_count` is the cumulative + /// assistant-turn count on the resulting active branch. Returns the + /// number of rows written — 0 means the conversation is gone or is not + /// this user's. + fn update_persona_chat( + &mut self, + cx: &opentelemetry::Context, + user_id: i32, + conversation_id: &str, messages_json: &str, turn_count: i32, updated_at: i64, - ) -> Result<(), DbError>; + ) -> Result; - /// Wipe the rolling transcript. Reserved for a future "New - /// conversation" affordance; the row itself stays (so a subsequent - /// get returns None, not 404). - fn clear_persona_chat( + /// Name a conversation. Written once, after the first turn produces + /// enough of an exchange to summarize. + fn set_persona_chat_title( &mut self, cx: &opentelemetry::Context, user_id: i32, - persona_id: &str, + conversation_id: &str, + title: &str, + ) -> Result<(), DbError>; + + /// Delete a conversation outright. Backs both the list screen's delete + /// affordance and the reset endpoint. + fn delete_persona_chat( + &mut self, + cx: &opentelemetry::Context, + user_id: i32, + conversation_id: &str, ) -> Result<(), DbError>; } @@ -351,84 +416,155 @@ impl PersonaDao for SqlitePersonaDao { &mut self, cx: &opentelemetry::Context, uid: i32, - pid: &str, + cid: &str, ) -> Result, DbError> { trace_db_call(cx, "query", "get_persona_chat", |_span| { use schema::persona_chat_conversations::dsl::*; let mut conn = self.connection.lock().expect("PersonaDao lock"); persona_chat_conversations + .filter(conversation_id.eq(cid)) + // Scoped by user as well as id: the id is a bearer token for + // someone's transcript, so it never grants access on its own. .filter(user_id.eq(uid)) - .filter(persona_id.eq(pid)) - .select((messages_json, turn_count, updated_at)) - .first::<(String, i32, i64)>(conn.deref_mut()) + .select(( + conversation_id, + persona_id, + title, + messages_json, + turn_count, + created_at, + updated_at, + )) + .first::<(String, String, String, String, i32, i64, i64)>(conn.deref_mut()) .optional() - .map(|opt| { - opt.map(|(m, t, u)| PersonaChatRow { - messages_json: m, - turn_count: t, - updated_at: u, - }) - }) + .map(|opt| opt.map(persona_chat_row)) .map_err(|e| anyhow::anyhow!("Query error: {}", e)) }) .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) } - fn upsert_persona_chat( + fn list_persona_chats( + &mut self, + cx: &opentelemetry::Context, + uid: i32, + ) -> Result, DbError> { + trace_db_call(cx, "query", "list_persona_chats", |_span| { + use schema::persona_chat_conversations::dsl::*; + let mut conn = self.connection.lock().expect("PersonaDao lock"); + persona_chat_conversations + .filter(user_id.eq(uid)) + .order(updated_at.desc()) + .select(( + conversation_id, + persona_id, + title, + messages_json, + turn_count, + created_at, + updated_at, + )) + .load::<(String, String, String, String, i32, i64, i64)>(conn.deref_mut()) + .map(|rows| rows.into_iter().map(persona_chat_row).collect()) + .map_err(|e| anyhow::anyhow!("Query error: {}", e)) + }) + .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) + } + + fn create_persona_chat( &mut self, cx: &opentelemetry::Context, uid: i32, pid: &str, - json: &str, - count: i32, - updated_at: i64, - ) -> Result<(), DbError> { - trace_db_call(cx, "upsert", "upsert_persona_chat", |_span| { + created: i64, + ) -> Result { + trace_db_call(cx, "insert", "create_persona_chat", |_span| { + use schema::persona_chat_conversations::dsl::*; let mut conn = self.connection.lock().expect("PersonaDao lock"); - // INSERT OR REPLACE on the (user_id, persona_id) PRIMARY KEY — - // single rolling transcript, so a new write always supersedes - // the prior one in full. The mobile hook serialises turns with - // a per-persona mutex, so this never races. Plain - // `sql_query` sidesteps a Diesel type-recursion blow-up that - // hits `insert_into(...).on_conflict(...).do_update().set(...)` - // with this many typed columns. - diesel::sql_query( - "INSERT INTO persona_chat_conversations \ - (user_id, persona_id, messages_json, turn_count, updated_at) \ - VALUES (?, ?, ?, ?, ?) \ - ON CONFLICT(user_id, persona_id) DO UPDATE SET \ - messages_json = excluded.messages_json, \ - turn_count = excluded.turn_count, \ - updated_at = excluded.updated_at", - ) - .bind::(uid) - .bind::(pid) - .bind::(json) - .bind::(count) - .bind::(updated_at) - .execute(conn.deref_mut()) - .map_err(|e| anyhow::anyhow!("Upsert error: {}", e))?; - Ok(()) + let new_id = uuid::Uuid::new_v4().to_string(); + diesel::insert_into(persona_chat_conversations) + .values(( + conversation_id.eq(&new_id), + user_id.eq(uid), + persona_id.eq(pid), + title.eq(""), + // The empty tree; the first turn seeds it with the + // persona's system prompt. + messages_json.eq("[]"), + turn_count.eq(0), + created_at.eq(created), + updated_at.eq(created), + )) + .execute(conn.deref_mut()) + .map_err(|e| anyhow::anyhow!("Insert error: {}", e))?; + Ok(new_id) }) .map_err(|e| DbError::log(DbErrorKind::InsertError, e)) } - fn clear_persona_chat( + fn update_persona_chat( &mut self, cx: &opentelemetry::Context, uid: i32, - pid: &str, - ) -> Result<(), DbError> { - trace_db_call(cx, "delete", "clear_persona_chat", |_span| { + cid: &str, + json: &str, + count: i32, + updated: i64, + ) -> Result { + trace_db_call(cx, "update", "update_persona_chat", |_span| { + use schema::persona_chat_conversations::dsl::*; + let mut conn = self.connection.lock().expect("PersonaDao lock"); + diesel::update( + persona_chat_conversations + .filter(conversation_id.eq(cid)) + .filter(user_id.eq(uid)), + ) + .set(( + messages_json.eq(json), + turn_count.eq(count), + updated_at.eq(updated), + )) + .execute(conn.deref_mut()) + .map_err(|e| anyhow::anyhow!("Update error: {}", e)) + }) + .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) + } + + fn set_persona_chat_title( + &mut self, + cx: &opentelemetry::Context, + uid: i32, + cid: &str, + new_title: &str, + ) -> Result<(), DbError> { + trace_db_call(cx, "update", "set_persona_chat_title", |_span| { + use schema::persona_chat_conversations::dsl::*; + let mut conn = self.connection.lock().expect("PersonaDao lock"); + diesel::update( + persona_chat_conversations + .filter(conversation_id.eq(cid)) + .filter(user_id.eq(uid)), + ) + .set(title.eq(new_title)) + .execute(conn.deref_mut()) + .map_err(|e| anyhow::anyhow!("Update error: {}", e))?; + Ok(()) + }) + .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) + } + + fn delete_persona_chat( + &mut self, + cx: &opentelemetry::Context, + uid: i32, + cid: &str, + ) -> Result<(), DbError> { + trace_db_call(cx, "delete", "delete_persona_chat", |_span| { use schema::persona_chat_conversations::dsl::*; let mut conn = self.connection.lock().expect("PersonaDao lock"); - // Delete the row entirely so the next get returns None. - // The next send will INSERT a fresh seed (system + greeting) - // via the same code path a brand-new persona uses. diesel::delete( persona_chat_conversations - .filter(user_id.eq(uid)) - .filter(persona_id.eq(pid)), + .filter(conversation_id.eq(cid)) + .filter(user_id.eq(uid)), ) .execute(conn.deref_mut()) .map_err(|e| anyhow::anyhow!("Delete error: {}", e))?; @@ -437,7 +573,6 @@ impl PersonaDao for SqlitePersonaDao { .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) } } - #[cfg(test)] mod tests { use super::*; @@ -587,74 +722,170 @@ mod tests { // ── Persona-chat DAO tests ───────────────────────────────────── - #[test] - fn persona_chat_get_returns_none_for_never_started() { - let cx = opentelemetry::Context::new(); - let (mut dao, uid) = dao_with_user("p1"); - let row = dao.get_persona_chat(&cx, uid, "default").unwrap(); - assert!(row.is_none()); + /// Second user, for the isolation tests. + fn second_user(dao: &SqlitePersonaDao, username: &str) -> i32 { + use crate::database::schema::users::dsl as u; + let conn = dao.connection.clone(); + diesel::insert_into(u::users) + .values((u::username.eq(username), u::password.eq("x"))) + .execute(conn.lock().unwrap().deref_mut()) + .unwrap(); + u::users + .filter(u::username.eq(username)) + .select(u::id) + .first(conn.lock().unwrap().deref_mut()) + .unwrap() } #[test] - fn persona_chat_upsert_then_get_round_trip() { + fn persona_chat_get_returns_none_for_an_unknown_conversation() { + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p1"); + assert!( + dao.get_persona_chat(&cx, uid, "no-such-id") + .unwrap() + .is_none() + ); + } + + #[test] + fn persona_chat_create_then_get_round_trip() { let cx = opentelemetry::Context::new(); let (mut dao, uid) = dao_with_user("p2"); - dao.upsert_persona_chat(&cx, uid, "journal", "[]", 1, 100) - .unwrap(); - let row = dao.get_persona_chat(&cx, uid, "journal").unwrap().unwrap(); - assert_eq!(row.messages_json, "[]"); - assert_eq!(row.turn_count, 1); + let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + + let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap(); + assert_eq!(row.conversation_id, cid); + assert_eq!(row.persona_id, "journal"); + assert_eq!(row.title, "", "unnamed until the first turn completes"); + assert_eq!(row.turn_count, 0); + assert_eq!(row.created_at, 100); assert_eq!(row.updated_at, 100); } #[test] - fn persona_chat_upsert_replaces_existing_row() { + fn persona_chat_create_makes_a_separate_row_each_time() { + // The whole point of the conversation_id key: one persona, several + // independent threads. let cx = opentelemetry::Context::new(); let (mut dao, uid) = dao_with_user("p3"); - dao.upsert_persona_chat(&cx, uid, "journal", "first", 1, 100).unwrap(); - dao.upsert_persona_chat(&cx, uid, "journal", "second", 2, 200).unwrap(); - let row = dao.get_persona_chat(&cx, uid, "journal").unwrap().unwrap(); + let first = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + let second = dao.create_persona_chat(&cx, uid, "journal", 200).unwrap(); + + assert_ne!(first, second); + assert_eq!(dao.list_persona_chats(&cx, uid).unwrap().len(), 2); + } + + #[test] + fn persona_chat_update_replaces_the_transcript() { + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p4"); + let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + + let rows = dao + .update_persona_chat(&cx, uid, &cid, "first", 1, 150) + .unwrap(); + assert_eq!(rows, 1); + let rows = dao + .update_persona_chat(&cx, uid, &cid, "second", 2, 200) + .unwrap(); + assert_eq!(rows, 1); + + let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap(); assert_eq!(row.messages_json, "second"); assert_eq!(row.turn_count, 2); assert_eq!(row.updated_at, 200); - // Single rolling transcript → exactly one row per (user, persona). + assert_eq!(row.created_at, 100, "creation time is not disturbed"); } #[test] - fn persona_chat_isolation_between_users() { + fn persona_chat_update_reports_zero_rows_for_an_unknown_conversation() { + // The turn loop treats 0 as "the conversation went away mid-turn" + // rather than silently succeeding. let cx = opentelemetry::Context::new(); - let (mut dao, uid1) = dao_with_user("u1"); - let uid2: i32 = { - let conn = dao.connection.clone(); - use crate::database::schema::users::dsl as u; - diesel::insert_into(u::users) - .values((u::username.eq("u2"), u::password.eq("x"))) - .execute(conn.lock().unwrap().deref_mut()) - .unwrap(); - u::users - .filter(u::username.eq("u2")) - .select(u::id) - .first(conn.lock().unwrap().deref_mut()) - .unwrap() - }; - dao.upsert_persona_chat(&cx, uid1, "default", "u1-row", 1, 1).unwrap(); - dao.upsert_persona_chat(&cx, uid2, "default", "u2-row", 1, 2).unwrap(); + let (mut dao, uid) = dao_with_user("p5"); + let rows = dao + .update_persona_chat(&cx, uid, "no-such-id", "x", 1, 1) + .unwrap(); + assert_eq!(rows, 0); + } + + #[test] + fn persona_chat_set_title_names_the_conversation() { + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p6"); + let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + + dao.set_persona_chat_title(&cx, uid, &cid, "June recap") + .unwrap(); + let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap(); + assert_eq!(row.title, "June recap"); + } + + #[test] + fn persona_chat_list_orders_newest_first() { + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p7"); + let older = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + let newer = dao.create_persona_chat(&cx, uid, "coach", 300).unwrap(); + let middle = dao.create_persona_chat(&cx, uid, "default", 200).unwrap(); + + let ids: Vec = dao + .list_persona_chats(&cx, uid) + .unwrap() + .into_iter() + .map(|r| r.conversation_id) + .collect(); assert_eq!( - dao.get_persona_chat(&cx, uid1, "default").unwrap().unwrap().messages_json, - "u1-row" - ); - assert_eq!( - dao.get_persona_chat(&cx, uid2, "default").unwrap().unwrap().messages_json, - "u2-row" + ids, + vec![newer, middle, older], + "ordered by updated_at descending so the list screen needs no re-sort" ); } #[test] - fn persona_chat_clear_wipes_the_row_so_get_returns_none() { + fn persona_chat_list_is_empty_for_a_user_who_has_never_chatted() { let cx = opentelemetry::Context::new(); - let (mut dao, uid) = dao_with_user("p4"); - dao.upsert_persona_chat(&cx, uid, "default", "[]", 1, 1).unwrap(); - dao.clear_persona_chat(&cx, uid, "default").unwrap(); - assert!(dao.get_persona_chat(&cx, uid, "default").unwrap().is_none()); + let (mut dao, uid) = dao_with_user("p8"); + assert!(dao.list_persona_chats(&cx, uid).unwrap().is_empty()); + } + + #[test] + fn persona_chat_delete_removes_the_conversation() { + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p9"); + let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap(); + + dao.delete_persona_chat(&cx, uid, &cid).unwrap(); + assert!(dao.get_persona_chat(&cx, uid, &cid).unwrap().is_none()); + assert!(dao.list_persona_chats(&cx, uid).unwrap().is_empty()); + } + + #[test] + fn persona_chat_reads_and_writes_are_scoped_to_the_owner() { + // A conversation_id is a bearer token for someone's transcript, so + // holding one must not grant another user access to it. + let cx = opentelemetry::Context::new(); + let (mut dao, uid1) = dao_with_user("owner"); + let uid2 = second_user(&dao, "intruder"); + let cid = dao.create_persona_chat(&cx, uid1, "journal", 100).unwrap(); + dao.update_persona_chat(&cx, uid1, &cid, "private", 1, 100) + .unwrap(); + + assert!( + dao.get_persona_chat(&cx, uid2, &cid).unwrap().is_none(), + "another user cannot read it" + ); + assert_eq!( + dao.update_persona_chat(&cx, uid2, &cid, "tampered", 9, 999) + .unwrap(), + 0, + "another user cannot write it" + ); + dao.delete_persona_chat(&cx, uid2, &cid).unwrap(); + + let row = dao.get_persona_chat(&cx, uid1, &cid).unwrap().unwrap(); + assert_eq!(row.messages_json, "private", "and cannot delete it"); + assert!(dao.list_persona_chats(&cx, uid2).unwrap().is_empty()); } } diff --git a/src/database/schema.rs b/src/database/schema.rs index 8eeee1c..a0d053e 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -172,11 +172,14 @@ diesel::table! { } diesel::table! { - persona_chat_conversations (user_id, persona_id) { + persona_chat_conversations (conversation_id) { + conversation_id -> Text, user_id -> Integer, persona_id -> Text, + title -> Text, messages_json -> Text, turn_count -> Integer, + created_at -> BigInt, updated_at -> BigInt, } } diff --git a/src/main.rs b/src/main.rs index dbd633c..4ad8714 100644 --- a/src/main.rs +++ b/src/main.rs @@ -384,9 +384,14 @@ fn main() -> std::io::Result<()> { .service(ai::cancel_turn_handler) .service(ai::persona_chat_history_handler) .service(ai::persona_chat_turn_handler) - .service(ai::persona_chat_reset_handler) + .service(ai::persona_chat_create_conversation_handler) + .service(ai::persona_chat_delete_conversation_handler) .service(ai::persona_turn_replay_handler) .service(ai::persona_turn_cancel_handler) + .service(ai::persona_chat_rewind_handler) + .service(ai::persona_chat_switch_branch_handler) + .service(ai::persona_chat_branches_handler) + .service(ai::persona_chat_conversations_handler) .service(ai::rate_insight_handler) .service(ai::export_training_data_handler) .service(ai::tts_speech_handler) diff --git a/src/thumbnails.rs b/src/thumbnails.rs index 75a456f..c5ade67 100644 --- a/src/thumbnails.rs +++ b/src/thumbnails.rs @@ -175,11 +175,7 @@ fn encode_large_jpeg(img: image::DynamicImage, dest: &Path) -> std::io::Result<( /// ffmpeg path for HEIC/HEIF (image crate can't decode these). Mirrors /// [`crate::video::actors::generate_image_thumbnail_ffmpeg`] but scales /// to the large-preview cap instead of 200. -fn generate_large_preview_ffmpeg( - src: &Path, - dest: &Path, - orientation: i32, -) -> std::io::Result<()> { +fn generate_large_preview_ffmpeg(src: &Path, dest: &Path, orientation: i32) -> std::io::Result<()> { // Rotation + scale + colorspace. HEIC sources use Display P3; without // colorspace=bt709 the mjpeg encoder treats P3 values as sRGB, producing // warm/oversaturated output. The min(iw,cap) trick caps the long edge