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-21-000000_persona_chat/down.sql b/migrations/2026-08-21-000000_persona_chat/down.sql new file mode 100644 index 0000000..89964c8 --- /dev/null +++ b/migrations/2026-08-21-000000_persona_chat/down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS persona_chat_conversations; \ No newline at end of file diff --git a/migrations/2026-08-21-000000_persona_chat/up.sql b/migrations/2026-08-21-000000_persona_chat/up.sql new file mode 100644 index 0000000..d35cd83 --- /dev/null +++ b/migrations/2026-08-21-000000_persona_chat/up.sql @@ -0,0 +1,22 @@ +-- Open chat with persona — file-anchored insight chat is keyed by +-- (library_id, file_path). The open chat is keyed by (user_id, persona_id) +-- with a single rolling transcript per pair. No tree branching in v1 +-- (matches the locked-in "single rolling conversation per persona" scope); +-- a flat JSON blob is enough and avoids forcing a tree shape onto a +-- surface that's intentionally linear. +-- +-- `messages_json` is the same `Vec` shape the file-anchored +-- chat persists, so the SSE replay / UI rendering on the mobile client +-- can reuse the same parser without a second schema. + +CREATE TABLE persona_chat_conversations ( + 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) +); + +CREATE INDEX idx_persona_chat_updated + ON persona_chat_conversations (user_id, updated_at DESC); \ No newline at end of file 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 022588f..1a98d35 100644 --- a/src/ai/handlers.rs +++ b/src/ai/handlers.rs @@ -1819,6 +1819,19 @@ pub async fn turn_replay_handler( path: web::Path, query: web::Query, app_state: web::Data, +) -> HttpResponse { + turn_replay_impl(http_request, path, query, app_state).await +} + +/// Core of the SSE replay, kept attribute-free so the persona-chat routes +/// can reuse it under `/persona_chat/turn/{turn_id}`. The registry is keyed +/// on `turn_id` only, and `render_turn_info_frame` already scopes the +/// identity (persona turns carry `persona_id` and no `file_path`). +pub(crate) async fn turn_replay_impl( + http_request: HttpRequest, + path: web::Path, + query: web::Query, + app_state: web::Data, ) -> HttpResponse { use crate::ai::turn_registry::ReplayOutcome; @@ -1949,15 +1962,31 @@ pub async fn turn_replay_handler( } fn render_turn_info_frame(info: &crate::ai::turn_registry::TurnInfo) -> String { - let payload = serde_json::json!({ - "turn_id": info.turn_id, - "file_path": info.file_path, - "library_id": info.library_id, - "status": info.status.as_str(), - "total_events_pushed": info.total_events_pushed, - "buffered_count": info.buffered_count, - }); - let data = serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string()); + // Persona-scoped turns leak no `file_path` to the client — the open + // chat is keyed on `(persona_id)` only and a stray path echo from a + // quoted SMS would be a privacy regression. Insight-scoped turns + // include the path as before. + let mut payload = serde_json::Map::new(); + payload.insert("turn_id".into(), serde_json::json!(info.turn_id)); + payload.insert("library_id".into(), serde_json::json!(info.library_id)); + payload.insert("scope".into(), serde_json::json!(info.scope)); + payload.insert("status".into(), serde_json::json!(info.status.as_str())); + payload.insert( + "total_events_pushed".into(), + serde_json::json!(info.total_events_pushed), + ); + payload.insert( + "buffered_count".into(), + serde_json::json!(info.buffered_count), + ); + if info.scope == "insight" { + payload.insert("file_path".into(), serde_json::json!(info.file_path)); + } + if let Some(ref pid) = info.persona_id { + payload.insert("persona_id".into(), serde_json::json!(pid)); + } + let data = serde_json::to_string(&serde_json::Value::Object(payload)) + .unwrap_or_else(|_| "{}".to_string()); format!("event: turn_info\ndata: {}\n\n", data) } @@ -1967,6 +1996,16 @@ pub async fn cancel_turn_handler( http_request: HttpRequest, path: web::Path, app_state: web::Data, +) -> impl Responder { + cancel_turn_impl(http_request, path, app_state).await +} + +/// Core of the turn-cancel, attribute-free so the persona-chat DELETE route +/// can reuse it under `/persona_chat/turn/{turn_id}`. +pub(crate) async fn cancel_turn_impl( + http_request: HttpRequest, + path: web::Path, + app_state: web::Data, ) -> impl Responder { let turn_id = path.into_inner(); @@ -2017,6 +2056,13 @@ pub async fn cancel_turn_handler( })) } +// Persona-chat handlers (`/persona_chat/*`) live in `ai::persona_chat` and +// are registered in main.rs. The persona SSE replay and cancel routes are +// thin wrappers over `turn_replay_handler` + `cancel_turn_handler`: both +// are keyed on `turn_id`, not file_path, and `TurnInfo::scope` ("persona") +// lets the shared `turn_info` frame carry the right identity (persona_id, +// no file_path). + #[cfg(test)] mod turn_replay_tests { use super::{cancel_turn_handler, render_indexed_frame, turn_replay_handler}; diff --git a/src/ai/insight_chat.rs b/src/ai/insight_chat.rs index 75d050c..bc070f2 100644 --- a/src/ai/insight_chat.rs +++ b/src/ai/insight_chat.rs @@ -20,7 +20,7 @@ use crate::utils::{normalize_path, retry_with_backoff}; use futures::stream::{BoxStream, StreamExt}; use uuid::Uuid; -const DEFAULT_MAX_ITERATIONS: usize = 6; +pub const DEFAULT_MAX_ITERATIONS: usize = 6; /// Assumed context window when the request doesn't specify `num_ctx`. /// The llama-swap chat slots serve 20k-131k contexts and real conversations /// rarely pass ~16k tokens, so 32k keeps the truncation pass from gutting @@ -30,11 +30,11 @@ 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()); @@ -1303,7 +1312,12 @@ impl InsightChatService { } /// Agentic loop variant that pushes events to a `TurnEntry` buffer. - async fn run_streaming_agentic_loop_with_entry( + /// Same as `run_streaming_agentic_loop` but emits events to a + /// `TurnEntry` for SSE replay. Thin wrapper around the free function + /// `run_streaming_agentic_loop_with_entry` so the persona chat + /// (and any future chat-without-file surface) can reuse the loop + /// body without instantiating an `InsightChatService`. + pub async fn run_streaming_agentic_loop_with_entry( &self, backend: &ResolvedBackend, messages: &mut Vec, @@ -1315,160 +1329,19 @@ impl InsightChatService { max_iterations: usize, entry: &Arc, ) -> Result { - let mut tool_calls_made = 0usize; - let mut iterations_used = 0usize; - let mut last_prompt_eval_count: Option = None; - let mut last_eval_count: Option = None; - let mut final_content = String::new(); - - for iteration in 0..max_iterations { - // Cooperative cancellation: a DELETE flips status out of Running - // (and aborts this task). Check at the iteration boundary so an - // in-flight tool round finishes cleanly rather than mid-write. - if !entry.is_running() { - return Ok(AgenticLoopOutcome { - tool_calls_made, - iterations_used, - last_prompt_eval_count, - last_eval_count, - final_content, - cancelled: true, - }); - } - - iterations_used = iteration + 1; - let _ = entry - .push_event(ChatStreamEvent::IterationStart { - n: iterations_used, - max: max_iterations, - }) - .await; - - let mut stream = backend - .chat() - .chat_with_tools_stream(messages.clone(), tools.clone()) - .await?; - - let mut final_message: Option = None; - while let Some(ev) = stream.next().await { - let ev = ev?; - match ev { - LlmStreamEvent::TextDelta(delta) => { - let _ = entry.push_event(ChatStreamEvent::TextDelta(delta)).await; - } - LlmStreamEvent::Done { - message, - prompt_eval_count, - eval_count, - } => { - last_prompt_eval_count = prompt_eval_count; - last_eval_count = eval_count; - final_message = Some(message); - break; - } - } - } - let mut response = - final_message.ok_or_else(|| anyhow!("stream ended without a Done event"))?; - - if let Some(ref mut tcs) = response.tool_calls { - for tc in tcs.iter_mut() { - if !tc.function.arguments.is_object() { - tc.function.arguments = serde_json::Value::Object(Default::default()); - } - } - } - - messages.push(response.clone()); - - if let Some(ref tool_calls) = response.tool_calls - && !tool_calls.is_empty() - { - for tool_call in tool_calls { - tool_calls_made += 1; - let call_index = tool_calls_made - 1; - let _ = entry - .push_event(ChatStreamEvent::ToolCall { - index: call_index, - name: tool_call.function.name.clone(), - arguments: tool_call.function.arguments.clone(), - }) - .await; - let cx = opentelemetry::Context::new(); - let result = self - .generator - .execute_tool( - &tool_call.function.name, - &tool_call.function.arguments, - backend, - image_base64, - normalized, - user_id, - active_persona, - &cx, - ) - .await; - let (result_preview, result_truncated) = truncate_tool_result(&result); - let _ = entry - .push_event(ChatStreamEvent::ToolResult { - index: call_index, - name: tool_call.function.name.clone(), - result: result_preview, - result_truncated, - }) - .await; - messages.push(ChatMessage::tool_result(result)); - } - continue; - } - - final_content = response.content; - break; - } - - // No-tools fallback - if final_content.is_empty() { - let synthetic_idx = push_synthetic_final_prompt(messages); - let mut stream = backend - .chat() - .chat_with_tools_stream(messages.clone(), vec![]) - .await?; - let mut final_message: Option = None; - while let Some(ev) = stream.next().await { - let ev = ev?; - match ev { - LlmStreamEvent::TextDelta(delta) => { - let _ = entry.push_event(ChatStreamEvent::TextDelta(delta)).await; - } - LlmStreamEvent::Done { - message, - prompt_eval_count, - eval_count, - } => { - last_prompt_eval_count = prompt_eval_count; - last_eval_count = eval_count; - final_message = Some(message); - break; - } - } - } - let final_response = - final_message.ok_or_else(|| anyhow!("final stream ended without a Done event"))?; - final_content = final_response.content.clone(); - messages.push(final_response); - remove_synthetic_final_prompt(messages, synthetic_idx); - } - - Ok(AgenticLoopOutcome { - tool_calls_made, - iterations_used, - last_prompt_eval_count, - last_eval_count, - // Strip any leaked reasoning block from the content the - // caller persists as title/summary (the raw transcript keeps it). - final_content: crate::ai::llm_client::strip_think_blocks(&final_content), - cancelled: false, - }) + crate::ai::insight_chat::run_streaming_agentic_loop_with_entry( + &self.generator, + backend, + messages, + tools, + image_base64, + normalized, + user_id, + active_persona, + max_iterations, + entry, + ) + .await } async fn run_streaming_turn( @@ -2212,17 +2085,20 @@ fn resolve_bootstrap_backend(supplied: Option<&str>) -> Result { } /// Outcome of one streaming agentic loop pass. Shared between bootstrap -/// and continuation. -struct AgenticLoopOutcome { - tool_calls_made: usize, - iterations_used: usize, - last_prompt_eval_count: Option, - last_eval_count: Option, - final_content: String, +/// and continuation. `pub` so the persona chat surface (and any future +/// chat-without-file surface) can read the result of +/// `run_streaming_agentic_loop_with_entry` without reaching back into +/// private fields. +pub struct AgenticLoopOutcome { + pub tool_calls_made: usize, + pub iterations_used: usize, + pub last_prompt_eval_count: Option, + pub last_eval_count: Option, + pub final_content: String, /// True when the loop exited early because the turn was cancelled /// (status flipped out of `Running`). Callers skip persistence and the /// terminal `Done` push — the cancel handler owns the terminal event. - cancelled: bool, + pub cancelled: bool, } /// Events emitted by `chat_turn_stream`. One stream per turn; ends after @@ -2341,7 +2217,7 @@ pub(crate) fn find_raw_cut( /// Read AGENTIC_CHAT_MAX_ITERATIONS once per call. Cheap; keeps the code /// free of static globals and lets the operator change the cap by env without /// a restart in test harnesses (the running server still caches via Default). -fn env_max_iterations() -> usize { +pub fn env_max_iterations() -> usize { std::env::var("AGENTIC_CHAT_MAX_ITERATIONS") .ok() .and_then(|s| s.parse::().ok()) @@ -2352,7 +2228,7 @@ 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()) @@ -2395,6 +2271,174 @@ fn restore_system_content(messages: &mut [ChatMessage], original: Option } /// Append the synthetic "write your final answer" user prompt, returning the +/// Free-function form of `InsightChatService::run_streaming_agentic_loop_with_entry`. +/// Persona chat (and any future chat-without-file surface) doesn't need an +/// `InsightChatService` — it just needs the agent loop. This is the same +/// body lifted out of the `InsightChatService` impl, taking +/// `&InsightGenerator` so any caller in the crate can drive it. +/// +/// Public so the persona chat session can call into it directly. +pub async fn run_streaming_agentic_loop_with_entry( + generator: &crate::ai::insight_generator::InsightGenerator, + backend: &ResolvedBackend, + messages: &mut Vec, + tools: Vec, + image_base64: &Option, + normalized: &str, + user_id: i32, + active_persona: &str, + max_iterations: usize, + entry: &Arc, +) -> Result { + let mut tool_calls_made = 0usize; + let mut iterations_used = 0usize; + let mut last_prompt_eval_count: Option = None; + let mut last_eval_count: Option = None; + let mut final_content = String::new(); + + for iteration in 0..max_iterations { + if !entry.is_running() { + return Ok(AgenticLoopOutcome { + tool_calls_made, + iterations_used, + last_prompt_eval_count, + last_eval_count, + final_content, + cancelled: true, + }); + } + + iterations_used = iteration + 1; + let _ = entry + .push_event(ChatStreamEvent::IterationStart { + n: iterations_used, + max: max_iterations, + }) + .await; + + let mut stream = backend + .chat() + .chat_with_tools_stream(messages.clone(), tools.clone()) + .await?; + + let mut final_message: Option = None; + while let Some(ev) = stream.next().await { + let ev = ev?; + match ev { + LlmStreamEvent::TextDelta(delta) => { + let _ = entry.push_event(ChatStreamEvent::TextDelta(delta)).await; + } + LlmStreamEvent::Done { + message, + prompt_eval_count, + eval_count, + } => { + last_prompt_eval_count = prompt_eval_count; + last_eval_count = eval_count; + final_message = Some(message); + break; + } + } + } + let mut response = + final_message.ok_or_else(|| anyhow!("stream ended without a Done event"))?; + + if let Some(ref mut tcs) = response.tool_calls { + for tc in tcs.iter_mut() { + if !tc.function.arguments.is_object() { + tc.function.arguments = serde_json::Value::Object(Default::default()); + } + } + } + + messages.push(response.clone()); + + if let Some(ref tool_calls) = response.tool_calls + && !tool_calls.is_empty() + { + for tool_call in tool_calls { + tool_calls_made += 1; + let call_index = tool_calls_made - 1; + let _ = entry + .push_event(ChatStreamEvent::ToolCall { + index: call_index, + name: tool_call.function.name.clone(), + arguments: tool_call.function.arguments.clone(), + }) + .await; + let cx = opentelemetry::Context::new(); + let result = generator + .execute_tool( + &tool_call.function.name, + &tool_call.function.arguments, + backend, + image_base64, + normalized, + user_id, + active_persona, + &cx, + ) + .await; + let (result_preview, result_truncated) = truncate_tool_result(&result); + let _ = entry + .push_event(ChatStreamEvent::ToolResult { + index: call_index, + name: tool_call.function.name.clone(), + result: result_preview, + result_truncated, + }) + .await; + messages.push(ChatMessage::tool_result(result)); + } + continue; + } + + final_content = response.content; + break; + } + + if final_content.is_empty() { + let synthetic_idx = push_synthetic_final_prompt(messages); + let mut stream = backend + .chat() + .chat_with_tools_stream(messages.clone(), vec![]) + .await?; + let mut final_message: Option = None; + while let Some(ev) = stream.next().await { + let ev = ev?; + match ev { + LlmStreamEvent::TextDelta(delta) => { + let _ = entry.push_event(ChatStreamEvent::TextDelta(delta)).await; + } + LlmStreamEvent::Done { + message, + prompt_eval_count, + eval_count, + } => { + last_prompt_eval_count = prompt_eval_count; + last_eval_count = eval_count; + final_message = Some(message); + break; + } + } + } + let final_response = + final_message.ok_or_else(|| anyhow!("final stream ended without a Done event"))?; + final_content = final_response.content.clone(); + messages.push(final_response); + remove_synthetic_final_prompt(messages, synthetic_idx); + } + + Ok(AgenticLoopOutcome { + tool_calls_made, + iterations_used, + last_prompt_eval_count, + last_eval_count, + final_content: crate::ai::llm_client::strip_think_blocks(&final_content), + cancelled: false, + }) +} + /// index the caller must later hand to [`remove_synthetic_final_prompt`]. /// Used when the agentic loop exhausts its budget: the model gets one more /// (tool-free) request, but the nudge itself must never persist — it would @@ -2499,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>) { @@ -2638,9 +2682,9 @@ pub struct ToolInvocation { /// Soft cap for tool-result bodies returned via the history API. Keeps /// payloads small for the mobile client — verbose SMS / geocoding responses /// don't need to ship in full for inspection. -const TOOL_RESULT_PREVIEW_MAX: usize = 2000; +pub(crate) const TOOL_RESULT_PREVIEW_MAX: usize = 2000; -fn truncate_tool_result(s: &str) -> (String, bool) { +pub(crate) fn truncate_tool_result(s: &str) -> (String, bool) { if s.len() <= TOOL_RESULT_PREVIEW_MAX { (s.to_string(), false) } else { 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 3f17064..d39645a 100644 --- a/src/ai/mod.rs +++ b/src/ai/mod.rs @@ -13,6 +13,7 @@ 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; @@ -38,6 +39,12 @@ pub use llamacpp::LlamaCppClient; 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 new file mode 100644 index 0000000..596a809 --- /dev/null +++ b/src/ai/persona_chat.rs @@ -0,0 +1,2187 @@ +//! Open chat with persona — file-anchored insight chat's sibling. +//! +//! 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 +//! (gated by the persona's `ToolGateOpts` — no `describe_photo`, no +//! `get_file_tags`, etc., because there is no file). +//! 2. `InsightChatService::run_streaming_agentic_loop_with_entry` for the +//! actual turn loop. It already pushes events to a `TurnEntry` and +//! handles cancellation; we just feed it a no-photo argument set. +//! 3. The shared `TurnRegistry` — extended with a `scope` label so the +//! SSE `turn_info` payload can distinguish the two. +//! +//! What this module owns: +//! - The persona-scoped persistence DAO. +//! - Pure helpers (prompt builder, validation gates, seed-message shape). +//! - The session that wires the agent loop to the persona DAO + turn +//! registry. +//! - The HTTP handlers under ` /persona_chat/*`. + +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}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex as StdMutex}; +use tokio::sync::Mutex as TokioMutex; +use uuid::Uuid; + +use crate::ai::backend::{BackendKind, SamplingOverrides}; +use crate::ai::handlers::ReplayQuery; +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::{ChatHistoryStore, ChatMessage, StoredChatNode}; +use crate::ai::turn_registry::{TurnEntry, TurnRegistry}; +use crate::data::Claims; +use crate::database::PersonaDao; +use crate::libraries; +use crate::otel::{extract_context_from_request, global_tracer}; +use crate::state::AppState; + +// ───────────────────────────────────────────────────────────────────── +// Errors +// ───────────────────────────────────────────────────────────────────── + +#[derive(Debug)] +pub enum PersonaChatError { + UnknownPersona(String), + UnknownConversation(String), + EmptyMessage, + MessageTooLong, + ConcurrentTurn, + Db(anyhow::Error), +} + +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::Db(e) => write!(f, "database error: {e}"), + } + } +} + +impl std::error::Error for PersonaChatError {} + +impl From for PersonaChatError { + fn from(e: anyhow::Error) -> Self { + PersonaChatError::Db(e) + } +} + +// ───────────────────────────────────────────────────────────────────── +// Pure helpers — covered by unit tests at the bottom of this file. +// ───────────────────────────────────────────────────────────────────── + +/// Resolve the system prompt for a persona chat turn. +/// +/// In the v1 scope there's no per-turn persona override: the persona's +/// stored `systemPrompt` is authoritative. A future "voice override" can +/// extend this signature without touching the call sites. +pub fn resolve_persona_system_prompt( + persona: Option<&crate::database::models::Persona>, +) -> Option { + persona + .filter(|p| !p.system_prompt.trim().is_empty()) + .map(|p| p.system_prompt.clone()) +} + +/// Build the seed messages for a fresh persona conversation. +/// +/// 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())] +} + +/// 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_store(store: &ChatHistoryStore) -> Result { + serde_json::to_string(store).with_context(|| "failed to serialize persona chat history") +} + +/// 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) + { + messages.remove(idx); + } + messages +} + +fn is_seed_greeting(content: &str) -> bool { + content.starts_with("Hi — I'm") && content.contains("ready to help") +} + +// ───────────────────────────────────────────────────────────────────── +// Wire shapes — snake_case, matching the file-chat convention. +// ───────────────────────────────────────────────────────────────────── + +#[derive(Debug, Deserialize)] +pub struct PersonaChatTurnRequest { + /// 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)] + pub model: Option, + #[serde(default)] + pub backend: Option, + #[serde(default)] + pub num_ctx: Option, + #[serde(default)] + pub temperature: Option, + #[serde(default)] + pub top_p: Option, + #[serde(default)] + pub top_k: Option, + #[serde(default)] + pub min_p: Option, + #[serde(default)] + pub enable_thinking: Option, + /// Per-turn system-prompt override. Appended to the persona's prompt + /// for this turn only; not persisted (persona voice survives a + /// voice-flipped turn). Empty / whitespace = no override. + #[serde(default)] + pub system_prompt: Option, + #[serde(default)] + pub library: Option, +} + +#[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, + /// 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 +/// `RenderedHistoryMessage` wire shape: `tools` carries the tool +/// invocations that led to the assistant reply (empty for user turns). +#[derive(Debug, Serialize)] +pub struct RenderedPersonaMessage { + pub role: String, + pub content: String, + pub is_initial: bool, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub tools: Vec, +} + +/// 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) +} + +/// 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")) +} + +/// 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()) +} + +/// 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(), + backend: String::new(), + active_leaf_id: 0, + viewing_branch_id: 0, + fork_info: Vec::new(), + } + } +} + +// ───────────────────────────────────────────────────────────────────── +// Per-persona in-flight mutex +// ───────────────────────────────────────────────────────────────────── + +/// 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 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 { + turns: StdMutex>, +} + +impl InFlightPersonaTurns { + pub fn new() -> Self { + Self::default() + } + + /// 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(conversation_id.to_string(), turn_id.to_string()); + Ok(()) + } + + pub fn release(&self, conversation_id: &str) { + if let Ok(mut map) = self.turns.lock() { + map.remove(conversation_id); + } + } +} + +/// Frees the in-flight slot when the spawned turn task ends — on completion, +/// error, or abort. +struct InFlightGuard { + in_flight: Arc, + conversation_id: String, +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + self.in_flight.release(&self.conversation_id); + } +} + +#[derive(Clone)] +pub struct PersonaChatSession { + generator: Arc, + persona_dao: Arc>>, + chat_locks: Arc, + in_flight: Arc, +} + +impl PersonaChatSession { + pub fn new( + generator: Arc, + persona_dao: Arc>>, + chat_locks: Arc, + ) -> Self { + Self { + generator, + persona_dao, + chat_locks, + in_flight: Arc::new(InFlightPersonaTurns::new()), + } + } + + /// 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 { + 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"); + // 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 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 + }) + } + + /// Dispatch a new turn. Returns the `turn_id` immediately; the + /// streaming agent loop runs in a Tokio task and pushes events to + /// the registered `TurnEntry`. + pub async fn dispatch_turn( + self: Arc, + registry: Arc, + user_id: i32, + library_id: i32, + req: PersonaChatTurnRequest, + ) -> Result { + let trimmed = req.user_message.trim().to_string(); + if trimmed.is_empty() { + return Err(PersonaChatError::EmptyMessage); + } + if trimmed.len() > 8192 { + return Err(PersonaChatError::MessageTooLong); + } + 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(); + // 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"); + 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 turn_id = Uuid::new_v4().to_string(); + self.in_flight + .claim(&conversation_id, &turn_id) + .map_err(|_| PersonaChatError::ConcurrentTurn)?; + + let entry = Arc::new(TurnEntry::new_persona( + turn_id.clone(), + persona_id.clone(), + library_id, + )); + registry.insert(entry.clone()).await; + + let svc = self.clone(); + let persona_for_task = persona.clone(); + let entry_clone = entry.clone(); + let turn_id_for_task = turn_id.clone(); + let req_for_task = PersonaChatTurnRequest { + user_message: trimmed.clone(), + ..req + }; + let in_flight_guard = InFlightGuard { + in_flight: self.in_flight.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 + .run_streaming_turn( + persona_for_task, + req_for_task, + entry_clone.clone(), + user_id, + library_id, + ) + .await; + match result { + Ok(()) => { + span.set_attribute(KeyValue::new("status", "done")); + span.set_status(Status::Ok); + } + Err(e) => { + span.set_attribute(KeyValue::new("status", "error")); + span.set_status(Status::error(format!("{e}"))); + let _ = entry_clone + .push_event(ChatStreamEvent::Error(format!("{e}"))) + .await; + } + } + entry_clone.set_terminal_status(crate::ai::turn_registry::TurnStatus::Done); + // Frees the in-flight slot. Placed explicitly so the guard is + // captured by the spawned task: it must drop when the task ends + // (completion, error, or abort) — not when dispatch returns. + drop(in_flight_guard); + }); + + entry.set_abort_handle(handle.abort_handle()); + Ok(turn_id) + } + + /// The streaming turn body. Loads the persona, builds the messages + /// vector (system + persisted history + new user turn), runs the + /// shared agent loop, persists the updated transcript. + async fn run_streaming_turn( + self: Arc, + persona: crate::database::models::Persona, + req: PersonaChatTurnRequest, + entry: Arc, + user_id: i32, + _library_id: i32, + ) -> Result<()> { + // 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(conversation_id.clone()) + .or_insert_with(|| Arc::new(TokioMutex::new(()))) + .clone() + }; + let _guard = lock.lock().await; + + // 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, &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 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. + let backend_kind = match req + .backend + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_lowercase) + .as_deref() + { + Some("hybrid") => BackendKind::Hybrid, + _ => BackendKind::Local, + }; + let max_iterations = DEFAULT_MAX_ITERATIONS.clamp(1, env_max_iterations()); + let overrides = SamplingOverrides { + model: req.model.clone().filter(|m| !m.is_empty()), + num_ctx: req.num_ctx, + temperature: req.temperature, + top_p: req.top_p, + top_k: req.top_k, + min_p: req.min_p, + enable_thinking: req.enable_thinking, + }; + 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 tools = InsightGenerator::build_tool_definitions(gate_opts); + + // Append the new user turn + apply the per-turn system-prompt + // override. Per-persona overrides are NOT persisted (the persona + // voice survives a voice-flipped turn); applied ephemerally. + 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()) + } + } else { + None + }; + + // 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. + // Restored to the persisted value below before serialization. + if let Some(sys) = messages.iter_mut().find(|m| m.role == "system") { + sys.content = format!("{}\n\n{}", sys.content, note); + } + } + + // Run the shared agent loop. `image_base64 = None` and `normalized = ""` + // because there's no photo — `execute_tool` is gated by tool name and + // returns "tool not found" for any file-specific tool the LLM might + // hallucinate. + let outcome = crate::ai::insight_chat::run_streaming_agentic_loop_with_entry( + &self.generator, + &backend, + &mut messages, + tools, + &None, + "", + user_id, + &persona.persona_id, + max_iterations, + &entry, + ) + .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 + && let Some(sys) = messages.iter_mut().find(|m| m.role == "system") + { + let trimmed = format!("\n\n{}", note); + if sys.content.ends_with(&trimmed) { + sys.content.truncate(sys.content.len() - trimmed.len()); + } + } + + // 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"); + 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, + prompt_tokens: outcome.last_prompt_eval_count, + eval_tokens: outcome.last_eval_count, + num_ctx: req.num_ctx, + amended_insight_id: None, + backend_used: backend_kind.as_str().to_string(), + model_used, + cancelled: outcome.cancelled, + }) + .await; + Ok(()) + } + + /// 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"); + 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)) + } +} + +// ───────────────────────────────────────────────────────────────────── +// HTTP handlers — /persona_chat/* +// ───────────────────────────────────────────────────────────────────── + +/// Query params for GET /persona_chat/history. `library` is accepted for +/// client parity with the file chat but does not scope persistence — a +/// persona transcript is keyed on `(user_id, persona_id)` only. +#[derive(Debug, Deserialize)] +pub struct PersonaChatHistoryQuery { + 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/conversations — start a new conversation. +#[derive(Debug, Deserialize)] +pub struct PersonaChatCreateRequest { + pub persona_id: String, +} + +/// GET /persona_chat/history?persona_id=... — the rendered rolling +/// transcript. A persona that has never been chatted returns the empty +/// envelope (200), so the client can render its empty state. +#[get("/persona_chat/history")] +pub async fn persona_chat_history_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.load_history( + user_id, + &query.conversation_id, + query.branch_id, + ) { + Ok(view) => HttpResponse::Ok().json(view), + Err(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 })) + } + } + } +} + +/// POST /persona_chat/turn — dispatch an async turn. Returns 202 with +/// the `turn_id` immediately; the streaming agent loop runs in a +/// background task and the client opens the SSE replay stream with it. +#[post("/persona_chat/turn")] +pub async fn persona_chat_turn_handler( + http_request: HttpRequest, + claims: Claims, + request: web::Json, + app_state: web::Data, +) -> impl Responder { + 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( + "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 + // file chat's resolution so a `?library=` pick steers the tools. + let library = + match libraries::resolve_library_param_state(&app_state, request.library.as_deref()) { + Ok(Some(lib)) => lib, + Ok(None) => app_state.primary_library(), + Err(e) => { + return HttpResponse::BadRequest().json(serde_json::json!({ + "error": format!("invalid library: {e}") + })); + } + }; + + let user_id = claims.sub.parse::().unwrap_or(1); + let session = app_state.persona_chat_session.clone(); + let registry = app_state.turn_registry.clone(); + + match session + .dispatch_turn(registry, user_id, library.id, request.into_inner()) + .await + { + Ok(turn_id) => { + span.set_attribute(KeyValue::new("turn_id", turn_id.clone())); + span.set_status(Status::Ok); + HttpResponse::Accepted().json(serde_json::json!({ + "turn_id": turn_id, + "status": "running" + })) + } + Err(e) => { + span.set_status(Status::error(format!("{e}"))); + match &e { + PersonaChatError::UnknownPersona(_) | PersonaChatError::UnknownConversation(_) => { + HttpResponse::NotFound() + } + PersonaChatError::EmptyMessage | PersonaChatError::MessageTooLong => { + HttpResponse::BadRequest() + } + PersonaChatError::ConcurrentTurn => HttpResponse::Conflict(), + PersonaChatError::Db(_) => { + log::error!("persona chat dispatch failed: {e}"); + HttpResponse::InternalServerError() + } + } + .json(serde_json::json!({ "error": format!("{e}") })) + } + } +} + +/// 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); + match app_state + .persona_chat_session + .create_conversation(user_id, &request.persona_id) + { + 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 delete failed: {e}"); + HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("{e}") + })) + } + } +} + +/// 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 +/// `scope: "persona"` + `persona_id` (and no `file_path`) via the +/// shared renderer. +#[get("/persona_chat/turn/{turn_id}")] +pub async fn persona_turn_replay_handler( + http_request: HttpRequest, + path: web::Path, + query: web::Query, + app_state: web::Data, +) -> HttpResponse { + crate::ai::handlers::turn_replay_impl(http_request, path, query, app_state).await +} + +/// DELETE /persona_chat/turn/{turn_id} — cancel a running persona turn. +#[delete("/persona_chat/turn/{turn_id}")] +pub async fn persona_turn_cancel_handler( + http_request: HttpRequest, + path: web::Path, + app_state: web::Data, +) -> impl Responder { + crate::ai::handlers::cancel_turn_impl(http_request, path, app_state).await +} + +// ───────────────────────────────────────────────────────────────────── +// Unit tests — pure helpers only. DAO + handler tests live in their +// respective test modules (`persona_chat_dao.rs` and `handlers.rs`). +// ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + fn persona_with_prompt(p: &str) -> crate::database::models::Persona { + crate::database::models::Persona { + id: 0, + user_id: 1, + persona_id: "test".to_string(), + name: "Tester".to_string(), + system_prompt: p.to_string(), + is_built_in: false, + include_all_memories: false, + created_at: 0, + updated_at: 0, + reviewed_only_facts: false, + allow_agent_corrections: false, + } + } + + #[test] + fn resolve_persona_system_prompt_returns_stored_prompt() { + let p = persona_with_prompt("You are X."); + assert_eq!( + resolve_persona_system_prompt(Some(&p)).as_deref(), + Some("You are X.") + ); + } + + #[test] + fn resolve_persona_system_prompt_falls_back_to_none_when_empty() { + let p = persona_with_prompt(""); + assert!(resolve_persona_system_prompt(Some(&p)).is_none()); + } + + #[test] + fn resolve_persona_system_prompt_handles_missing_persona() { + assert!(resolve_persona_system_prompt(None).is_none()); + } + + fn assistant_msg(content: &str) -> ChatMessage { + ChatMessage { + role: "assistant".to_string(), + content: content.to_string(), + tool_calls: None, + images: None, + } + } + + #[test] + 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."); + } + + 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 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 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_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(); + assert_eq!(v.messages.len(), 0); + assert_eq!(v.turn_count, 0); + assert_eq!(v.active_leaf_id, 0); + assert_eq!(v.viewing_branch_id, 0); + assert!(v.fork_info.is_empty()); + } + + #[test] + fn persona_chat_validation_flags_empty_user_message() { + let _req = PersonaChatTurnRequest { + conversation_id: "conv-1".to_string(), + user_message: " ".to_string(), + model: None, + backend: None, + num_ctx: None, + temperature: None, + top_p: None, + top_k: None, + min_p: None, + enable_thinking: None, + system_prompt: None, + library: None, + }; + assert_eq!( + format!("{}", PersonaChatError::EmptyMessage), + "user_message must not be empty" + ); + } + + #[test] + fn persona_chat_validation_flags_oversized_user_message() { + let _req = PersonaChatTurnRequest { + conversation_id: "conv-1".to_string(), + user_message: "x".repeat(8193), + model: None, + backend: None, + num_ctx: None, + temperature: None, + top_p: None, + top_k: None, + min_p: None, + enable_thinking: None, + system_prompt: None, + library: None, + }; + assert_eq!( + format!("{}", PersonaChatError::MessageTooLong), + "user_message exceeds 8192 chars" + ); + } + + #[test] + fn persona_chat_validation_flags_unknown_persona() { + let err = PersonaChatError::UnknownPersona("nope".to_string()); + assert_eq!(format!("{err}"), "persona 'nope' not found"); + } + + #[test] + fn persona_chat_error_concurrent_turn_displays_message() { + assert_eq!( + format!("{}", PersonaChatError::ConcurrentTurn), + "another turn for this persona is in flight" + ); + } + + #[test] + fn persona_chat_turn_returns_409_when_concurrent_turn_in_flight() { + let in_flight = InFlightPersonaTurns::new(); + assert!(in_flight.claim("conv-a", "t1").is_ok()); + + // 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("conv-a", "t2").unwrap_err(), "t1"); + assert_eq!(in_flight.claim("conv-a", "t3").unwrap_err(), "t1"); + + // 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("conv-a"); + assert!(in_flight.claim("conv-a", "t6").is_ok()); + } + + #[test] + fn in_flight_guard_releases_slot_on_drop() { + let in_flight = Arc::new(InFlightPersonaTurns::new()); + let tracker = in_flight.clone(); + // Mirrors dispatch_turn: claim first, guard holds the slot. + assert!(tracker.claim("conv-a", "t-guard").is_ok()); + { + let _guard = InFlightGuard { + in_flight, + conversation_id: "conv-a".to_string(), + }; + assert_eq!(tracker.claim("conv-a", "other").unwrap_err(), "t-guard"); + } + // Guard dropped → slot free again. + assert!(tracker.claim("conv-a", "t7").is_ok()); + } + + #[test] + fn persona_chat_turn_request_deserializes_snake_case_wire_body() { + // Pins the mobile-client contract: snake_case keys, same as the + // file-chat request shapes. A camelCase regression here 400s the + // client's dispatch with "missing field persona_id". + let body = r#"{ + "conversation_id": "conv-1", + "user_message": "hello", + "num_ctx": 4096, + "top_p": 0.9, + "top_k": 40, + "min_p": 0.05, + "enable_thinking": true, + "system_prompt": "be brief", + "library": "main" + }"#; + let req: PersonaChatTurnRequest = serde_json::from_str(body).unwrap(); + 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)); + assert_eq!(req.top_k, Some(40)); + assert_eq!(req.min_p, Some(0.05)); + assert_eq!(req.enable_thinking, Some(true)); + assert_eq!(req.system_prompt.as_deref(), Some("be brief")); + assert_eq!(req.library.as_deref(), Some("main")); + } + + #[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(), + is_initial: true, + tools: Vec::new(), + }], + turn_count: 1, + model_version: "x".to_string(), + backend: "local".to_string(), + 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"], 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_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"); + } + + /// 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/ai/turn_registry.rs b/src/ai/turn_registry.rs index 2a5d432..fb82638 100644 --- a/src/ai/turn_registry.rs +++ b/src/ai/turn_registry.rs @@ -52,6 +52,10 @@ pub struct TurnInfo { pub turn_id: String, pub file_path: String, pub library_id: i32, + /// Persona id for persona-scoped turns. None for insight-scoped turns. + pub persona_id: Option, + /// `"insight"` (file-anchored chat) | `"persona"` (open chat). + pub scope: String, pub status: TurnStatus, pub total_events_pushed: u32, pub buffered_count: u32, @@ -78,8 +82,20 @@ pub enum ReplayOutcome { /// replay connections (readers). pub struct TurnEntry { pub turn_id: String, + /// Stable identity used to scope the turn. For insight chat this is + /// `(file_path, library_id)`. For persona chat it is `(persona_id)`. + /// The generic fields stay populated regardless of scope so the SSE + /// `turn_info` frame can always surface a useful label — a future + /// "list active turns" debugging surface won't need to discriminate. pub file_path: String, pub library_id: i32, + /// Persona id for persona-scoped turns (`scope == "persona"`). None for + /// the existing insight-scoped turns. The field is the cheapest way to + /// let the SSE `turn_info` payload round-trip without a parallel map. + pub persona_id: Option, + /// `"insight"` (default — file-anchored chat) | `"persona"` (open chat). + /// Stable string so the SSE `turn_info` payload can carry it as-is. + pub scope: String, /// Shared event buffer — multiple SSE connections can read independently. /// Each connection tracks its own `skip_before` offset. events: Mutex>, @@ -100,10 +116,35 @@ pub struct TurnEntry { impl TurnEntry { pub fn new(turn_id: String, file_path: String, library_id: i32) -> Self { + Self::with_scope(turn_id, file_path, library_id, None, "insight") + } + + /// Persona-scoped constructor. `file_path` is unused by the persona + /// chat loop but kept on the struct so the SSE `turn_info` frame has + /// the same shape across scopes. + pub fn new_persona(turn_id: String, persona_id: String, library_id: i32) -> Self { + Self::with_scope( + turn_id, + String::new(), + library_id, + Some(persona_id), + "persona", + ) + } + + fn with_scope( + turn_id: String, + file_path: String, + library_id: i32, + persona_id: Option, + scope: &str, + ) -> Self { Self { turn_id, file_path, library_id, + persona_id, + scope: scope.to_string(), events: Mutex::new(Vec::new()), total_events_pushed: AtomicU32::new(0), base_index: AtomicU32::new(0), @@ -170,6 +211,8 @@ impl TurnEntry { turn_id: self.turn_id.clone(), file_path: self.file_path.clone(), library_id: self.library_id, + persona_id: self.persona_id.clone(), + scope: self.scope.clone(), status: self.status.load(Ordering::Relaxed).into(), total_events_pushed: total, buffered_count: buffered, @@ -745,4 +788,67 @@ mod tests { let from_base = events_of(entry.replay_from(5).await); assert_eq!(from_base.len(), MAX_BUFFERED_EVENTS); } + + // ── Persona-scope regression guards ─────────────────────────── + // + // The open-chat surface relies on `scope == "persona"` and the + // optional `persona_id` surviving a round-trip through `info()`. + // A future refactor that hard-codes "insight" or drops the field + // would silently break the SSE `turn_info` payload — guard it. + + #[tokio::test] + async fn turn_entry_supports_persona_scope_label() { + let entry = Arc::new(TurnEntry::new_persona( + "tp1".to_string(), + "journal".to_string(), + 1, + )); + assert_eq!(entry.scope, "persona"); + assert_eq!(entry.persona_id.as_deref(), Some("journal")); + let info = entry.info().await; + assert_eq!(info.scope, "persona"); + assert_eq!(info.persona_id.as_deref(), Some("journal")); + // File-anchored fields stay blank — the persona chat has no file. + assert!(info.file_path.is_empty()); + } + + #[tokio::test] + async fn turn_info_payload_for_persona_does_not_leak_file_path() { + // Defensive: even if some future code path accidentally populates + // file_path on a persona entry, the SSE `turn_info` payload + // should not surface it to clients. Today file_path is always + // empty for persona entries, so we assert that contract. + let entry = Arc::new(TurnEntry::new_persona( + "tp2".to_string(), + "default".to_string(), + 1, + )); + let info = entry.info().await; + let json = serde_json::to_string(&serde_json::json!({ + "turn_id": info.turn_id, + "persona_id": info.persona_id, + "scope": info.scope, + "status": info.status.as_str(), + "total_events_pushed": info.total_events_pushed, + "buffered_count": info.buffered_count, + })) + .unwrap(); + // Privacy: a persona chat reply that quotes an SMS must not echo + // the file path that message originally surfaced from. + assert!( + !json.contains("file_path"), + "persona turn_info payload should not leak file_path: {json}" + ); + } + + #[tokio::test] + async fn insight_scope_defaults_when_using_legacy_constructor() { + // The legacy `TurnEntry::new` constructor (used by the existing + // file-anchored insight chat) must still report scope="insight" + // so callers can discriminate without a parallel map. + let entry = Arc::new(TurnEntry::new("t-insight".into(), "/p.jpg".into(), 1)); + let info = entry.info().await; + assert_eq!(info.scope, "insight"); + assert!(info.persona_id.is_none()); + } } diff --git a/src/database/persona_dao.rs b/src/database/persona_dao.rs index 6ceb2af..23a8ec0 100644 --- a/src/database/persona_dao.rs +++ b/src/database/persona_dao.rs @@ -10,6 +10,51 @@ use crate::database::schema; use crate::database::{DbError, DbErrorKind, connect}; 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)`). +/// +/// 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). @@ -80,6 +125,76 @@ pub trait PersonaDao: Sync + Send { user_id: i32, personas: &[ImportPersona], ) -> Result; + + // ── Persona-chat (open chat with persona) persistence ─────────── + // + // 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 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, + conversation_id: &str, + ) -> Result, DbError>; + + /// 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; + + /// 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, + 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>; } pub struct SqlitePersonaDao { @@ -251,12 +366,28 @@ impl PersonaDao for SqlitePersonaDao { pid: &str, ) -> Result { trace_db_call(cx, "delete", "delete_persona", |_span| { - use schema::personas::dsl::*; let mut conn = self.connection.lock().expect("PersonaDao lock"); - let n = diesel::delete(personas.filter(user_id.eq(uid)).filter(persona_id.eq(pid))) - .execute(conn.deref_mut()) - .map_err(|e| anyhow::anyhow!("Delete error: {}", e))?; - Ok(n > 0) + // One transaction so a persona and its conversations can't get + // out of step. There is no FK between the two tables, and the + // chat list skips conversations whose persona is gone — so + // without this the transcripts would linger invisibly forever. + conn.deref_mut().transaction::<_, anyhow::Error, _>(|tx| { + { + use schema::persona_chat_conversations::dsl::*; + diesel::delete( + persona_chat_conversations + .filter(user_id.eq(uid)) + .filter(persona_id.eq(pid)), + ) + .execute(tx) + .map_err(|e| anyhow::anyhow!("Delete error: {}", e))?; + } + use schema::personas::dsl::*; + let n = diesel::delete(personas.filter(user_id.eq(uid)).filter(persona_id.eq(pid))) + .execute(tx) + .map_err(|e| anyhow::anyhow!("Delete error: {}", e))?; + Ok(n > 0) + }) }) .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) } @@ -296,8 +427,168 @@ impl PersonaDao for SqlitePersonaDao { }) .map_err(|e| DbError::log(DbErrorKind::InsertError, e)) } -} + fn get_persona_chat( + &mut self, + cx: &opentelemetry::Context, + uid: i32, + 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)) + .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(persona_chat_row)) + .map_err(|e| anyhow::anyhow!("Query error: {}", e)) + }) + .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) + } + + 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, + 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"); + 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 update_persona_chat( + &mut self, + cx: &opentelemetry::Context, + uid: i32, + 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"); + diesel::delete( + persona_chat_conversations + .filter(conversation_id.eq(cid)) + .filter(user_id.eq(uid)), + ) + .execute(conn.deref_mut()) + .map_err(|e| anyhow::anyhow!("Delete error: {}", e))?; + Ok(()) + }) + .map_err(|e| DbError::log(DbErrorKind::QueryError, e)) + } +} #[cfg(test)] mod tests { use super::*; @@ -444,4 +735,201 @@ mod tests { .unwrap(); assert!(updated.include_all_memories); } + + // ── Persona-chat DAO tests ───────────────────────────────────── + + /// 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_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"); + 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_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"); + 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); + assert_eq!(row.created_at, 100, "creation time is not disturbed"); + } + + #[test] + 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, 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!( + ids, + vec![newer, middle, older], + "ordered by updated_at descending so the list screen needs no re-sort" + ); + } + + #[test] + 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("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 deleting_a_persona_takes_its_conversations_with_it() { + // No FK between the tables, and the chat list hides conversations + // whose persona is gone — so an orphan row would be an invisible, + // unreachable transcript sitting in the database forever. + let cx = opentelemetry::Context::new(); + let (mut dao, uid) = dao_with_user("p-cascade"); + dao.create_persona(&cx, uid, "custom-1", "Custom", "prompt", false, false) + .unwrap(); + dao.create_persona_chat(&cx, uid, "custom-1", 100).unwrap(); + dao.create_persona_chat(&cx, uid, "custom-1", 200).unwrap(); + let survivor = dao.create_persona_chat(&cx, uid, "other", 300).unwrap(); + + assert!(dao.delete_persona(&cx, uid, "custom-1").unwrap()); + + let remaining: Vec = dao + .list_persona_chats(&cx, uid) + .unwrap() + .into_iter() + .map(|r| r.conversation_id) + .collect(); + assert_eq!( + remaining, + vec![survivor], + "only the untouched persona's conversation survives" + ); + } + + #[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 846542d..a0d053e 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -171,6 +171,19 @@ diesel::table! { } } +diesel::table! { + 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, + } +} + diesel::table! { personas (id) { id -> Integer, @@ -345,6 +358,7 @@ diesel::allow_tables_to_appear_in_same_query!( insight_generation_jobs, libraries, location_history, + persona_chat_conversations, personas, persons, photo_insights, diff --git a/src/main.rs b/src/main.rs index 9d27f2e..4ad8714 100644 --- a/src/main.rs +++ b/src/main.rs @@ -382,6 +382,16 @@ fn main() -> std::io::Result<()> { .service(ai::turn_async_handler) .service(ai::turn_replay_handler) .service(ai::cancel_turn_handler) + .service(ai::persona_chat_history_handler) + .service(ai::persona_chat_turn_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/state.rs b/src/state.rs index 33e8e3f..3f456e7 100644 --- a/src/state.rs +++ b/src/state.rs @@ -4,6 +4,7 @@ use crate::ai::face_client::FaceClient; use crate::ai::insight_chat::{ChatLockMap, InsightChatService}; use crate::ai::llamacpp::LlamaCppClient; use crate::ai::openrouter::OpenRouterClient; +use crate::ai::persona_chat::PersonaChatSession; use crate::ai::turn_registry::TurnRegistry; use crate::ai::{InsightGenerator, OllamaClient, SmsApiClient}; use crate::database::{ @@ -84,6 +85,9 @@ pub struct AppState { pub insight_generator: InsightGenerator, /// Chat continuation service. Hold an Arc so handlers can clone cheaply. pub insight_chat: Arc, + /// Open chat with persona service. Same shape as insight_chat but + /// anchored on (user_id, persona_id) instead of (library_id, file_path). + pub persona_chat_session: Arc, pub turn_registry: Arc, pub face_client: FaceClient, pub clip_client: ClipClient, @@ -133,6 +137,7 @@ impl AppState { sms_client: SmsApiClient, insight_generator: InsightGenerator, insight_chat: Arc, + persona_chat_session: Arc, turn_registry: Arc, preview_dao: Arc>>, face_client: FaceClient, @@ -194,6 +199,7 @@ impl AppState { sms_client, insight_generator, insight_chat, + persona_chat_session, turn_registry, face_client, clip_client, @@ -316,7 +322,7 @@ impl Default for AppState { tag_dao.clone(), face_dao.clone(), knowledge_dao, - persona_dao, + persona_dao.clone(), libraries_vec.clone(), ); @@ -330,6 +336,13 @@ impl Default for AppState { chat_locks, )); + // Open chat with persona: reuses the generator + persona DAO. + let persona_chat_session = Arc::new(PersonaChatSession::new( + Arc::new(insight_generator.clone()), + persona_dao.clone(), + Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())), + )); + // Turn registry for reconnectable chat turns. 5-minute timeout for // stale turns (background cleaner drops entries older than this). let timeout_secs: u64 = env::var("INSIGHT_CHAT_TURN_TIMEOUT_SECS") @@ -360,6 +373,7 @@ impl Default for AppState { sms_client, insight_generator, insight_chat, + persona_chat_session, turn_registry, preview_dao, face_client, @@ -528,7 +542,7 @@ impl AppState { tag_dao.clone(), face_dao.clone(), knowledge_dao, - persona_dao, + persona_dao.clone(), vec![test_lib], ); @@ -540,6 +554,13 @@ impl AppState { chat_locks, )); + // Open chat with persona (test). + let persona_chat_session = Arc::new(PersonaChatSession::new( + Arc::new(insight_generator.clone()), + persona_dao.clone(), + Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())), + )); + // Turn registry for test state. let turn_registry = Arc::new(TurnRegistry::new(300)); @@ -571,6 +592,7 @@ impl AppState { sms_client, insight_generator, insight_chat, + persona_chat_session, turn_registry, preview_dao, FaceClient::new(None), // disabled in test 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