From 65cceaa67c1574c12f92640703cd8d0f73fd72dd Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Tue, 25 Aug 2026 18:23:10 -0400 Subject: [PATCH] feat: multiple persona conversations with branching and generated titles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Persona chat stored a flat Vec keyed on (user_id, persona_id), which meant one rolling transcript per persona and no way to revisit a turn. This moves it onto the same ChatHistoryStore tree the file chat uses and gives conversations their own identity. Storage - messages_json now holds a serialized ChatHistoryStore. Reads accept the old flat array and upgrade it in place, so existing transcripts survive without a data migration. The upgrade drops the v1 seed greeting, which the flat renderer hid but the tree renderer would surface as a bubble the user has never seen. - New migration re-keys persona_chat_conversations on an opaque conversation_id and adds title + created_at, so one persona can hold any number of separate threads. Every DAO read and write is scoped by user_id as well: a conversation id is a bearer token for someone's transcript and must never grant access on its own. - The per-conversation lock and in-flight turn slot key on conversation_id, so two threads with the same persona can run turns concurrently. Endpoints - POST/DELETE /persona_chat/conversations — start and remove a thread, replacing /persona_chat/reset. - GET /persona_chat/conversations — the chat list, with snippet and counts derived from each tree's active branch. - POST /persona_chat/rewind, POST /persona_chat/switch-branch, GET /persona_chat/branches — rewind and fork, mirroring the file chat. Index 0 is rewindable here (it is the user's own first question, not a synthetic prompt) and re-anchors on the seed node. - history/turn/rewind/switch-branch/branches all key on conversation_id; history gained branch_id and now returns real fork_info, active_leaf_id and viewing_branch_id instead of placeholders. The turn body no longer carries a persona at all — it is read from the stored conversation, so a stale client cannot swap a thread's voice midway. Titles After the first turn persists, the conversation is named from its opening exchange on the same backend the turn ran on. Small models wrap titles in quotes, prefix them with "Title:" and append explanations, so sanitize_title strips all of that and truncates on a character boundary. Any failure falls back to the user's opening question. Generation runs after persistence: a failed title must not cost the turn. Fixes found along the way - insight_chat: both file-chat turn paths captured path.len() before apply_context_budget drained messages out of the middle, then sliced messages[path_len..] for the new tree nodes. Once truncation fired that dropped the user turn from the tree or panicked on an out-of-range start index. Now read after the budget pass as history_len. - Persona chat had no context budget at all and hardcoded truncated: false in the done frame, so a rolling transcript grew unbounded. - A cancelled turn persisted a half-finished transcript and pushed a second terminal frame; it now returns early like the file chat. - The seeded system prompt was frozen at conversation creation, so editing a persona never reached a thread already in flight. Re-resolved per turn. - turn_count was overwritten each write with the per-turn message delta; it is now the cumulative assistant-turn count on the active branch. - is_initial is always false: the file chat reserves it for its synthetic "describe this photo" prompt, and marking a persona chat's first question with it made the opening reply impossible to regenerate. 600 lib tests pass, clippy --all-targets clean. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 2 +- Cargo.toml | 2 +- .../down.sql | 32 + .../up.sql | 50 + src/ai/handlers.rs | 2 +- src/ai/insight_chat.rs | 25 +- src/ai/llm_client.rs | 4 +- src/ai/mod.rs | 12 +- src/ai/persona_chat.rs | 1697 ++++++++++++++--- src/database/persona_dao.rs | 459 +++-- src/database/schema.rs | 5 +- src/main.rs | 7 +- src/thumbnails.rs | 6 +- 13 files changed, 1848 insertions(+), 455 deletions(-) create mode 100644 migrations/2026-08-25-000000_persona_chat_multi/down.sql create mode 100644 migrations/2026-08-25-000000_persona_chat_multi/up.sql 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