Compare commits

9 Commits

Author SHA1 Message Date
Cameron Cordes 48a1b753f0 AI: add enable_thinking reasoning toggle plumbed to llama.cpp
New optional SamplingOverride forwarded to llama-server as
chat_template_kwargs.enable_thinking (gates Qwen3-style reasoning
blocks). None leaves the template default; other backends ignore it.
Wired through the agentic-insight and chat-turn request bodies/handlers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-17 18:14:44 -04:00
Cameron Cordes f2ab8d3740 Unified search: use ANY-mode tag matching, not ALL
ALL-mode over-constrains NL queries — the model maps several query words to
tags and few photos carry every one, zeroing the candidate set. Switch to
ANY (a photo matches if it has any named tag); the semantic CLIP ranking
provides precision within that pool. Exclude tags still filter out.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 02:25:24 -04:00
Cameron Cordes 6e5898e766 Unified search: rank within filtered set instead of pre-thresholding CLIP
When structured filters are present they're the constraint and CLIP only ranks
within the candidate set, so drop the global similarity threshold for that
case. Previously the 0.2 whole-library threshold ran BEFORE intersecting with
the filters, discarding filter-matching photos that scored just under it (e.g.
a 2022 beach photo at 0.18) — producing after_struct_filter=0 even when matches
existed. Plain semantic (no filters) keeps the user's threshold.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 02:20:06 -04:00
Cameron Cordes 6c315edacc clip_client: log encode_text failures (URL + status/body or network error)
The CLIP encode failure reason was only ever returned in the HTTP response
body, never logged server-side, making 502s from /photos/search opaque. Log
the underlying cause — network error to the URL, or the Apollo HTTP status +
response body — so CLIP-service problems are diagnosable from the ImageApi log.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 02:02:57 -04:00
Cameron Cordes 0a40e78528 Unified search: UNIFIED_SEARCH_MODEL env override for the translation step
Pin the NL->structured translation to a small, fast model that can stay
co-resident with CLIP (and the chat model) so it never evicts them on a tight
VRAM budget. Precedence: UNIFIED_SEARCH_MODEL env > client-selected model >
configured default. Logs the effective model (backend.model()) so model A/B
tests are visible. Documented in .env.example.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 01:58:48 -04:00
Cameron Cordes e56235acc5 Unified search: stage-by-stage logging to debug empty results
Log the translated query (semantic/tags/place/date/media + has_struct), the
tag-filter file count, candidate-row + allowed-hash counts, and the CLIP
considered/hits/after-filter counts. Pinpoints which stage drops results to
zero (over-extracted filter, tag path mismatch, Any/All over-constraint, or
CLIP threshold). info-level for now while debugging.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 01:29:21 -04:00
Cameron Cordes fcbd7e2733 Unified search: accept client model override (avoid model swapping)
Add an optional `model` query param to /photos/search/unified, passed into
resolve_backend's overrides. The client sends the user's currently-selected
local model so the translation step reuses an already-loaded model instead of
forcing a llama-swap eviction + cold start. Falls back to the configured
default when absent. Still local only (no hybrid).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 01:19:53 -04:00
Cameron Cordes e4c875f473 Unified NL search Phase 2: /photos/search/unified endpoint
Composes the two existing engines (Path A orchestration):
- Translate NL -> StructuredQuery via local LLM, respecting LLM_BACKEND
  (resolve_backend(Local) -> ollama or llama-swap; no hybrid).
- Forward-geocode the place name into a gps circle.
- Structured filters (tags/EXIF/geo/date/media) build a candidate set of EXIF
  rows; CLIP ranks within it, joined by content_hash. Degenerate cases match
  existing behavior: semantic-only -> plain CLIP; filters-only -> date-sorted.
- Echoes the interpreted query (incl. resolved place) for editable client chips.

Refactor: extracted reusable cores from clip_search (score_photos, resolve_hits,
parse_library_scope, score_error_response) shared by both endpoints. Removed the
Phase 1 allow-until-wired attributes now that nl_query + geo are consumed.

fmt + clippy clean; 23 backend tests pass (7 geo, 12 nl_query, 4 unified).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 01:03:43 -04:00
Cameron Cordes 50ed780844 Unified NL search Phase 1: NL→structured-query translator + forward geocoding
Foundation for the /photos/search/unified endpoint (Phase 2). Two new,
fully unit-tested pieces, not yet wired into a route (allow-until-wired,
mirroring llm_client.rs):

- ai/nl_query.rs: translate a free-text query into a StructuredQuery via one
  grounded LLM call. Two-stage — the model emits names/ISO dates, then a pure
  resolve step maps tag names against the real vocab and converts dates to
  unix seconds. Hallucinated (non-vocab) tags are surfaced in unmatched_tags
  rather than silently used as hard filters — the anti-noise guard. 12 tests.

- geo::forward_geocode + bbox_to_circle: resolve a place name to a circle via
  Nominatim /search, collapsing the bounding box to centroid + circumscribing
  radius so "Portland" and "Italy" both map onto the existing gps circle
  filter with no schema change. Radius is the max centroid-to-corner distance
  (corners aren't equidistant on a sphere). 4 tests.

fmt + clippy clean; 19 new tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 00:44:16 -04:00
21 changed files with 404 additions and 4635 deletions
Generated
+1 -1
View File
@@ -2051,7 +2051,7 @@ dependencies = [
[[package]]
name = "image-api"
version = "1.5.0"
version = "1.4.0"
dependencies = [
"actix",
"actix-cors",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "image-api"
version = "1.5.0"
version = "1.4.0"
authors = ["Cameron Cordes <cameronc.dev@gmail.com>"]
edition = "2024"
@@ -1 +0,0 @@
DROP TABLE IF EXISTS persona_chat_conversations;
@@ -1,22 +0,0 @@
-- 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<ChatMessage>` 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);
@@ -1,32 +0,0 @@
-- 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);
@@ -1,50 +0,0 @@
-- 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);
+32 -229
View File
@@ -1310,11 +1310,6 @@ pub struct ChatHistoryQuery {
pub path: String,
#[serde(default)]
pub library: Option<String>,
/// When set, load the branch anchored at this leaf ID instead of the
/// active branch. Allows the client to preview alternate conversation
/// paths without making them active.
#[serde(default)]
pub branch_id: Option<u64>,
}
#[derive(Debug, Serialize)]
@@ -1323,21 +1318,6 @@ pub struct ChatHistoryHttpResponse {
pub turn_count: usize,
pub model_version: String,
pub backend: String,
pub active_leaf_id: u64,
/// The branch leaf ID being viewed. Equals `active_leaf_id` when viewing
/// the active branch, or the `branch_id` query param for alternates.
pub viewing_branch_id: u64,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub fork_info: Vec<Option<ChatForkInfo>>,
}
#[derive(Debug, Clone, Serialize)]
pub struct ChatForkInfo {
pub position: usize,
pub total: usize,
/// The divergence tree node. Pass as `node_id` to the branches endpoint
/// for a list scoped to this exact fork.
pub node_id: u64,
}
#[derive(Debug, Serialize)]
@@ -1434,50 +1414,31 @@ pub async fn chat_history_handler(
.flatten()
.unwrap_or_else(|| app_state.primary_library());
match app_state
.insight_chat
.load_history(library.id, &query.path, query.branch_id)
{
Ok(view) => {
let fork_info: Vec<Option<ChatForkInfo>> = view
.fork_info
match app_state.insight_chat.load_history(library.id, &query.path) {
Ok(view) => HttpResponse::Ok().json(ChatHistoryHttpResponse {
messages: view
.messages
.into_iter()
.map(|f| {
f.map(|fi| ChatForkInfo {
position: fi.position,
total: fi.total,
node_id: fi.node_id,
})
.map(|m| RenderedHistoryMessage {
role: m.role,
content: m.content,
is_initial: m.is_initial,
tools: m
.tools
.into_iter()
.map(|t| HistoryToolInvocation {
name: t.name,
arguments: t.arguments,
result: t.result,
result_truncated: t.result_truncated,
})
.collect(),
})
.collect();
HttpResponse::Ok().json(ChatHistoryHttpResponse {
messages: view
.messages
.into_iter()
.map(|m| RenderedHistoryMessage {
role: m.role,
content: m.content,
is_initial: m.is_initial,
tools: m
.tools
.into_iter()
.map(|t| HistoryToolInvocation {
name: t.name,
arguments: t.arguments,
result: t.result,
result_truncated: t.result_truncated,
})
.collect(),
})
.collect(),
turn_count: view.turn_count,
model_version: view.model_version,
backend: view.backend,
active_leaf_id: view.active_leaf_id,
viewing_branch_id: view.viewing_branch_id,
fork_info,
})
}
.collect(),
turn_count: view.turn_count,
model_version: view.model_version,
backend: view.backend,
}),
Err(e) => {
let msg = format!("{}", e);
if msg.contains("no insight found") {
@@ -1491,118 +1452,6 @@ pub async fn chat_history_handler(
}
}
/// Query for GET /insights/chat/branches.
#[derive(Debug, Deserialize)]
pub struct ChatBranchesQuery {
pub path: String,
#[serde(default)]
pub library: Option<String>,
/// When set (from `ForkInfo.node_id`), scope the list to the sibling
/// branches diverging at this tree node instead of every leaf.
#[serde(default)]
pub node_id: Option<u64>,
/// The leaf the client is currently viewing. Scoped options in the
/// same subtree anchor to it so the client can identify its own branch.
/// Defaults to the active leaf when absent.
#[serde(default)]
pub viewing_branch_id: Option<u64>,
}
/// GET /insights/chat/branches — return the branch list for a photo's
/// conversation tree. Without `node_id`, each entry is a leaf node;
/// with `node_id`, entries are the position-ranked sibling branches at
/// that divergence point. Use an entry's `id` with `branch_id` on the
/// history endpoint (or the switch-branch endpoint) to load that path.
#[get("/insights/chat/branches")]
pub async fn chat_branches_handler(
_claims: Claims,
query: web::Query<ChatBranchesQuery>,
app_state: web::Data<AppState>,
) -> impl Responder {
let library = libraries::resolve_library_param_state(&app_state, query.library.as_deref())
.ok()
.flatten()
.unwrap_or_else(|| app_state.primary_library());
let (branches, active_leaf_id) = match app_state.insight_chat.get_branches(
library.id,
&query.path,
query.node_id,
query.viewing_branch_id,
) {
Ok(result) => result,
Err(e) => {
let msg = format!("{}", e);
if msg.contains("no insight found") {
return HttpResponse::NotFound().json(serde_json::json!({ "error": msg }));
} else if msg.contains("no chat history") {
return HttpResponse::Conflict().json(serde_json::json!({ "error": msg }));
} else if msg.contains("not found in tree") {
return HttpResponse::BadRequest().json(serde_json::json!({ "error": msg }));
} else {
return HttpResponse::InternalServerError()
.json(serde_json::json!({ "error": msg }));
}
}
};
HttpResponse::Ok().json(serde_json::json!({
"branches": branches,
"active_leaf_id": active_leaf_id,
}))
}
/// POST /insights/chat/switch-branch — switch the active branch to a
/// different conversation path. The new branch becomes the active
/// conversation; the previous active branch becomes a regular fork.
#[post("/insights/chat/switch-branch")]
pub async fn chat_switch_branch_handler(
_claims: Claims,
request: web::Json<ChatSwitchBranchRequest>,
app_state: web::Data<AppState>,
) -> impl Responder {
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)
}));
}
};
match app_state
.insight_chat
.switch_branch(library.id, &request.file_path, request.branch_id)
.await
{
Ok(()) => HttpResponse::Ok().json(serde_json::json!({ "success": true })),
Err(e) => {
let msg = format!("{}", e);
log::error!("Switch branch failed: {}", msg);
if msg.contains("no insight 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") {
HttpResponse::BadRequest().json(serde_json::json!({ "error": msg }))
} else {
HttpResponse::InternalServerError().json(serde_json::json!({ "error": msg }))
}
}
}
}
#[derive(Debug, Deserialize)]
pub struct ChatSwitchBranchRequest {
pub file_path: String,
#[serde(default)]
pub library: Option<String>,
/// The leaf node ID to switch to. Must be a valid leaf in the tree.
pub branch_id: u64,
}
/// POST /insights/chat/stream — streaming variant of /insights/chat.
/// Returns `text/event-stream` with one event per chat stream event.
#[post("/insights/chat/stream")]
@@ -1819,19 +1668,6 @@ pub async fn turn_replay_handler(
path: web::Path<String>,
query: web::Query<ReplayQuery>,
app_state: web::Data<AppState>,
) -> 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<String>,
query: web::Query<ReplayQuery>,
app_state: web::Data<AppState>,
) -> HttpResponse {
use crate::ai::turn_registry::ReplayOutcome;
@@ -1962,31 +1798,15 @@ pub(crate) async fn turn_replay_impl(
}
fn render_turn_info_frame(info: &crate::ai::turn_registry::TurnInfo) -> 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());
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());
format!("event: turn_info\ndata: {}\n\n", data)
}
@@ -1996,16 +1816,6 @@ pub async fn cancel_turn_handler(
http_request: HttpRequest,
path: web::Path<String>,
app_state: web::Data<AppState>,
) -> 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<String>,
app_state: web::Data<AppState>,
) -> impl Responder {
let turn_id = path.into_inner();
@@ -2056,13 +1866,6 @@ pub(crate) async fn cancel_turn_impl(
}))
}
// 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};
+329 -733
View File
File diff suppressed because it is too large Load Diff
+10 -15
View File
@@ -26,7 +26,7 @@ use crate::libraries::Library;
use crate::memories::extract_date_from_filename;
use crate::otel::global_tracer;
use crate::tags::TagDao;
use crate::utils::{earliest_fs_time, normalize_path, retry_with_backoff};
use crate::utils::{earliest_fs_time, normalize_path};
/// Max location records rendered by `tool_get_location_history`. The DAO
/// query is range-bounded, not limited, so the tool caps the rendered list
@@ -1539,13 +1539,10 @@ impl InsightGenerator {
eval_count: None,
};
let dao = self.insight_dao.clone();
let result = retry_with_backoff("store_insight", 3, || {
let cx = opentelemetry::Context::new();
let mut d = dao.lock().expect("Unable to lock InsightDao");
d.store_insight(&cx, insight.clone())
})
.map_err(|e| anyhow::anyhow!("Failed to store insight: {:?}", e));
let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
let result = dao
.store_insight(&insight_cx, insight)
.map_err(|e| anyhow::anyhow!("Failed to store insight: {:?}", e));
match &result {
Ok(_) => {
@@ -4468,13 +4465,11 @@ Return ONLY the summary, nothing else."#,
eval_count: last_eval_count,
};
let dao = self.insight_dao.clone();
let stored = retry_with_backoff("store_insight", 3, || {
let cx = opentelemetry::Context::new();
let mut d = dao.lock().expect("Unable to lock InsightDao");
d.store_insight(&cx, insight.clone())
})
.map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e));
let stored = {
let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
dao.store_insight(&insight_cx, insight)
.map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e))
};
match &stored {
Ok(_) => {
-566
View File
@@ -171,292 +171,6 @@ pub struct ModelCapabilities {
pub has_tool_calling: bool,
}
/// A single node in the conversation tree. Wraps ChatMessage with structural
/// metadata so the tree can be persisted as JSON in `training_messages`.
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct StoredChatNode {
pub id: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub parent_id: Option<u64>,
pub message: ChatMessage,
}
/// Top-level container for tree-based chat history. Stored as JSON in the
/// `training_messages` column. Backward-compatible reader detects whether
/// the stored value is a flat array (old format) or this object (new format).
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct ChatHistoryStore {
pub nodes: Vec<StoredChatNode>,
pub active_leaf_id: u64,
}
impl ChatHistoryStore {
/// Create an empty store. Test-only: production stores are always built
/// from `from_flat_array` or deserialized from `training_messages`.
#[cfg(test)]
pub fn empty() -> Self {
Self {
nodes: Vec::new(),
active_leaf_id: 0,
}
}
/// Convert a flat `Vec<ChatMessage>` (old format) into a tree.
/// Each message becomes a node chained by parent_id.
pub fn from_flat_array(messages: Vec<ChatMessage>) -> Self {
let mut nodes = Vec::with_capacity(messages.len());
let mut prev_id: Option<u64> = None;
for (i, msg) in messages.into_iter().enumerate() {
let id = (i + 1) as u64;
nodes.push(StoredChatNode {
id,
parent_id: prev_id,
message: msg,
});
prev_id = Some(id);
}
let active_leaf_id = prev_id.unwrap_or(0);
Self {
nodes,
active_leaf_id,
}
}
/// Allocate the next node ID (max existing + 1, or 1 if empty).
pub fn next_id(&self) -> u64 {
self.nodes.iter().map(|n| n.id).max().unwrap_or(0) + 1
}
/// Append a new node as a child of the given parent. Returns the new node's id.
pub fn append_node(&mut self, parent_id: Option<u64>, message: ChatMessage) -> u64 {
let id = self.next_id();
self.nodes.push(StoredChatNode {
id,
parent_id,
message,
});
id
}
/// Traverse from root to the given leaf_id, returning the path in order.
/// Returns None if the leaf_id doesn't exist.
pub fn path_to_leaf(&self, leaf_id: u64) -> Option<Vec<&StoredChatNode>> {
// Build a map from id to node for fast lookup.
let node_map: std::collections::HashMap<u64, usize> = self
.nodes
.iter()
.enumerate()
.map(|(i, n)| (n.id, i))
.collect();
let &_leaf_idx = node_map.get(&leaf_id)?;
let mut path_indices = Vec::new();
let mut current_id = leaf_id;
while let Some(&idx) = node_map.get(&current_id) {
path_indices.push(idx);
match self.nodes[idx].parent_id {
Some(pid) => current_id = pid,
None => break,
}
}
path_indices.reverse();
Some(path_indices.into_iter().map(|i| &self.nodes[i]).collect())
}
/// All direct children of a given node, ordered by id (chronological).
pub fn children_of(&self, node_id: u64) -> Vec<&StoredChatNode> {
self.nodes
.iter()
.filter(|n| n.parent_id == Some(node_id))
.collect()
}
/// For a given node, return (position, total) where:
/// - total = number of children of the parent of this node (i.e., siblings + self)
/// - position = 1-based rank of this node among siblings, ordered by id
///
/// Returns None if the node has no parent (root) or the parent has 0 or 1 child.
pub fn fork_at_node(&self, node_id: u64) -> Option<(usize, usize)> {
let node = self.nodes.iter().find(|n| n.id == node_id)?;
let parent_id = node.parent_id?;
let children = self.children_of(parent_id);
let total = children.len();
if total <= 1 {
return None;
}
let position = children.iter().position(|c| c.id == node_id)? + 1;
Some((position, total))
}
/// All leaf node IDs within the subtree rooted at `node_id` (the node
/// itself when it has no children). Empty when the node doesn't exist.
pub fn descendant_leaves(&self, node_id: u64) -> Vec<u64> {
if !self.nodes.iter().any(|n| n.id == node_id) {
return Vec::new();
}
let mut result = Vec::new();
let mut stack = vec![node_id];
while let Some(id) = stack.pop() {
let children = self.children_of(id);
if children.is_empty() {
result.push(id);
} else {
stack.extend(children.iter().map(|c| c.id));
}
}
result
}
/// Branch options at a divergence point: one entry per direct child of
/// `fork_node_id`, ordered by id (chronological — matches the stable
/// position rank shown in "X/Y" fork chips).
///
/// Each entry's `id` is the leaf to switch to for that branch: the
/// `preferred_leaf` when it lives in that child's subtree (so the
/// caller's current view maps onto its own option and can be filtered
/// out client-side), otherwise the most recent (max-id) descendant leaf.
///
/// Snippets preview the first visible message that *differs* between
/// the branches, not blindly the first child: Rewind & Regenerate
/// resends the identical question, so the first child is often shared
/// across siblings and would make every option read the same.
pub fn branch_options_at(&self, fork_node_id: u64, preferred_leaf: u64) -> Vec<BranchLeafInfo> {
struct Candidate {
rep: u64,
message_count: usize,
/// User/assistant contents after the divergence, tool
/// scaffolding (tool results, empty dispatch turns) skipped.
visible: Vec<String>,
}
let candidates: Vec<Candidate> = self
.children_of(fork_node_id)
.iter()
.map(|child| {
let leaves = self.descendant_leaves(child.id);
let rep = if leaves.contains(&preferred_leaf) {
preferred_leaf
} else {
leaves.iter().copied().max().unwrap_or(child.id)
};
let path = self.path_to_leaf(rep).unwrap_or_default();
let message_count = path.len();
let visible: Vec<String> = path
.iter()
.skip_while(|n| n.id != child.id)
.filter(|n| {
(n.message.role == "user" || n.message.role == "assistant")
&& !n.message.content.trim().is_empty()
})
.map(|n| n.message.content.clone())
.collect();
Candidate {
rep,
message_count,
visible,
}
})
.collect();
// Walk forward past positions where every branch has the same
// content; stop at the first divergence (or when any branch runs
// out — a length difference is itself distinguishing).
let mut snippet_idx = 0usize;
loop {
let contents: Vec<Option<&String>> = candidates
.iter()
.map(|c| c.visible.get(snippet_idx))
.collect();
let all_present_and_equal =
contents.iter().all(|c| c.is_some()) && contents.windows(2).all(|w| w[0] == w[1]);
if !all_present_and_equal {
break;
}
snippet_idx += 1;
}
candidates
.into_iter()
.enumerate()
.map(|(i, c)| {
// A branch exhausted before the divergence index falls back
// to its own last message so it still shows *something*.
let snippet = c
.visible
.get(snippet_idx)
.or_else(|| c.visible.last())
.map(|s| s.chars().take(100).collect::<String>())
.unwrap_or_default();
BranchLeafInfo {
id: c.rep,
snippet,
message_count: c.message_count,
position: Some(i + 1),
}
})
.collect()
}
/// All leaf node IDs (nodes with no children).
pub fn leaves(&self) -> Vec<u64> {
let child_ids: std::collections::HashSet<u64> =
self.nodes.iter().filter_map(|n| n.parent_id).collect();
self.nodes
.iter()
.filter(|n| !child_ids.contains(&n.id))
.map(|n| n.id)
.collect()
}
/// 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?;
self.nodes.iter().find(|n| n.id == parent_id)
}
/// Return all leaf nodes with metadata: id, snippet (first 100 chars of
/// the first user-visible message after the divergence point), and
/// message_count (number of nodes in the path from root to leaf).
pub fn leaves_with_info(&self) -> Vec<BranchLeafInfo> {
let leaf_ids = self.leaves();
leaf_ids
.into_iter()
.map(|leaf_id| {
let path = self.path_to_leaf(leaf_id).unwrap_or_default();
let message_count = path.len();
// Snippet: first 100 chars of the first user/assistant message
// (skip system messages).
let snippet = path
.iter()
.find(|n| n.message.role != "system")
.map(|n| n.message.content.chars().take(100).collect::<String>())
.unwrap_or_default();
BranchLeafInfo {
id: leaf_id,
snippet,
message_count,
position: None,
}
})
.collect()
}
}
/// Metadata about a branch leaf, returned by the branches endpoint.
#[derive(Serialize, Clone, Debug)]
pub struct BranchLeafInfo {
pub id: u64,
#[serde(skip_serializing_if = "String::is_empty")]
pub snippet: String,
pub message_count: usize,
/// 1-based sibling rank at the divergence point. Only present for
/// scoped queries (`branch_options_at`); tree-wide leaf listings have
/// no single divergence to rank against.
#[serde(skip_serializing_if = "Option::is_none")]
pub position: Option<usize>,
}
/// Strip a leading `<think>…</think>` reasoning block from model output.
///
/// Thinking models sometimes emit chain-of-thought inside think tags before
@@ -507,284 +221,4 @@ mod tests {
let raw = "<think>thinking forever";
assert_eq!(strip_think_blocks(raw), raw);
}
// ── ChatHistoryStore tree traversal tests ──────────────────────────
fn mk_msg(role: &str, content: &str) -> ChatMessage {
match role {
"system" => ChatMessage::system(content),
"user" => ChatMessage::user(content),
_ => ChatMessage {
role: role.to_string(),
content: content.to_string(),
tool_calls: None,
images: None,
},
}
}
#[test]
fn store_from_flat_array_chains_parent_ids() {
let msgs = vec![
mk_msg("system", "sys"),
mk_msg("user", "hello"),
mk_msg("assistant", "hi there"),
];
let store = ChatHistoryStore::from_flat_array(msgs);
assert_eq!(store.nodes.len(), 3);
assert_eq!(store.nodes[0].id, 1);
assert_eq!(store.nodes[0].parent_id, None);
assert_eq!(store.nodes[1].parent_id, Some(1));
assert_eq!(store.nodes[2].parent_id, Some(2));
assert_eq!(store.active_leaf_id, 3);
}
#[test]
fn store_path_to_leaf_returns_ordered_path() {
let mut store = ChatHistoryStore::empty();
let n1 = store.append_node(None, mk_msg("user", "u1"));
let n2 = store.append_node(Some(n1), mk_msg("assistant", "a1"));
let n3 = store.append_node(Some(n2), mk_msg("user", "u2"));
let path = store.path_to_leaf(n3).unwrap();
assert_eq!(path.len(), 3);
assert_eq!(path[0].id, n1);
assert_eq!(path[1].id, n2);
assert_eq!(path[2].id, n3);
}
#[test]
fn store_path_to_leaf_returns_none_for_missing_id() {
let store = ChatHistoryStore::empty();
assert!(store.path_to_leaf(999).is_none());
}
#[test]
fn store_children_of_returns_direct_children() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let c1 = store.append_node(Some(root), mk_msg("assistant", "c1"));
let c2 = store.append_node(Some(root), mk_msg("user", "c2"));
let _d1 = store.append_node(Some(c1), mk_msg("assistant", "d1"));
let children = store.children_of(root);
assert_eq!(children.len(), 2);
assert_eq!(children[0].id, c1);
assert_eq!(children[1].id, c2);
}
#[test]
fn store_fork_at_node_detects_fork() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let c1 = store.append_node(Some(root), mk_msg("assistant", "c1"));
let c2 = store.append_node(Some(root), mk_msg("user", "c2"));
let c3 = store.append_node(Some(root), mk_msg("user", "c3"));
// c1 is position 1 of 3 siblings
assert_eq!(store.fork_at_node(c1), Some((1, 3)));
assert_eq!(store.fork_at_node(c2), Some((2, 3)));
assert_eq!(store.fork_at_node(c3), Some((3, 3)));
// root has no parent, so no fork
assert!(store.fork_at_node(root).is_none());
}
#[test]
fn store_fork_at_node_returns_none_for_single_child() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let _only = store.append_node(Some(root), mk_msg("assistant", "only"));
assert!(store.fork_at_node(2).is_none());
}
#[test]
fn store_leaves_returns_leaf_ids() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let c1 = store.append_node(Some(root), mk_msg("assistant", "c1"));
let c2 = store.append_node(Some(root), mk_msg("user", "c2"));
let _d1 = store.append_node(Some(c1), mk_msg("assistant", "d1"));
let leaves = store.leaves();
assert_eq!(leaves.len(), 2);
assert!(leaves.contains(&c2));
assert!(leaves.contains(&_d1));
assert!(!leaves.contains(&root));
assert!(!leaves.contains(&c1));
}
#[test]
fn store_parent_of_returns_parent() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let child = store.append_node(Some(root), mk_msg("assistant", "child"));
let parent = store.parent_of(child).unwrap();
assert_eq!(parent.id, root);
assert!(store.parent_of(root).is_none());
}
#[test]
fn store_next_id_increments() {
let mut store = ChatHistoryStore::empty();
assert_eq!(store.next_id(), 1);
store.append_node(None, mk_msg("user", "a"));
assert_eq!(store.next_id(), 2);
store.append_node(Some(1), mk_msg("assistant", "b"));
assert_eq!(store.next_id(), 3);
}
#[test]
fn store_descendant_leaves_walks_subtree() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
let c1 = store.append_node(Some(root), mk_msg("assistant", "c1"));
let c2 = store.append_node(Some(root), mk_msg("user", "c2"));
let d1 = store.append_node(Some(c1), mk_msg("assistant", "d1"));
let d2 = store.append_node(Some(c1), mk_msg("assistant", "d2"));
// c1's subtree has two leaves; c2 is its own leaf; root sees all.
let mut c1_leaves = store.descendant_leaves(c1);
c1_leaves.sort_unstable();
assert_eq!(c1_leaves, vec![d1, d2]);
assert_eq!(store.descendant_leaves(c2), vec![c2]);
let mut all = store.descendant_leaves(root);
all.sort_unstable();
assert_eq!(all, vec![c2, d1, d2]);
assert!(store.descendant_leaves(999).is_empty());
}
#[test]
fn branch_options_ranks_children_and_prefers_viewing_leaf() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
// First branch: two leaves (older + newer).
let c1 = store.append_node(Some(root), mk_msg("assistant", "reply one"));
let old_leaf = store.append_node(Some(c1), mk_msg("user", "old follow-up"));
let new_leaf = store.append_node(Some(c1), mk_msg("user", "new follow-up"));
// Second branch: single leaf.
let c2 = store.append_node(Some(root), mk_msg("assistant", "reply two"));
// Viewing the *older* leaf of branch 1: its option must anchor to
// that exact leaf, not the max-id one, so the client's
// "hide the branch I'm on" filter matches.
let opts = store.branch_options_at(root, old_leaf);
assert_eq!(opts.len(), 2);
assert_eq!(opts[0].id, old_leaf);
assert_eq!(opts[0].position, Some(1));
assert_eq!(opts[0].snippet, "reply one");
assert_eq!(opts[1].id, c2);
assert_eq!(opts[1].position, Some(2));
// Viewing branch 2: branch 1 falls back to its most recent leaf.
let opts = store.branch_options_at(root, c2);
assert_eq!(opts[0].id, new_leaf);
assert_eq!(opts[1].id, c2);
}
#[test]
fn branch_options_snippet_skips_shared_resent_question() {
// Rewind & Regenerate resends the identical question, so both
// branches' first child matches — the preview must advance to the
// first *differing* message (the assistant replies) or every
// option would read the same.
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("assistant", "prior reply"));
let q1 = store.append_node(Some(root), mk_msg("user", "same question"));
let _a1 = store.append_node(Some(q1), mk_msg("assistant", "first answer"));
let q2 = store.append_node(Some(root), mk_msg("user", "same question"));
let a2 = store.append_node(Some(q2), mk_msg("assistant", "second answer"));
let opts = store.branch_options_at(root, a2);
assert_eq!(opts.len(), 2);
assert_eq!(opts[0].snippet, "first answer");
assert_eq!(opts[1].snippet, "second answer");
}
#[test]
fn branch_options_snippet_falls_back_when_branch_exhausted() {
// One branch is just the shared question (no reply yet); the other
// continues past it. The shorter branch falls back to its own last
// message rather than showing an empty preview.
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("assistant", "prior reply"));
let _q1 = store.append_node(Some(root), mk_msg("user", "same question"));
let q2 = store.append_node(Some(root), mk_msg("user", "same question"));
let a2 = store.append_node(Some(q2), mk_msg("assistant", "second answer"));
let opts = store.branch_options_at(root, a2);
assert_eq!(opts[0].snippet, "same question");
assert_eq!(opts[1].snippet, "second answer");
}
#[test]
fn branch_options_snippet_skips_tool_scaffolding() {
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "root"));
// Branch whose diverging child is an empty tool-dispatch turn.
let dispatch = store.append_node(
Some(root),
ChatMessage {
role: "assistant".to_string(),
content: String::new(),
tool_calls: None,
images: None,
},
);
let tool = store.append_node(Some(dispatch), mk_msg("tool", "raw tool output"));
let reply = store.append_node(Some(tool), mk_msg("assistant", "final answer"));
let _other = store.append_node(Some(root), mk_msg("assistant", "direct reply"));
let opts = store.branch_options_at(root, reply);
assert_eq!(opts[0].snippet, "final answer");
assert_eq!(opts[1].snippet, "direct reply");
}
#[test]
fn store_append_node_sets_parent() {
let mut store = ChatHistoryStore::empty();
let parent = store.append_node(None, mk_msg("user", "parent"));
let child = store.append_node(Some(parent), mk_msg("assistant", "child"));
assert_eq!(store.nodes.len(), 2);
assert_eq!(store.nodes[1].parent_id, Some(parent));
assert_eq!(store.nodes[1].id, child);
}
#[test]
fn branch_options_handles_nested_forks() {
// Regression: verify sub-fork display when there are multiple forks.
// Structure:
// root (user "q1")
// ├── a1 (assistant "answer 1")
// └── a2 (assistant "answer 2")
// ├── q2 (user "q2")
// │ ├── r1 (assistant "reply 1")
// │ └── r2 (assistant "reply 2") <- active leaf
//
// branch_options_at(root) should return [a1, a2] (2 branches)
// branch_options_at(q2) should return [r1, r2] (2 sub-branches)
let mut store = ChatHistoryStore::empty();
let root = store.append_node(None, mk_msg("user", "q1"));
let a1 = store.append_node(Some(root), mk_msg("assistant", "answer 1"));
let a2 = store.append_node(Some(root), mk_msg("assistant", "answer 2"));
let q2 = store.append_node(Some(a2), mk_msg("user", "q2"));
let r1 = store.append_node(Some(q2), mk_msg("assistant", "reply 1"));
let r2 = store.append_node(Some(q2), mk_msg("assistant", "reply 2"));
store.active_leaf_id = r2;
// Root fork: should show [a1, a2's subtree]
let opts = store.branch_options_at(root, r2);
assert_eq!(opts.len(), 2);
assert_eq!(opts[0].id, a1);
assert_eq!(opts[0].position, Some(1));
assert_eq!(opts[0].snippet, "answer 1");
// a2's subtree includes r2, so the rep should be r2 (preferred leaf)
assert_eq!(opts[1].id, r2);
assert_eq!(opts[1].position, Some(2));
assert_eq!(opts[1].snippet, "answer 2");
// Sub-fork at q2: should show [r1, r2]
let opts = store.branch_options_at(q2, r2);
assert_eq!(opts.len(), 2);
assert_eq!(opts[0].id, r1);
assert_eq!(opts[0].position, Some(1));
assert_eq!(opts[0].snippet, "reply 1");
assert_eq!(opts[1].id, r2);
assert_eq!(opts[1].position, Some(2));
assert_eq!(opts[1].snippet, "reply 2");
}
}
+6 -13
View File
@@ -13,7 +13,6 @@ 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;
@@ -26,12 +25,12 @@ pub use daily_summary_job::{
generate_daily_summaries, strip_summary_boilerplate,
};
pub use handlers::{
cancel_generation_handler, cancel_turn_handler, chat_branches_handler, chat_history_handler,
chat_rewind_handler, chat_stream_handler, chat_switch_branch_handler, chat_turn_handler,
delete_insight_handler, export_training_data_handler, generate_agentic_insight_handler,
generate_insight_handler, generation_status_handler, get_all_insights_handler,
get_available_models_handler, get_insight_handler, get_insight_history_handler,
get_openrouter_models_handler, rate_insight_handler, turn_async_handler, turn_replay_handler,
cancel_generation_handler, cancel_turn_handler, chat_history_handler, chat_rewind_handler,
chat_stream_handler, chat_turn_handler, delete_insight_handler, export_training_data_handler,
generate_agentic_insight_handler, generate_insight_handler, generation_status_handler,
get_all_insights_handler, get_available_models_handler, get_insight_handler,
get_insight_history_handler, get_openrouter_models_handler, rate_insight_handler,
turn_async_handler, turn_replay_handler,
};
pub use insight_generator::InsightGenerator;
pub use llamacpp::LlamaCppClient;
@@ -39,12 +38,6 @@ 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;
File diff suppressed because it is too large Load Diff
-106
View File
@@ -52,10 +52,6 @@ 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<String>,
/// `"insight"` (file-anchored chat) | `"persona"` (open chat).
pub scope: String,
pub status: TurnStatus,
pub total_events_pushed: u32,
pub buffered_count: u32,
@@ -82,20 +78,8 @@ 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<String>,
/// `"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<Vec<ChatStreamEvent>>,
@@ -116,35 +100,10 @@ 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<String>,
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),
@@ -211,8 +170,6 @@ 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,
@@ -788,67 +745,4 @@ 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());
}
}
+1 -1
View File
@@ -192,7 +192,7 @@ pub struct ImageExif {
pub clip_model_version: Option<String>,
}
#[derive(Insertable, Clone)]
#[derive(Insertable)]
#[diesel(table_name = photo_insights)]
pub struct InsertPhotoInsight {
pub library_id: i32,
+6 -494
View File
@@ -10,51 +10,6 @@ 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).
@@ -125,76 +80,6 @@ pub trait PersonaDao: Sync + Send {
user_id: i32,
personas: &[ImportPersona],
) -> Result<usize, DbError>;
// ── 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<Option<PersonaChatRow>, 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<Vec<PersonaChatRow>, 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<String, DbError>;
/// 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<usize, DbError>;
/// 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 {
@@ -366,28 +251,12 @@ impl PersonaDao for SqlitePersonaDao {
pid: &str,
) -> Result<bool, DbError> {
trace_db_call(cx, "delete", "delete_persona", |_span| {
use schema::personas::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
// 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)
})
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)
})
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
@@ -427,168 +296,8 @@ 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<Option<PersonaChatRow>, 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<Vec<PersonaChatRow>, 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<String, DbError> {
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<usize, DbError> {
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::*;
@@ -735,201 +444,4 @@ 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<String> = 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<String> = 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());
}
}
-14
View File
@@ -171,19 +171,6 @@ 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,
@@ -358,7 +345,6 @@ diesel::allow_tables_to_appear_in_same_query!(
insight_generation_jobs,
libraries,
location_history,
persona_chat_conversations,
personas,
persons,
photo_insights,
-12
View File
@@ -377,21 +377,9 @@ fn main() -> std::io::Result<()> {
.service(ai::chat_stream_handler)
.service(ai::chat_history_handler)
.service(ai::chat_rewind_handler)
.service(ai::chat_branches_handler)
.service(ai::chat_switch_branch_handler)
.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)
+2 -24
View File
@@ -4,7 +4,6 @@ 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::{
@@ -85,9 +84,6 @@ pub struct AppState {
pub insight_generator: InsightGenerator,
/// Chat continuation service. Hold an Arc so handlers can clone cheaply.
pub insight_chat: Arc<InsightChatService>,
/// 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<PersonaChatSession>,
pub turn_registry: Arc<TurnRegistry>,
pub face_client: FaceClient,
pub clip_client: ClipClient,
@@ -137,7 +133,6 @@ impl AppState {
sms_client: SmsApiClient,
insight_generator: InsightGenerator,
insight_chat: Arc<InsightChatService>,
persona_chat_session: Arc<PersonaChatSession>,
turn_registry: Arc<TurnRegistry>,
preview_dao: Arc<Mutex<Box<dyn PreviewDao>>>,
face_client: FaceClient,
@@ -199,7 +194,6 @@ impl AppState {
sms_client,
insight_generator,
insight_chat,
persona_chat_session,
turn_registry,
face_client,
clip_client,
@@ -322,7 +316,7 @@ impl Default for AppState {
tag_dao.clone(),
face_dao.clone(),
knowledge_dao,
persona_dao.clone(),
persona_dao,
libraries_vec.clone(),
);
@@ -336,13 +330,6 @@ 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")
@@ -373,7 +360,6 @@ impl Default for AppState {
sms_client,
insight_generator,
insight_chat,
persona_chat_session,
turn_registry,
preview_dao,
face_client,
@@ -542,7 +528,7 @@ impl AppState {
tag_dao.clone(),
face_dao.clone(),
knowledge_dao,
persona_dao.clone(),
persona_dao,
vec![test_lib],
);
@@ -554,13 +540,6 @@ 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));
@@ -592,7 +571,6 @@ impl AppState {
sms_client,
insight_generator,
insight_chat,
persona_chat_session,
turn_registry,
preview_dao,
FaceClient::new(None), // disabled in test
+10 -45
View File
@@ -98,7 +98,7 @@ pub fn generate_image_thumbnail(src: &Path, thumb_path: &Path) -> std::io::Resul
}
if file_types::needs_ffmpeg_thumbnail(src) {
return generate_image_thumbnail_ffmpeg(src, thumb_path, orientation);
return generate_image_thumbnail_ffmpeg(src, thumb_path);
}
let img = image::open(src).map_err(|e| {
@@ -140,7 +140,7 @@ pub fn generate_large_preview(src: &Path, dest: &Path) -> std::io::Result<()> {
}
if file_types::needs_ffmpeg_thumbnail(src) {
return generate_large_preview_ffmpeg(src, dest, orientation);
return generate_large_preview_ffmpeg(src, dest);
}
let img = image::open(src).map_err(|e| {
@@ -175,30 +175,14 @@ 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<()> {
// 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
// regardless of orientation, mirroring image::thumbnail.
let rotation = match orientation {
2 => "hflip",
3 => "transpose=2",
4 => "vflip",
5 => "transpose=0,hflip",
6 => "transpose=0",
7 => "transpose=1,hflip",
8 => "transpose=1",
_ => "",
};
let scale_expr = format!(
fn generate_large_preview_ffmpeg(src: &Path, dest: &Path) -> std::io::Result<()> {
// scale=W:-1 with force_original_aspect_ratio=decrease + the min(iw,W)
// trick caps the long edge regardless of orientation, mirroring what
// image::thumbnail does for the non-ffmpeg branch.
let vf = format!(
"scale='if(gt(iw,ih),min(iw,{cap}),-1)':'if(gt(iw,ih),-1,min(ih,{cap}))'",
cap = LARGE_PREVIEW_MAX_DIM
);
let vf = if rotation.is_empty() {
format!("{},colorspace=bt709", scale_expr)
} else {
format!("{},{},colorspace=bt709", rotation, scale_expr)
};
let output = Command::new("ffmpeg")
.arg("-y")
.arg("-i")
@@ -248,7 +232,7 @@ pub fn generate_xlarge_preview(src: &Path, dest: &Path) -> std::io::Result<()> {
}
if file_types::needs_ffmpeg_thumbnail(src) {
return generate_xlarge_preview_ffmpeg(src, dest, orientation);
return generate_xlarge_preview_ffmpeg(src, dest);
}
let img = image::open(src).map_err(|e| {
@@ -276,30 +260,11 @@ fn encode_xlarge_jpeg(img: image::DynamicImage, dest: &Path) -> std::io::Result<
Ok(())
}
fn generate_xlarge_preview_ffmpeg(
src: &Path,
dest: &Path,
orientation: i32,
) -> std::io::Result<()> {
let rotation = match orientation {
2 => "hflip",
3 => "transpose=2",
4 => "vflip",
5 => "transpose=0,hflip",
6 => "transpose=0",
7 => "transpose=1,hflip",
8 => "transpose=1",
_ => "",
};
let scale_expr = format!(
fn generate_xlarge_preview_ffmpeg(src: &Path, dest: &Path) -> std::io::Result<()> {
let vf = format!(
"scale='if(gt(iw,ih),min(iw,{cap}),-1)':'if(gt(iw,ih),-1,min(ih,{cap}))'",
cap = XLARGE_PREVIEW_MAX_DIM
);
let vf = if rotation.is_empty() {
format!("{},colorspace=bt709", scale_expr)
} else {
format!("{},{},colorspace=bt709", rotation, scale_expr)
};
let output = Command::new("ffmpeg")
.arg("-y")
.arg("-i")
+1 -55
View File
@@ -1,58 +1,4 @@
use rand::Rng;
use std::time::{Duration, SystemTime};
/// Retry a fallible operation with exponential backoff + jitter.
///
/// Runs the closure immediately, then retries up to `max_retries` times on
/// failure. Each retry sleeps for `base_delay * 2^attempt` plus a uniform
/// jitter of ±25%. Non-final errors are logged at `debug` with the label;
/// the final error is logged at `error`.
///
/// # Examples
/// ```
/// use image_api::utils::retry_with_backoff;
///
/// let result = retry_with_backoff("my-op", 3, || {
/// // something that may fail transiently
/// Ok::<_, anyhow::Error>(42)
/// });
/// ```
pub fn retry_with_backoff<F, T, E>(label: &str, max_retries: u32, mut op: F) -> Result<T, E>
where
F: FnMut() -> Result<T, E>,
E: std::fmt::Debug,
{
let mut last_err = match op() {
Ok(v) => return Ok(v),
Err(e) => e,
};
for attempt in 1..=max_retries {
let base = Duration::from_millis(100).saturating_mul(1_u32.pow(attempt - 1));
let jitter_range = base.as_millis() as f64 * 0.25;
let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range) as u128;
let delay = base.as_millis() as i128 + jitter as i128;
std::thread::sleep(Duration::from_millis(delay.max(0) as u64));
log::debug!(
"{}: attempt {}/{} failed ({:?}), retrying in {}ms",
label,
attempt,
max_retries,
last_err,
delay.max(0)
);
match op() {
Ok(v) => return Ok(v),
Err(e) => last_err = e,
}
}
log::error!(
"{}: all {} retries exhausted: {:?}",
label,
max_retries,
last_err
);
Err(last_err)
}
use std::time::SystemTime;
/// Normalize a file path to use forward slashes for cross-platform consistency
/// This ensures paths stored in the database always use `/` regardless of OS
+5 -34
View File
@@ -90,39 +90,10 @@ pub fn generate_video_thumbnail(path: &Path, destination: &Path) -> std::io::Res
Ok(())
}
/// Build the ffmpeg filter chain for image thumbnails: rotation (from EXIF
/// orientation) + scale + color-space conversion. HEIC sources use Display P3
/// primaries; without `colorspace=bt709` the mjpeg encoder treats P3 values
/// as sRGB, producing warm/oversaturated output.
fn build_image_thumb_filter(orientation: i32, scale_w: u32) -> String {
let rotation = match orientation {
2 => "hflip",
3 => "transpose=2",
4 => "vflip",
5 => "transpose=0,hflip",
6 => "transpose=0",
7 => "transpose=1,hflip",
8 => "transpose=1",
_ => "", // orientation 1 or unknown — no rotation needed
};
if rotation.is_empty() {
format!("scale={}:{{-1}},colorspace=bt709", scale_w)
} else {
format!("{},scale={}:{{-1}},colorspace=bt709", rotation, scale_w)
}
}
/// Use ffmpeg to extract a thumbnail from formats the `image` crate can't
/// decode (HEIC/HEIF, RAW: NEF/ARW). `orientation` is the EXIF Orientation
/// tag value (1..=8) — baked into the pixels so the saved JPEG is
/// canonically oriented. Writes JPEG bytes to `destination` regardless of
/// its extension.
pub fn generate_image_thumbnail_ffmpeg(
path: &Path,
destination: &Path,
orientation: i32,
) -> std::io::Result<()> {
let vf = build_image_thumb_filter(orientation, 200);
/// Use ffmpeg to extract a 200px-wide thumbnail from formats the `image` crate
/// can't decode (RAW: NEF/ARW, HEIC/HEIF). Writes JPEG bytes to `destination`
/// regardless of its extension.
pub fn generate_image_thumbnail_ffmpeg(path: &Path, destination: &Path) -> std::io::Result<()> {
let output = Command::new("ffmpeg")
.arg("-y")
.arg("-i")
@@ -130,7 +101,7 @@ pub fn generate_image_thumbnail_ffmpeg(
.arg("-vframes")
.arg("1")
.arg("-vf")
.arg(&vf)
.arg("scale=200:-1")
.arg("-f")
.arg("image2")
.arg("-c:v")